Basic conceptual problem: (apologies if this is covered somewhere already, but I couldn’t find anything). I have a large-enough pipeline that I have a dict of solid outputs that I want to collect as input to a final, fan-in node in my graph. That is, I have a pipeline definition that looks as follows (pseudoish code):

@solid()
def fan_in(context, results_dict):
   ...

@pipeline()
def ppln():
   
   results = {solid.name: solid() for solid in solids_list}
   final = fan_in(results)

Is this kind of thing possible? Or is this an antipattern? I get a error that indicated to me that Dagster can’t parse this as a DAG appropriately.

dagster.core.errors.DagsterInvalidDefinitionError: In @pipeline process_tables, received invalid type <class 'dict'> for input "table_dict" (at position 0) in solid invocation "write_all_tables". Must pass the output from previous solid invocations or inputs to the composition function as inputs when invoking solids during composition.

It seems like Dagster has a carve-out for the type

List[solid_output_type]

but not for example

Dict[str,solid_output_type]

Hi everyone, not sure if this question has been answered (I can't find anything regarding this). I have followed the example of deploy using Docker (https://docs.dagster.io/deployment/guides/docker#example). Need to make some changes but it works nicely now, able to execute manually and run on schedule. However, every run job leaves behind a container that I need to prune. Is there any settings to remove them after finished running? or maybe a way to pass argument to Docker run so I can pass --remove at runtime ? Thanks in advance!

Quite frequently, I am unable to retry solid errors and I receive the error below. It seems to happen with solids that are downstream of dynamic outputs, but that may not always be the case. At a loss as to where to start troubleshooting and am hoping someone can point me in the right direction. Right now I am running this on my local machine using the multiprocess executor. Full codebase is here: https://github.com/xmarcosx/dagster-etl

grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with:
status = StatusCode.RESOURCE_EXHAUSTED
details = "Received message larger than max (10985123 vs. 10485760)"
debug_error_string = "{"created":"@1634491696.811821461","description":"Received message larger than max (10985123 vs. 10485760)","file":"src/core/ext/filters/message_size/message_size_filter.cc","file_line":206,"grpc_status":8}"

Hi there! Is it possible to express a dependency between two parallel solids and another solid? Two parallel solids have no output and 3rd solid should wait them to finish. I have came up with this, but it looks kinda ugly:

from dagster import solid, pipeline, InputDefinition, Nothing


@solid
def solid1():
    pass


@solid
def solid2():
    pass


@solid
def solid3(s1=[InputDefinition("start", Nothing)], s2=[InputDefinition("start", Nothing)]):
    return 1


@pipeline
def pl():
    solid3(solid1(), solid2())

Hi! I am trying to deploy dagster locally. I have this error when i run

dagit

Traceback (most recent call last):
  File "/Users/jon/python_env/dagster/bin/dagit", line 5, in <module>
    from dagit.cli import main
  File "/Users/jon/python_env/dagster/lib/python3.8/site-packages/dagit/cli.py", line 22, in <module>
    from .app import create_app_from_workspace_process_context
  File "/Users/jon/python_env/dagster/lib/python3.8/site-packages/dagit/app.py", line 23, in <module>
    from .subscription_server import DagsterSubscriptionServer
  File "/Users/jon/python_env/dagster/lib/python3.8/site-packages/dagit/subscription_server.py", line 4, in <module>
    from graphql_ws.gevent import GeventSubscriptionServer, SubscriptionObserver
ImportError: cannot import name 'SubscriptionObserver' from 'graphql_ws.gevent' (/Users/jon/python_env/dagster/lib/python3.8/site-packages/graphql_ws/gevent.py)

Does anyone know why?

I’m trying to setup error handling in dagster I’m new to dagster, and not a Python expert In my first solid I calling an API. What I want is, if the response code is anything other than 200, I want to log response code and stop the entire pipeline. There is no need to run the rest is I don’t get a 200 How do I do that?

Is it mandatory to have a pipeline as an argument for the sensor decorator and yield a run request or can we run any type of operation?

I’m using Dagster to get data from an API and to put it in an azure blob One thing I need it to track

from

parameters between each run. Are there any Dagster feature that can keep this information, or are there a best practice - or does anyone have an advice :)

Hey all, I currently have two solids, one of which triggers a job in DBT, and the other which waits for a completed status. Recently I've had issues where the job will fail in DBT for a random reason and since I have retries set up it will retry the waiting solid. If I make these into a composite solid and have a retry policy will it retry both steps if one of them fails in the composite?

Hi all - is there a known performance issue with large size DynamicOutput mapping? I have a web scrapping pipeline that generates a list of urls (which can be several thousands), fans these into a job to scrape each of this pages and at the end collects the result into one large output. The problem is that while it works perfectly when ran on a small test sample of 10 or so urls, when scaled up to full size, the scheduler just lags to a point where there's several minutes between when one dynamic job finishes and the next starts. The jobs themselves only take about 10ms, but there's a huge pause between them. This appears to be the case regardless of whether I try to use multiprocessor execution or single threaded.

Hi everyone, I'm having an issue loading a repository. After executing the

dagit

command, I'm met with an

Error loading repository location repo_name:dagster.serdes.ipc.DagsterIPCProtocolError: Timeout: read stream has not received any data in 60 seconds

. Does anyone have experience with this error? What could be the issue?

Hey! I'm curious how the sensors' code is executed by Dagster daemon? I have a daemon running as AWS ECS task and I have a sensor defined which calls some AWS services via boto3 SDK and sensor execution always fails with the following error:

botocore.exceptions.NoCredentialsError: Unable to locate credentials

even though I have a proper IAM task role assigned to the ECS task. I also tried providing AWS credentials via environment variables, but had the same result. FWIW besides the problem with sensors the daemon works fine and I'm able to call AWS services from inside the solids without any problems

I am using dagster_pandas and I want to parametrize dagster_pandas_dataframe_types to validate a DataFrame depending on the file I am reading in. An example: if I have a PandasColumn.categorical_column, the expected values in this column will change depending on the file that is being read in, which is inferred by the file name. How should I approach this? Thanks!

Hey team, our Dagster k8s deployment fails from time to time and it shows this error, we are on Dagster 0.12.13, any ideas?

Hey! How should I specify multiple python environments in

workplace.yml

for deploy_docker example? When I try adding

load_from:
  # - grpc_server:
  #     host: docker_example_pipelines
  #     port: 4000
  #     location_name: "example_pipelines"
  - python_file: 
      relative_path: repo.py
      executable_path:  /opt/conda/envs/test/bin/python

Run queue gives error

Caught an error for run c59de7e6-0b8e-48e4-a45a-86e33ccc0881 while removing it from the queue. Marking the run as failed and dropping it from the queue: Exception: No docker image specified by the instance config or repository

Hi all! I have a pattern where I trigger a pipeline async with GraphQL client. (How) can I get access to the return value of the pipeline once it has run to completion?

Does Dagster have plugin support? Hi all! I'm working on setting up some new pipelines and figuring out what tools to use. One pattern we have now is that the software team maintains the core ETL, and the data science contributes a few final stages to the pipeline to compute new metrics. The DS team contributions tend to change more rapidly, may not be as robust, and the DS team may want to re-run stages with new algorithms more frequently than the base pipeline run. So it would be nice to have separate versioning/releasing of pipeline stages from the different teams. Is there a way with Dagster to pull some portions of the pipeline from a separate repo or anything like that? (We could also manage this outside of the orchestration framework, like have CD combine portions of the pipeline from different sources, or have two dependent pipelines with one triggering the next.) Thanks in advance for advice!

Hi!! I would like to make the execution of my pipeline vary in real time depending on the environment. For instance, 4 solids: solid1, solid2, solid3, solid4 pipeline: - If x> 5: solid4 (solid3(solid2(solid1()))) - Else: solid4(solid2(solid1())) How could this be done? Greetings!

hey! i am using dynamic mapping in my pipeline. when i fan out, i want a series of solids to be executed for each branch. i was going to use a composite solid, but it seems that if i do that then i cant execute the solids that make up the composite separately (at least thats what it seems like from dagit), which is something i want to do. is that correct? is there a way around this? thanks!

Hello, I'm doing some scalability testing, running Dagster on K8s connected to a large managed Postgresql DB on Azure. I am trying to run 100 pipelines simultaneously and have provisioned plenty of compute with high resource and high limit to the cluster. Looking at the timestamps in the event log for the solid I noticed that there is quite a _*large delay* when running many pods (6-7 seconds between LOGS_CAPTURED and STEP_START)_. As I'm running the default in_process executor I don't expect io to be the issue. The execution of solids in general also seems very slow compared when running a single pipeline, but I can't seem to understand what the bottleneck might be. Can anyone think of what could be the cause? Cheers, Gabe PS: I'm using Lens to monitor the cluster and none of the pods seem to be reaching their limits.

Hello all. I'm building a sensor to watch for files which are lists of other files (manifests). I'm wanting to generate run requests for valid manifests, and yield a skipreason for invalid manifests. The issue I'm encountering is that I am only able to yield a single SkipReason. I think this makes sense, but is there a standard or desired approach to yielding multiple RunRequest and multiple failure reasons? Am I limited to just logging an error here?

Hello all, i am trying to run databricks job using dagster. i take the code from the documentation

from dagster import ModeDefinition, pipeline, graph
from dagster_databricks import databricks_client, create_databricks_job_op

sparkpi = create_databricks_job_op().configured(
    {
        "job": {
            "name": "Python job",
            "new_cluster": {
                "spark_version": "7.3.x-scala2.12",
                "node_type_id": "i3.xlarge",
                "num_workers": 2,
            },
            "notebook_task": {"notebook_path": "/Users/mohit.asingh@gmail.com/E1"},
        }
    },
    name="sparkpi"
)

@pipeline(
    mode_defs=[
        ModeDefinition(
            resource_defs={
                "databricks_client": databricks_client.configured(
                    {"host": "***<some_value>***.<http://azuredatabricks.net|azuredatabricks.net>",
                     "token": "***<token>***"}
                )
            }
        )
    ]
)
def my_pipeline():
    sparkpi()

while running i am getting this error can anyone tell did i miss any config?

2021-10-19 23:46:17 - dagster - DEBUG - my_pipeline - 18c90b7b-7d41-4d43-868d-bc7ea3f07915 - 3788 - PIPELINE_START - Started execution of pipeline "my_pipeline".
2021-10-19 23:46:17 - dagster - DEBUG - my_pipeline - 18c90b7b-7d41-4d43-868d-bc7ea3f07915 - 3788 - ENGINE_EVENT - Executing steps in process (pid: 3788)
2021-10-19 23:46:17 - dagster - DEBUG - my_pipeline - 18c90b7b-7d41-4d43-868d-bc7ea3f07915 - 3788 - sparkpi - ENGINE_EVENT - Starting initialization of resources [databricks_client, io_manager].
2021-10-19 23:46:17 - dagster - DEBUG - my_pipeline - 18c90b7b-7d41-4d43-868d-bc7ea3f07915 - 3788 - sparkpi - ENGINE_EVENT - Finished initialization of resources [databricks_client, io_manager].
2021-10-19 23:46:17 - dagster - DEBUG - my_pipeline - 18c90b7b-7d41-4d43-868d-bc7ea3f07915 - 3788 - sparkpi - LOGS_CAPTURED - Started capturing logs for solid: sparkpi.
2021-10-19 23:46:17 - dagster - DEBUG - my_pipeline - 18c90b7b-7d41-4d43-868d-bc7ea3f07915 - 3788 - sparkpi - STEP_START - Started execution of step "sparkpi".
2021-10-19 23:46:17 - dagster - ERROR - my_pipeline - 18c90b7b-7d41-4d43-868d-bc7ea3f07915 - 3788 - sparkpi - STEP_FAILURE - Execution of step "sparkpi" failed.

dagster.core.errors.DagsterExecutionStepExecutionError: Error occurred while executing solid "sparkpi"::

TypeError: submit_run() got an unexpected keyword argument 'name'

Stack Trace:
  File "/Users/Mohit.Asingh/PycharmProjects/Dagster/venv/lib/python3.7/site-packages/dagster/core/execution/plan/utils.py", line 44, in solid_execution_error_boundary
    yield
  File "/Users/Mohit.Asingh/PycharmProjects/Dagster/venv/lib/python3.7/site-packages/dagster/utils/__init__.py", line 383, in iterate_with_context
    next_output = next(iterator)
  File "/Users/Mohit.Asingh/PycharmProjects/Dagster/venv/lib/python3.7/site-packages/dagster/core/execution/plan/compute_generator.py", line 65, in _coerce_solid_compute_fn_to_iterator
    result = fn(context, **kwargs) if context_arg_provided else fn(**kwargs)
  File "/Users/Mohit.Asingh/PycharmProjects/Dagster/venv/lib/python3.7/site-packages/dagster_databricks/solids.py", line 173, in databricks_fn
    run_id = databricks_client.submit_run(**job_config)
  File "/Users/Mohit.Asingh/PycharmProjects/Dagster/venv/lib/python3.7/site-packages/dagster_databricks/databricks.py", line 33, in submit_run
    return self.client.jobs.submit_run(*args, **kwargs)["run_id"]  # pylint: disable=no-member

Hello everyone, I'm getting a very strange exception when I'm testing a solid using a resource. It isn't breaking my test pipeline, but I'd like to understand if I have something wrong in my setup, or possibly environment. Traceback in reply.

Hello everyone. I want to run pipeline locally with parallel execution:

from dagster import solid, pipeline, InputDefinition, Nothing, execute_pipeline, ModeDefinition, fs_io_manager, reconstructable, DagsterInstance


@solid
def solid1():
    pass


@solid
def solid2():
    pass


@solid
def solid3(input_defs=[InputDefinition("start", Nothing)]):
    pass


@pipeline(mode_defs=[ModeDefinition(resource_defs={"io_manager": fs_io_manager})])
def test_pipeline():
    solid3([solid1(), solid2()])


if __name__ == "__main__":
    result = execute_pipeline(
        reconstructable(test_pipeline),
        run_config={"execution": {"multiprocess": {}}},
        instance=DagsterInstance.local_temp(),
    )

The problem is that the pipeline is executed from system interpreter, ignoring virtual environment from which the script is executed. How to run this pipeline using virtual environment?

Hi team, I’m using dagster-dbt, How can I pass environment variables into

dbt_cli_resource

so that they can be referenced in dbt profiles? like this: https://github.com/dagster-io/dagster/blob/master/examples/dbt_example/dbt_example_project/profiles.yml

Hi, How can I define more than one repository in values.yaml files' dagerapigrpcargs ?

userDeployments:
  enabled: true
  deployments:
    - name: "my-user-code-1"
      image:
        repository: "<------>"
        tag: latest
        pullPolicy: Always
      dagsterApiGrpcArgs:  
        - "-f"
        - "/my_repo/repo.py"
      port: 3030

Hello again, I'm practicing with modes and resources and I don't quite understand how to apply a resource to a solid. This is my code:

from dagster import solid, pipeline, resource, ModeDefinition


class Velocity:
    def car_speed():
        speed = 200
        return speed

@resource
def select_speed(init_context):
    return Velocity()

high_speed_mode = ModeDefinition(
    "speed_mode",
    resource_defs= {

        "speed": select_speed  <-------------------
    },
)

@solid(required_resource_keys={"speed_mode"})
def print_velocity(context):
    pass
    speed = ???????????  <---------------------------------
    <http://context.log.info|context.log.info>(speed)

    
@pipeline(mode_defs=[high_speed_mode])
def my_pipeline():
    print_velocity()

Greetings!

Hi all, I'm trying to create a schedule that runs my pipeline at the start of each hour. I want to be able to run this through dagit. I have a pipelines.py file where I define solids and their dependencies and I'm deploying through dagster through kuberenetes. I'm trying to run a non partition based schedule: https://docs.dagster.io/concepts/partitions-schedules-sensors/schedules#non-partition-based-schedules I'm having some trouble initializing the schedule. I'm not sure how to trigger this schedule. On dagit, I see: no schedulers found for this instance This is what I have written for the code.

@schedule(cron_schedule="0 * * * *", pipeline_name="my_pipeline", execution_timezone="US/Eastern")

def my_schedule():
    return {}

Not sure what else I have to do here, do I need to pass this schedule to my repository for it to be defined? Thanks a lot!

and is there a way to pass

DynamicOutput

-specific metadata to an

OutputContext

? Say I want to yield 3 or 4 JSON objects that should be stored on s3 by an IOManager. How can I tell it which key to use, mime type to set, ACL, etc? I can return a dict containing this stuff as the

DynamicOutput

value

and parse it inside the IO Manager, just wondering if there’s a cleaner way to do it