Hi Team, I deployed in k8s with helm and separated dagster and user code. I have 2 questions: 1. Since user code container is a long running service, why dagster need to launch a new container with same user code image for each execution? 2. I need to mount a config file into user code container for different environment (qa/uat/prod). How can I mount the same config into the new launched job container?

Hi, We already know that workspace in Dagster refers to collection of multiple repository, which we specify in workspace.yaml. But, can we also have multiple workspaces in Dagster? I have a use case, where we want to divide the workspaces based on domain, so, that one can only see pipeline and repositories of his domain's. To understand it better let's take an example:-

Say we have two domain X and Y. Products of domain X are [x1,x2,x3] and of Y are [y1,y2,y3]. Pipeline name will based on products of domain. Now, in normal condition we will create two repositories with name X and Y, pipelines will be created under respective repositories. If go to the Dagster UI, person from Domain X can see what is in Domain Y and vice versa. That is what we don't want to do. We want that, person from Domain X should only see pipelines related to X and not of Y and vice versa.

Can someone provide me a simple dagster.success_hooks example? I need to know how the ModeDefinition is defined for the same

ModeDefinition("prod", resource_defs={"slack": slack_resource})

The Above mode def is not allowed since resource_defs requires us to send ResourceDefinition type. This is the same line from the example given in the documentation for success_hooks. https://docs.dagster.io/concepts/solids-pipelines/solid-hooks#environment-specific-hooks-using-modes

ModeDefinition("prod", resource_defs={"slack": ResourceDefinition(resource_fn =example_resource)})

The above mode_def with ResourceDefinition also gives a DagsterInvalidDefinitionError Thank you!

Hello! Is there a way to force consequent solid execution, even if they have nothing to return. They are independent and can be run in parallel, but they have to be consequent

Good day! Is there a possibility in dagster to create DynamicOutput inside DynamicOutput? for example: • generate date sub_ranges: ◦ start_date_1 - end_date_1 (DynamicOutput) ▪︎ for that period of time generate sub sub OutputDefinition in multiprocessing (for example) ◦ … ◦ start_date_N - end_date_N (DynamicOutput)

Hi team! @daniel I am concerned about running Dagster on windows. I am still facing this strange error that only occurs on OS windows. The company has at least 5000 employees with windows machines, so It's very difficult to replace all the systems. I am not sure how to find a solution for this error, and I need it. Is there some way to log this type of error with a more detailed log? The company is thinking to use Dagster as standard framework.

Hi team, when I run “dagster sensor cursor --delete” from user-code-deployment, it printed

Cleared cursor state for sensor xxx

. However, my sensors are still getting old cursor data. Did I miss something?

Hi, for conditional branching is there a way to merge the branches back together, even when certain steps don’t execute? (i.e. something like skipping certain steps and having those inputs be optional at the merge node). For example, instead of having to define a pipeline like this which is redundant:

@pipeline
def pipeline():
    one, two = solid_1()
    solid_4(solid_2(one))
    solid_4(solid_3(two))

Can it be changed to something like:

@pipeline
def pipeline():
    one, two = solid_1()
    solid_4([solid_2(one), solid_3(two)])

Assuming only one of

solid_2

or

solid_3

will execute?

is there a way to pass the scheduled date into the dbt_cli_run function (see

event_date

)

dbt_run_models = dbt_cli_run.configured(
    name="dbt_run_models",
    config_or_config_fn={
        "project-dir": PROJECT_DIR,
        "profiles-dir": PROFILES_DIR,
        "models": MODELS,
        "target": TARGET,
        "vars": '{event_date: ' + event_date +'}'
    }
)

Hello everyone! I want to do in dagster something like this:

import dagster as d

@d.solid
def get_data():
    try:
        load_data()
        yield d.Output(True) # Return True Output - everything loaded OK
    except Exception as ex:
        raise d.RetryRequested(max_retries=3)
        if max_retries == 3: # If fails 3 times, than return False output
            yield d.Output(False)

How can i write the last condition in dagster terminology?

Hey team, is there a way to have any sort of solid decorator that always unpacks the received results? I'm upgrading to 0.12 version and I have a lot of pipelines breaking because of that. I see you mentioned in the past that you used namedtuples to return multiple values, but I was wondering if I can do that with a wrapper to avoid having to refactor the whole codebase

is there a way to set a timeout for a job run? We ran into this scenario where the executor lost its connection to the Dagster db (the run storage? I guess) and Dagit left the run as “In Progress” for a day before we noticed and terminated it. It was especially problematic because I had set up a run concurrency limit of 1 so a few hundred runs got piled up behind it

Beginner question - I was testing out setting different Input and Output definitions and noticed that when I use List ( from dagster import List ) in the dagster UI it always shows the input and outputs as the Any type. When I use Int or Float it shows correctly in the UI. Is this expected behaviour or am I missing something really simple?

@solid(
    output_defs=[OutputDefinition(dagster_type=List)]
)

is it true we can now make async def solids?

Hey, I'm starting to use resources in Dagster and was wondering how can I configure the datadog resource "globally" so that all my pipelines have access to it? In the docs the example uses

execute_pipeline

but I am not calling it that way so I don't know where to place the config object with the credentials 😞

hello, I am using docker-compose with default run launcher for development. I noticed that even when mounting volumes from the pipeline container to my code I still need to restart the pipeline container to refresh. Any changes to that? Would love to have the easy refresh of pure local but with single host docker (easier to manage dependencies and faster to deploy a new local dev). https://docs.dagster.io/deployment/guides/docker#mounting-volumes

Hi ! I'm using dagster via docker-compose but I'm having some issues. Even though I set up the property

auto_remove:true

at file

dagster.yaml

I still have multiple residual containers after its execution. Could anyone help me ? I'm using dagster version 0.12.1

Question: are there plans to support large numbers of asset keys? We have about 13,000 unique asset keys stored in dagster’s postgres instance. Dagit takes about 50 seconds to just load the asset view, which makes debugging and monitoring a bit frustrating. Context: we run single-tenant pipelines, and all asset materializations include the tenant id in the asset key -- hence the large numbers of keys. I see there was some initial work to pull out a separate asset_keys table (https://github.com/dagster-io/dagster/issues/3083), but are there plans to optimize that table and interactions with dagit? Right now it appears the entire table is loaded by dagit.

Hi team, just curious if it is possible to switch between

DefaultRunLauncher

and

K8sRunLauncher

for different pipelines? We currently use

K8sRunLauncher

by default. However certain pipelines are very cheap (eg: just triggers an API) and can be easily executed as a process and use

DefaultRunLauncher

Hi , I want to run a pipeline sensor which executes only when a condition inside a solid is met. Also to be noted when i turn on the sensor from dagit and have dagster-daemon running it starts running the pipeline in an infinite loop.

Hi All! As part of one

solid

we need to run a python file using

subprocess.Popen

@solid
def start_prodigy(context):
    process_env = {**os.environ, 'PRODIGY_LOGGING': 'verbose'}
    p1 = subprocess.Popen(["prodigy", "stream_from_s3", "bucket", "customer1/", "-F", "s3_loader.py"], 
                env=process_env, stdout=subprocess.PIPE)

The problem is how do we upload the

s3_loader.py

from our current working dir to where the Dagster will run it?

Is there a Dagster-onic way of achieving something like this? My case is the following: I have a dataframe with image metadata. For each image, I want to download the file from S3 and break it up into tiles (solid A), write a json manifest (solid B) and upload everything back to S3 (solid C). The way we normally use solids that deal with multiple files is passing a list of paths between them, but in this case I specifically want to process one at a time so S3 always has the latest processed files (the whole thing is gonna take days, I don’t want to wait for it to finish to have the files on S3, and I don’t have enough storage on Render to keep everything there until it finishes)

Hi everyone, I have the following use case. Dynamic Repositories: I am working on a project where the number of repository locations is not defined beforehand. They can be added dynamically by the user.  The way this is being achieved currently is, once the repository location code is in place, the workspace.yaml gets updated with the new repository location. Then we reload the workspace(reload all) and our new repos are reflected in the dagit ui. (We are not using user-code-deployments for this) Issue:  https://dagster.slack.com/archives/C01U954MEER/p1631599114445200 https://dagster.slack.com/archives/C01U954MEER/p1631700873486500 Our understanding of error: I have had multiple interactions with the dagster-support regarding this. I got to know that when we use workspace.yaml for repositories, all the repositories are created in a single pod. This might be triggering the grpc error.  Apparently, solution lies in creating separate pods for each repository, which is possible with user-code-deployments. But, before we explore that, I wanted to know, in future will you be providing support for launching each repository in a separate pod in while using workspace.yaml? PROBLEMS WITH USER-CODE-DEPLOYMENTS: Reload functionality behaves differently here, which does not support our use-case. With this option,reload is more like a refresh. https://dagster.slack.com/archives/C01U954MEER/p1627647974003500?thread_ts=1627586675.161900&cid=C01U954MEER

My pipeline is taking over 30s to complete but it gets killed when I run it with

dagit

Error log:

Exception: Timed out waiting for gRPC server to start with arguments:
/bin/python -m dagster.grpc --lazy-load-user-code --socket /var/folders/mj/97j2r6pn6d95fcbtjc5kdvrm0000gn/T/tmpm6if675b --heartbeat --heartbeat-timeout 30 --fixed-server-id 56b10928
Stack Trace:
python3.9/site-packages/dagster/core/host_representation/grpc_server_registry.py", line 177, in _get_grpc_endpoint

STEP_FAILURE - Execution of step failed.
dagster.core.errors.DagsterExecutionInterruptedError
Stack Trace:
python3.9/site-packages/dagster/core/execution/plan/execute_plan.py", line 195, in _dagster_event_sequence_for_step

I am using Dagster version 0.12.11

@asset_sensor(asset_key=AssetKey("my_table"), pipeline_name="my_pipeline")
def my_asset_sensor(context, asset_event):
    yield RunRequest(
        run_key=context.cursor,
        run_config={
            "solids": {
                "read_materialization": {
                    "config": {
                        "asset_key": asset_event.asset_key.path,
                        "pipeline": asset_event.pipeline_name,
                    }
                }
            }
        },
    )

The above example is for asset_sensor in dagster docs. during execution an error occurs saying that 'EventLogEntry' object has no attribute 'asset_key'. What is the type for asset_event here?

Also can we access AssetMaterialization metadata by any means?

Hey, Linear dependencies dont seem to be working with composite solids, I cannot enforce a dependency of a solid on a composite solid, any ideas? https://docs.dagster.io/concepts/solids-pipelines/pipelines#linear-dependencies

Hey all, ever since updating to 0.11.13 I've been getting lots of these errors:

An exception was thrown during execution that is likely a framework error, rather than an error in user code.
dagster.check.CheckError: Invariant failed. Description: Pipeline run cleanup (e44bf39c-e57e-4318-ac8f-1f7c62667a40) in state PipelineRunStatus.STARTED, expected NOT_STARTED or STARTING.

Any idea what would be causing this? I'm running this on a k8s instance that's comprised of spot instances and from what I've seen it looks like that might be a cause if the nodes are going down frequently. But before on 0.11.12 it was failing less. I also switched from celery_k8s_exec to just the k8s_job_exec during this time too.

I am trying to run my working pipeline with a Multiprocess Executor but I have a Pandas Dataframe in my

run_config

. Dagster cannot serialize the Dataframe so I converted it to JSON but then I get this error: `OverflowError: string longer than INT_MAX bytes`I tried to to compress the string using zlib but then I am getting

TypeError: Object of type bytes is not JSON serializable

has anyone encountered this?

We moved to daylight saving time overnight in Sydney, and I don't know if this is linked or not but my Scheduler has been failing since then. I run the daemon as a service but restarting it (or restarting my machine) didn't fix it. It also looks like since then, the

dagster-daemon run

command is always eating 100% of one of my CPUs. Here is the error in journalctl (I'm on dagster 0.12.12):

Oct 03 08:49:03 raspberrypi bash[10714]: 2021-10-03 08:49:03 - dagster-daemon - ERROR - Thread for SCHEDULER did not shut down gracefully
Oct 03 08:49:03 raspberrypi bash[10714]: Traceback (most recent call last):
Oct 03 08:49:03 raspberrypi bash[10714]:   File "/home/pi/.envs/dagster/bin/dagster-daemon", line 8, in <module>
Oct 03 08:49:03 raspberrypi bash[10714]:     sys.exit(main())
Oct 03 08:49:03 raspberrypi bash[10714]:   File "/home/pi/.envs/dagster/lib/python3.7/site-packages/dagster/daemon/cli/__init__.py", line 135, in main
Oct 03 08:49:03 raspberrypi bash[10714]:     cli(obj={})  # pylint:disable=E1123
Oct 03 08:49:03 raspberrypi bash[10714]:   File "/home/pi/.envs/dagster/lib/python3.7/site-packages/click/core.py", line 829, in __call__
Oct 03 08:49:03 raspberrypi bash[10714]:     return self.main(*args, **kwargs)
Oct 03 08:49:03 raspberrypi bash[10714]:   File "/home/pi/.envs/dagster/lib/python3.7/site-packages/click/core.py", line 782, in main
Oct 03 08:49:03 raspberrypi bash[10714]:     rv = self.invoke(ctx)
Oct 03 08:49:03 raspberrypi bash[10714]:   File "/home/pi/.envs/dagster/lib/python3.7/site-packages/click/core.py", line 1259, in invoke
Oct 03 08:49:03 raspberrypi bash[10714]:     return _process_result(sub_ctx.command.invoke(sub_ctx))
Oct 03 08:49:03 raspberrypi bash[10714]:   File "/home/pi/.envs/dagster/lib/python3.7/site-packages/click/core.py", line 1066, in invoke
Oct 03 08:49:03 raspberrypi bash[10714]:     return ctx.invoke(self.callback, **ctx.params)
Oct 03 08:49:03 raspberrypi bash[10714]:   File "/home/pi/.envs/dagster/lib/python3.7/site-packages/click/core.py", line 610, in invoke
Oct 03 08:49:03 raspberrypi bash[10714]:     return callback(*args, **kwargs)
Oct 03 08:49:03 raspberrypi bash[10714]:   File "/home/pi/.envs/dagster/lib/python3.7/site-packages/dagster/daemon/cli/__init__.py", line 48, in run_command
Oct 03 08:49:03 raspberrypi bash[10714]:     controller.check_daemon_loop()
Oct 03 08:49:03 raspberrypi bash[10714]:   File "/home/pi/.envs/dagster/lib/python3.7/site-packages/dagster/daemon/controller.py", line 237, in check_daemon_loop
Oct 03 08:49:03 raspberrypi bash[10714]:     self.check_daemon_heartbeats()
Oct 03 08:49:03 raspberrypi bash[10714]:   File "/home/pi/.envs/dagster/lib/python3.7/site-packages/dagster/daemon/controller.py", line 212, in check_daemon_heartbeats
Oct 03 08:49:03 raspberrypi bash[10714]:     failed_daemons=failed_daemons
Oct 03 08:49:03 raspberrypi bash[10714]: Exception: Stopping dagster-daemon process since the following threads are no longer sending heartbeats: ['SCHEDULER']
Oct 03 08:49:04 raspberrypi systemd[1]: dagster-daemon.service: Main process exited, code=exited, status=1/FAILURE
Oct 03 08:49:04 raspberrypi systemd[1]: dagster-daemon.service: Failed with result 'exit-code'.
Oct 03 08:49:04 raspberrypi systemd[1]: dagster-daemon.service: Service RestartSec=100ms expired, scheduling restart.
Oct 03 08:49:04 raspberrypi systemd[1]: dagster-daemon.service: Scheduled restart job, restart counter is at 6.
Oct 03 08:49:04 raspberrypi systemd[1]: Stopped Daemon for dagster.
Oct 03 08:49:04 raspberrypi systemd[1]: Started Daemon for dagster.

The heath page tells me: "Not running - No recent heartbeat"