dagster #dagster-support

rex

04/14/2021, 11:59 PM

set the channel topic: 🙋 Ask your question after checking out our documentation! https://docs.dagster.io/

rex

04/15/2021, 12:02 AM

set the channel description: 🙋 Wondering how to accomplish your use case in Dagster? Encountering a problem with Dagster? Ask it here after checking out our documentation! https://docs.dagster.io/

🙌 2

Simon Späti

04/15/2021, 11:07 AM

Regarding logs on console: We could manage to get it to the console with configuring

compute_logs

properly:

Copy code

# compute_logs:
#   module: dagster_aws.s3.compute_log_manager
#   class: S3ComputeLogManager
#   config:
#     bucket:
#       env: AWS_STORAGE_BUCKET_NAME
#     local_dir: "/tmp/cool"
#     prefix: "compute_logs"
#     endpoint_url:
#       env: AWS_S3_ENDPOINT_URL      

compute_logs:
  module: dagster.core.storage.noop_compute_log_manager
  class: NoOpComputeLogManager

Now they appear on the console, but not any longer in dagit (View Raw Step Output). Is there an easy way to have it on both console and dagit? 😅 Or do we need to write our own

ComputeLogManager

? Any hint where to do that is much appreciated. cc: @alex. In the docs we missed exactly that part :).

Marco

04/15/2021, 12:10 PM

Hello, I am currently experiencing an issue related to a partition set now deleted. Still, the error persists, I assume because of the underlying state of the daemon, which is persisted after restart. How to solve? Thanks

David Smit

04/15/2021, 1:09 PM

I have been poking around with Dagster for about a week to replicate a co-worker's spreadsheet analysis, executed on a frequency that sends notifications based on undesired conditions to MS Teams. I was able to get dagster-daemon and dagit running locally, but I noticed that the pipeline runs significantly slower on dagit (35-40 seconds) compared to when I run it from the command prompt (about 5 seconds). If I watch the execution in the timeline view, the actual solids take very little time, but there is a 750-1000 ms delay between solids. Is there a reason for this, or maybe a setting or approach I could take to speed it up?

✅ 1

Yugal Sharma

04/15/2021, 1:37 PM

how to create and run a schedule using ScheduleDefinition but not using the Repository

Yugal Sharma

04/15/2021, 1:37 PM

is it necessary to use a repository for scheduling?

Stéphan Taljaard

04/15/2021, 1:43 PM

Related to my question yesterday, is there a way to specify this per solid/composite solid? That might also help

Eduardo Santizo

04/15/2021, 4:42 PM

Hello! When using the docker-compose dagster example new pipeline runs are put into a container, but is there a way to not create a new container but call an existing container, wait for it to finish and then shut it down?

✅ 1

Arun Kumar

04/15/2021, 7:56 PM

Hi, new to Dagster. I have created a toy implementation for my usecase. I am programmatically creating the pipelines and configuring the solids (at definition) using

configured

API based on the data in our DB. I have two questions 1. Could someone validate if my approach is correct? 2. I am not sure if this can be improved with Dynamic outputs. With Dynamic outputs, instead of looping the DB rows within the pipeline definition I am thinking if I can have a solid that can read the rows and yield them as Dynamic Output so that Dagster creates a downstream solid instance for each row. Also, can Dynamic Outputs be nested?

Copy code

def make_event_source_pipeline(source_name):
    pipeline_name = f"{source_name}_pipeline"

    @pipeline(
        name=pipeline_name,
        mode_defs=[
            ModeDefinition(
                name="default",
                resource_defs={"partition_config": partition_config},
            ),
        ]
    )
    def _event_source_pipeline():
        source = compute_source_events()
        # get all analysis and metrics for the source from DB
        # analyses = get_analyses(source)
        # metrics = get_metrics(analysis_name)
        for analysis_name in analyses:
            for metric_name in analyses[analysis_name]:
                metric = metrics[metric_name]

                primary_event_name = metric[0]["event"]
                primary_aggregate = metric[0]["aggregate"]
                primary_process = configured(join_aggregate_analysis_event,
                                             name=f"join_{analysis_name}_{primary_event_name}")(
                    {"analysis_name": analysis_name, "event_name": primary_event_name, "aggregate": primary_aggregate})

                secondary_event_name = metric[1]["event"]
                secondary_aggregate = metric[1]["aggregate"]
                secondary_process = configured(join_aggregate_analysis_event,
                                               name=f"join_{analysis_name}_{secondary_event_name}")(
                    {"analysis_name": analysis_name, "event_name": secondary_event_name,
                     "aggregate": secondary_aggregate})

                analysis_process = configured(analyze_metric,
                                              name=f"analyze_{analysis_name}_{metric_name}")(
                    {"analysis_name": analysis_name, "metric_name": metric_name})

                analysis_process(primary_process(source), secondary_process(source))

    PartitionSetDefinition(
        name=f"{source_name}_partition_set",
        pipeline_name=pipeline_name,
        partition_fn=partitions.date_partition_range(
            start=datetime.datetime(2021, 1, 1),
            delta_range="days",
            inclusive=True,
            fmt="%Y-%m-%d",
        ),
        run_config_fn_for_partition=run_config_for_date_partition,
    )

    return _event_source_pipeline

Yugal Sharma

04/16/2021, 4:26 AM

how to create a repository using RepositoryDefinition?

paul.q

04/16/2021, 6:49 AM

I'm trying to write a GraphQL query that will be used to provide stats to a dashboard (outside Dagit) about the status of pipeline run. If the pipeline has failed we need to provide the step that failed along with any error message. In simple terms, I can get the pipeline and step statuses as follows:

query PipelineRun {

pipelineRunOrError(runId: "7b509160-1661-4a2a-bdf7-3f0b22e7fae5") { __typename ... on PipelineRun { runId status stepStats { stepKey status } } } } Now, to get the error message, I see that the computeLogs member of PipelineRun requires a stepKey. Do I need a second call to get this? Something like:

query PipelineRun {

pipelineRunOrError(runId: "7b509160-1661-4a2a-bdf7-3f0b22e7fae5") { __typename ... on PipelineRun { computeLogs(stepKey: <step_that_failed>) { stderr { downloadUrl } } } } } Maybe I'm missing something? Doesn't seem possible to get what I want in one call. Thanks Paul

Sri Kadiyala

04/16/2021, 10:48 AM

Hello folks! I am trying to migrate a simple cron job into Dagster. I want to use the configSchema to configure my inputs. the specific config type I am interested is Enum. I was able to configure the solid with Enum field. However, on the dagit UI, I am seeing a validation error saying

Value at path root:solids:dev_datasets:config for enum type dataset_prefix must be a string

. Attached are the screenshots of code and the value we are inputing. I tried changing the input by enclosing it with single/ double quotes. I am not sure what is missing here. Thanks

Alexis

04/16/2021, 10:50 AM

Hi everyone, I have a repository with 2 pipelines using the same solids and I’m getting the “Duplicate solids found in repository with name ‘name’. Solid definition names must be unique within a repository” message. The obvious fix would be to give each solid an alias with a prefix for corresponding pipeline or something like that but I’m wondering if there isn’t a better way to do this ? Maybe at the pipeline level, so all solids within the pipeline get a prefixed name or end up unique ? Thanks for your help !

Alex Despotakis

04/16/2021, 1:49 PM

Hey all, in the former general chat I had this correspondence with @daniel about this sensor crashing issue: https://dagster.slack.com/archives/CCCR6P2UR/p1618321081148200 Unfortunately, upgrading didn't seem to work. The terminal does return a new error, stating that the sensor-daemon is unresponsive, then gives us the standard traceback. Then, after the terminal yielded control back to me, it prints one of the log statements back to me. This is also happening on another instance where we have dagster running, but below you'll find my local traceback:

Copy code

alex@PF2C27VY:~/mvp$ dagster-daemon run
2021-04-15 16:10:27 - dagster-daemon - INFO - instance is configured with the following daemons: ['BackfillDaemon', 'SchedulerDaemon', 'SensorDaemon']
2021-04-15 16:10:28 - SchedulerDaemon - INFO - Not checking for any runs since no schedules have been started.
2021-04-15 16:10:28 - BackfillDaemon - INFO - No backfill jobs requested.
2021-04-15 16:10:29 - SensorDaemon - INFO - Checking for new runs for sensor: Remediation
2021-04-15 16:10:58 - SchedulerDaemon - INFO - Not checking for any runs since no schedules have been started.
2021-04-15 16:10:58 - BackfillDaemon - INFO - No backfill jobs requested.
2021-04-15 16:11:28 - SchedulerDaemon - INFO - Not checking for any runs since no schedules have been started.
2021-04-15 16:11:28 - BackfillDaemon - INFO - No backfill jobs requested.
2021-04-15 16:11:58 - SchedulerDaemon - INFO - Not checking for any runs since no schedules have been started.
2021-04-15 16:11:58 - BackfillDaemon - INFO - No backfill jobs requested.
2021-04-15 16:12:58 - dagster-daemon - ERROR - Thread for SENSOR did not shut down gracefully
Traceback (most recent call last):
  File "/home/alex/miniconda3/envs/eip/bin/dagster-daemon", line 8, in <module>
    sys.exit(main())
  File "/home/alex/miniconda3/envs/eip/lib/python3.8/site-packages/dagster/daemon/cli/__init__.py", line 126, in main
    cli(obj={})  # pylint:disable=E1123
  File "/home/alex/miniconda3/envs/eip/lib/python3.8/site-packages/click/core.py", line 829, in __call__
    return self.main(*args, **kwargs)
  File "/home/alex/miniconda3/envs/eip/lib/python3.8/site-packages/click/core.py", line 782, in main
    rv = self.invoke(ctx)
  File "/home/alex/miniconda3/envs/eip/lib/python3.8/site-packages/click/core.py", line 1259, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/home/alex/miniconda3/envs/eip/lib/python3.8/site-packages/click/core.py", line 1066, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/home/alex/miniconda3/envs/eip/lib/python3.8/site-packages/click/core.py", line 610, in invoke
    return callback(*args, **kwargs)
  File "/home/alex/miniconda3/envs/eip/lib/python3.8/site-packages/dagster/daemon/cli/__init__.py", line 39, in run_command
    controller.check_daemon_loop()
  File "/home/alex/miniconda3/envs/eip/lib/python3.8/site-packages/dagster/daemon/controller.py", line 149, in check_daemon_loop
    self.check_daemon_heartbeats()
  File "/home/alex/miniconda3/envs/eip/lib/python3.8/site-packages/dagster/daemon/controller.py", line 130, in check_daemon_heartbeats
    raise Exception(
Exception: Stopping dagster-daemon process since the following threads are no longer sending heartbeats: ['SENSOR']
(eip) alex@PF2C27VY:~/mvp$ INFO:/home/alex/mvp/client/sensors/remediation.py:results are: []

Steve Pletcher

04/16/2021, 4:13 PM

will dagster pick up type constraints for inputs if you add type hints for them in the solid's function, or do you need to define an

InputDefinition

to use that functionality?

Mark

04/17/2021, 2:30 PM

Hi, i am just getting started with dagster and got a little stuck when trying to figure out what the development process actually will look like. I currently can not find how to get a running dagit instance to reload the pipeline/repository files after they have been changed. Somewhere in the issues i read something about a

--watch

CLI option but could not find that in the documentation. So if you could give me a hint how to develop and debug while having dagit open with always the latest

.py

files loaded automatically that would be great.

Mark

04/17/2021, 2:44 PM

And additionally, my final goal is to be able to develop with dagster running in docker containers. I saw the docker example and got it up and running, but dagit does not show the current

repo.py

content after changing the file. I am not sure if the pipeline container is not updating after file changes or if the dagit container is not updating via grpc or both. Should watching files without having to restart/recreate containers usually work somehow in this scenario?

Ronak Jain

04/19/2021, 5:23 AM

Hi All, When we are setup Dagster with PySpark, getting type of issues :

Copy code

When we are doing setupdagster.core.errors.DagsterExecutionStepExecutionError: Error occurred while executing solid "make_people":
  File "/home//.local/lib/python3.6/site-packages/dagster/core/execution/plan/execute_plan.py", line 190, in _dagster_event_sequence_for_step
    for step_event in check.generator(step_events):
  File "/home//.local/lib/python3.6/site-packages/dagster/core/execution/plan/execute_step.py", line 313, in core_dagster_event_sequence_for_step
    _step_output_error_checked_user_event_sequence(step_context, user_event_sequence)
  File "/home//.local/lib/python3.6/site-packages/dagster/core/execution/plan/execute_step.py", line 71, in _step_output_error_checked_user_event_sequence
    for user_event in user_event_sequence:
  File "/home//.local/lib/python3.6/site-packages/dagster/core/execution/plan/execute_step.py", line 608, in _user_event_sequence_for_step_compute_fn
    gen,
  File "/home//.local/lib/python3.6/site-packages/dagster/utils/__init__.py", line 364, in iterate_with_context
    return
  File "/usr/lib/python3.6/contextlib.py", line 99, in __exit__
    self.gen.throw(type, value, traceback)
  File "/home//.local/lib/python3.6/site-packages/dagster/core/errors.py", line 199, in user_code_error_boundary
    ) from e
The above exception was caused by the following exception:
py4j.protocol.Py4JError: org.apache.spark.api.python.PythonUtils.isEncryptionEnabled does not exist in the JVM
  File "/home//.local/lib/python3.6/site-packages/dagster/core/errors.py", line 187, in user_code_error_boundary
    yield
  File "/home//.local/lib/python3.6/site-packages/dagster/utils/__init__.py", line 362, in iterate_with_context
    next_output = next(iterator)
  File "/home//.local/lib/python3.6/site-packages/dagster/core/execution/plan/compute.py", line 126, in execute_core_compute
    for step_output in _yield_compute_results(compute_context, inputs, compute_fn):
  File "/home//.local/lib/python3.6/site-packages/dagster/core/execution/plan/compute.py", line 92, in _yield_compute_results
    user_event_generator = compute_fn(SolidExecutionContext(compute_context), inputs)
  File "/home//.local/lib/python3.6/site-packages/dagster/core/definitions/decorators/solid.py", line 304, in compute
    result = fn(context, **kwargs)
  File "repo.py", line 30, in make_people
    spark = SparkSession.builder.getOrCreate()
  File "/home//.local/lib/python3.6/site-packages/pyspark/sql/session.py", line 228, in getOrCreate
    sc = SparkContext.getOrCreate(sparkConf)
  File "/home//.local/lib/python3.6/site-packages/pyspark/context.py", line 384, in getOrCreate
    SparkContext(conf=conf or SparkConf())
  File "/home//.local/lib/python3.6/site-packages/pyspark/context.py", line 147, in __init__
    conf, jsc, profiler_cls)
  File "/home//.local/lib/python3.6/site-packages/pyspark/context.py", line 224, in _do_init
    self._encryption_enabled = self._jvm.PythonUtils.isEncryptionEnabled(self._jsc)
  File "/home//.local/lib/python3.6/site-packages/py4j/java_gateway.py", line 1531, in __getattr__
    "{0}.{1} does not exist in the JVM".format(self._fqn, name))

Ronak Jain

04/19/2021, 1:46 PM

@Alex Flanagan (Proof Zero) can you help in to dagster with docker configuration ?

Ronak Jain

04/19/2021, 1:48 PM

I configured like that & getting type of error :

Slackbot

04/19/2021, 1:48 PM

This message was deleted.

Brian Abelson

04/19/2021, 5:10 PM

I've been having a pipeline sporadically fail due to

DagsterApiServer] Pipeline execution process for 5899d526-300a-40c6-839d-b73bedbb3885 unexpectedly exited.

I suspect it might be a OOM error or something like that because its part of a large table copy (i run the same pipeline for other tables and those seem to be fine), but I have no way to debug this currently since the error message is opaque. Any advice here?

Vlad Dumitrascu

04/19/2021, 10:41 PM

Hello Dagster ninjas! Thank you so much for your recent help. Great module you have here. I have another (hopefully) quick question: I'm using dagit to run pipelines with a config designated in the '*playgrounds*' view. Works great! My only question is, where do these config files ( json ) live? They seem to persist between dagit runs. I'd like to put them in my project somewhere( back up in source control), as they are getting quite long and convoluted lately. Would hate to have to type that in again, if lost.

Mark

04/20/2021, 12:00 AM

Hi again, one more thing i am trying to figure out: Does someone have a hint how to debug with the VSCode debugger when running a pipeline via dagit? I am new to python and the best option i came up with so far is to use debugpy in a solid, but that only works if i start listening in the solid itself. Is there a better way to initialize debugpy globally and be able to set breakpoints in multiple solids to attach via VSCode debugger and then debug a whole pipeline?

Rubén Lopez Lozoya

04/20/2021, 10:19 AM

Hey guys, every time we re-deploy our dagster user deployments code we have to manually reload the workspace from the Dagit UI or the new changes wont be caught... Is there a workaround for this? It's risky and kinda tedious to depend on someone manually refreshing the workspace after every deployment

mrdavidlaing

04/20/2021, 11:03 AM

I have a PandasColumn.datetime_column that I'd like to constrain to only contain dates in UTC. Is this possible? Currently, if I define the column like this:

Copy code

PandasColumn.datetime_column("at_date", non_nullable=True, is_required=True)

and pass it a dataframe with a the column of type

datetime64[ns, UTC]

(from

<http://pandas.to|pandas.to>_datetime(my_dates, utc=True)

) I get the following error:

Copy code

Warning! Type check failed. Violated "ColumnDTypeInSetConstraint" for column "at_date" - Column dtype must be in the following set {'datetime64[ns]'}.. DTypes received: datetime64[ns, UTC]

Which is pretty much the exact opposite of the constraint I'm trying to place on the column 🙂

Ming Fang

04/20/2021, 6:17 PM

I’m following the “Create New Project” steps here https://docs.dagster.io/getting-started/create-new-project and step 3 doesn’t work. Running

dagster-daemon run

resulted in this error

Copy code

dagster-daemon run
Traceback (most recent call last):
  File "/usr/local/bin/dagster-daemon", line 8, in <module>
    sys.exit(main())
  File "/usr/local/lib/python3.9/site-packages/dagster/daemon/cli/__init__.py", line 126, in main
    cli(obj={})  # pylint:disable=E1123
  File "/usr/local/lib/python3.9/site-packages/click/core.py", line 829, in __call__
    return self.main(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/click/core.py", line 782, in main
    rv = self.invoke(ctx)
  File "/usr/local/lib/python3.9/site-packages/click/core.py", line 1259, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/usr/local/lib/python3.9/site-packages/click/core.py", line 1066, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/usr/local/lib/python3.9/site-packages/click/core.py", line 610, in invoke
    return callback(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/dagster/daemon/cli/__init__.py", line 32, in run_command
    raise Exception(
Exception: dagster-daemon can't run using an in-memory instance. Make sure the DAGSTER_HOME environment variable has been set correctly and that you have created a dagster.yaml file there.

Christian Lam

04/20/2021, 6:31 PM

I am running the app on port 80 as a non-root user. I have a daemon and Dagit container running as part of the Fargate service. The images run fine locally on port 80. Is it possible I need to grant some extra permissions to the user running dagit?

Christian Lam

04/20/2021, 6:31 PM

Copy code

| 1618942759883 | Serving on <http://0.0.0.0:80> in process 27                                                                                                                                                                                             |
| 1618942759886 | Traceback (most recent call last):                                                                                                                                                                                                     |
| 1618942759886 |   File "/usr/local/bin/dagit", line 8, in <module>                                                                                                                                                                                     |
| 1618942759886 |     sys.exit(main())                                                                                                                                                                                                                   |
| 1618942759886 |   File "/usr/local/lib/python3.7/site-packages/dagit/cli.py", line 205, in main                                                                                                                                                        |
| 1618942759886 |     cli(auto_envvar_prefix="DAGIT")  # pylint:disable=E1120                                                                                                                                                                            |
| 1618942759886 |   File "/usr/local/lib/python3.7/site-packages/click/core.py", line 829, in __call__                                                                                                                                                   |
| 1618942759886 |     return self.main(*args, **kwargs)                                                                                                                                                                                                  |
| 1618942759886 |   File "/usr/local/lib/python3.7/site-packages/click/core.py", line 782, in main                                                                                                                                                       |
| 1618942759886 |     rv = self.invoke(ctx)                                                                                                                                                                                                              |
| 1618942759886 |   File "/usr/local/lib/python3.7/site-packages/click/core.py", line 1066, in invoke                                                                                                                                                    |
| 1618942759886 |     return ctx.invoke(self.callback, **ctx.params)                                                                                                                                                                                     |
| 1618942759886 |   File "/usr/local/lib/python3.7/site-packages/click/core.py", line 610, in invoke                                                                                                                                                     |
| 1618942759886 |     return callback(*args, **kwargs)                                                                                                                                                                                                   |
| 1618942759886 |   File "/usr/local/lib/python3.7/site-packages/dagit/cli.py", line 109, in ui                                                                                                                                                          |
| 1618942759886 |     **kwargs,                                                                                                                                                                                                                          |
| 1618942759886 |   File "/usr/local/lib/python3.7/site-packages/dagit/cli.py", line 128, in host_dagit_ui                                                                                                                                               |
| 1618942759886 |     host_dagit_ui_with_workspace(instance, workspace, host, port, path_prefix, port_lookup)                                                                                                                                            |
| 1618942759886 |   File "/usr/local/lib/python3.7/site-packages/dagit/cli.py", line 137, in host_dagit_ui_with_workspace                                                                                                                                |
| 1618942759886 |     start_server(instance, host, port, path_prefix, app, port_lookup)                                                                                                                                                                  |
| 1618942759886 |   File "/usr/local/lib/python3.7/site-packages/dagit/cli.py", line 197, in start_server                                                                                                                                                |
| 1618942759886 |     raise os_error                                                                                                                                                                                                                     |
| 1618942759886 |   File "/usr/local/lib/python3.7/site-packages/dagit/cli.py", line 166, in start_server                                                                                                                                                |
| 1618942759886 |     server.serve_forever()                                                                                                                                                                                                             |
| 1618942759886 |   File "/usr/local/lib/python3.7/site-packages/gevent/baseserver.py", line 398, in serve_forever                                                                                                                                       |
| 1618942759886 |     self.start()                                                                                                                                                                                                                       |
| 1618942759886 |   File "/usr/local/lib/python3.7/site-packages/gevent/baseserver.py", line 336, in start                                                                                                                                               |
| 1618942759886 |     self.init_socket()                                                                                                                                                                                                                 |
| 1618942759886 |   File "/usr/local/lib/python3.7/site-packages/gevent/pywsgi.py", line 1545, in init_socket                                                                                                                                            |
| 1618942759886 |     StreamServer.init_socket(self)                                                                                                                                                                                                     |
| 1618942759886 |   File "/usr/local/lib/python3.7/site-packages/gevent/server.py", line 180, in init_socket                                                                                                                                             |
| 1618942759886 |     self.socket = self.get_listener(self.address, self.backlog, self.family)                                                                                                                                                           |
| 1618942759886 |   File "/usr/local/lib/python3.7/site-packages/gevent/server.py", line 192, in get_listener                                                                                                                                            |
| 1618942759886 |     return _tcp_listener(address, backlog=backlog, reuse_addr=cls.reuse_addr, family=family)                                                                                                                                           |
| 1618942759886 |   File "/usr/local/lib/python3.7/site-packages/gevent/server.py", line 288, in _tcp_listener                                                                                                                                           |
| 1618942759886 |     sock.bind(address)                                                                                                                                                                                                                 |
| 1618942759886 |   File "/usr/local/lib/python3.7/site-packages/gevent/_socketcommon.py", line 563, in bind                                                                                                                                             |
| 1618942759886 |     return self._sock.bind(address)                                                                                                                                                                                                    |
| 1618942759886 | PermissionError: [Errno 13] Permission denied: ('0.0.0.0', 80)                                                                                                                                                                         |
----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------