How to access metadata from context? I have nested dynamic fan outs, e.g.,

A/B

and

1/2/3

, and I want to access downstream from

3

a metadata property defined in

DynamicOutput

of

A

?

Question on project best practices (with regards to analytics/data science pipelines i.e. reading in a csv, dropping columns, renaming columns, filtering, modelling, etc.): what is the recommended project structure? I searched a bit on the dagster documentation and found 'Create a new project' but that seems more geared toward generating a general project skeleton. We've got a bunch of ipynb files that perform each step in the pipeline - does it make more sense to: wire them together using dagstermill? Convert them into python files and then wire them together with dagster? It's helpful to see cell outputs when the process runs, but it seems like you lose a bit of flexibility. Additionally when we are quickly validating data, conducting EDA, etc. it's nice that I can see plots/visualizations right in an ipynb. Before I embarked on a full blown conversion, just thought I'd check with the community of experts! Thank you in advanced!

Looks like I'm spending 3.15 seconds on starting a subprocess and creating a bigquery client. Is this typical, and coming from dagster, or could there be something about my dag that's causing this behavior?

I've sliced this job really fine, so the overhead adds up. EDIT: It's the

DefaultRunLauncher

. If I figure something out that speeds up local job tests I'll share.

From last night, we're seeing only 3 of our scheduled jobs running every hour, instead of all of them. The other scheduled tasks are marked as "Skipped" in dagit. What can we do to figure out why only 3 jobs run, and the others are being marked as skipped?

Hi team, our Dagster jobs are occasionally failing with the following error from multi-process executors with the following error. However it succeeds when I re-execute the job. Our op just runs a snowflake query using

dagster-snowflake

resource. Any thoughts on why this could happen?

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 analyses_exposures_loader (b15f4f49-05ae-45ad-ab00-60bf07c9797f) in state PipelineRunStatus.FAILURE, expected NOT_STARTED or STARTING

Stack Trace:
  File "/usr/local/lib/python3.7/site-packages/dagster/grpc/impl.py", line 86, in core_execute_run
    yield from execute_run_iterator(recon_pipeline, pipeline_run, instance)
,  File "/usr/local/lib/python3.7/site-packages/dagster/core/execution/api.py", line 77, in execute_run_iterator
    pipeline_run.pipeline_name, pipeline_run.run_id, pipeline_run.status
,  File "/usr/local/lib/python3.7/site-packages/dagster/check/__init__.py", line 170, in invariant
    raise CheckError(f"Invariant failed. Description: {desc}")

Hi all again: I did some searching and couldn't find any easy reference in the documentation on how to handle passing multiple outputs to multiple inputs in dagstermill. Essentially, how would I map step one to step 2 (see the attached image)? I've named the inputs and outputs, but when I chain them as such: PREPROCESS_RAW_TBLS000(ETL010()) I get the following error message: In @job pipeline, received a tuple of multiple outputs for input "test1" (at position 0) in op invocation 010ETL. Must pass individual output, available from tuple: ('test1', 'test2'). Apologies if this is a bit of a silly question, I wasn't able to find it after some google/documentation searches!

Hi! I'm trying to set-up k8s cluster with helm. I want to build a separate user code image as described here: https://docs.dagster.io/deployment/guides/kubernetes/deploying-with-helm#build-docker-image-for-user-code But I don't get the python versioning schema here. As I understand dagster-celery-k8s image uses python 3.7.8 which is a bit outdated. Can I use python 3.9 in my user image?

Hi! I’m trying to install all dependencies needed for a user code deployment in k8s with helm. However I am getting an error related to incompatible dependencies, it looks like dagit and celery (needed for dagster-celery) require incompatible versions of click. I include an excerpt of the constraints as a response to this message. Any suggestions?

Hi! How can i use resourse from sensor?

Hi 🙂 How can i pass a config yaml file and not directly from the code when Reusing an op definition with configured ? https://docs.dagster.io/concepts/configuration/configured#reusing-an-op-definition-with-configured From in the doc

sample_dataset = configured(get_dataset, name="sample_dataset")({"is_sample": True})

To something like that

sample_dataset = configured(get_dataset, name="sample_dataset")(context["is_sample"])

Is there a preferred pattern for defining sub-partitions? I have a job for which I'd like to define a set of static partitions, some of which might have time-based sub-partitions as well. My current thinking is to dynamically create a separate job for each of the static partitions, some of which will have partitioned config and some which will have regular config. Before I did this though I wanted to see if there was an accepted best practice yet.

I recently discovered Dagster and loved the way it does things. I am trying to build the ETL pipeline for a new system we are building using Dagster instead of Airflow. Though it is better in almost every ways, I am stuck with an important requirement and am not able to find a way out. There are two  problem statements :- 1. We have a Kafka queue which is populated at different rates based on when data comes in. 2. Sometimes it may be over a 1000 msgs per second and at other times no messages for long periods. 3. So the Job reading the data from the Kafka queue has to keep checking and run regularly to process the queue. 4. When no data is coming, it would check every 1 minute and stop. The job would end in a few seconds. 5. But if data is continuously coming, it would have to read data and process data. This would take 2-3 minutes. 6. Once the job is completed, it would be run again till the entire queue is again empty. I think I need a  Schedule/Sensor which runs say every 30 seconds. If another instance of the same job is already running, it skips this run. Else it runs the Job. I think even scripts which are receiving data from streaming sources would have a similar requirement. Is there any other pattern for such long running jobs. Would be grateful for any help.

Hello Team, I'm a newbie to Dagster. I'm trying to convert an existing S3 event notification into Dagster code. So I have a S3 sensor which watch for the new files in a specific prefix and yield a run request like below

yield RunRequest(run_key=s3_key, run_config = {"ops": {"op_trigger_lambda": {"config": {"s3_key": s3_key}}}})

We have planned to deploy this in AWS ECS (as mentioned in the "deploy_ecs" example, which uses EcsRunLauncher). My Question - Does each Run Request will be invoked in parallel or in serial, because in docs it mentioned as queue? If it is not parallel, then how can this be converted to parallel ? Could you please help me here ?

Hey, I got back to trying CeleryK8sRunLauncher for per-op limits. I have configuration in my dagster helm values.yaml:

runLauncher:
  type: CeleryK8sRunLauncher
  config:
    CeleryK8sRunLauncher:
      workerQueues:
        - name: "dagster"
          replicaCount: 5
        - name: "database"
          replicaCount: 50

But still only one deployment is created celery workkers. What am I doing incorrectly?

Hi team, When we executed pipelines k8s-jobs got stuck and there is no pod created, how to debug, can anyone help me on this issue?

Hey 👋 Am I correct in understanding that the Configured API can only be used on Ops only for fields defined in

config_schema

(and not

Ins

)? Since we can configure Op inputs in run configuration, I was expecting to also be able to do that with the Configured API, but I’m getting errors that make me think it isn’t the case.

Hi! Forgive me if im missing something obvious, but when i’ve looked at the example code for running Great Expectations or DBT from dagster, the way that that resources are configured is to always pass in a filepath for GE or DBT. can i instead pass in an image? i have docker images already in google cloud registry. i have it working locally with dagit using the filepath but am now trying to productionize

A couple of questions: 1)First how can I let a pipeline finish successful prematurely? We have a solid that checks the state and if it's fine it does not have to run. At the moment this pipeline raises "Failures" but that makes the run statistics very hard to read, which pipeline failed truly and which had a negative check? (this check needs to be done on the pipeline not externally) 2) We saw that our process takes 97 MB of RAM, when we use the multiprocessor it starts processes (more than configured) and each of them takes around 100 MB the containers grows big really fast, which makes it hard to let Dagster jobs scale. Using the in_process_executor has it's own flaws, no concurency and the failing solids that get retried don't show up so nicely. Is there a way to reduce the RAM usage?

Hey all! brand new to Dagster and checking it out as a potential option for replacing our current orchestration tools. I have a question which I can probably guess the answer too however just want to make sure. Do async operations in native python cause any problems in Dagster pipelines?

Any updates on order based ergonomics? It seems we can no longer use the workaround to test graphs written this way: https://github.com/dagster-io/dagster/issues/5990

Hi guys 🙂, I need to pass to the callback of the configured function

configured()(callback)

a configuration that i can write in the dagit interface as follow :

ops:
   my_op:
      config:
          callback: 0

So my question is : what i need to write in the callback of configured, in order to have access to 

context.op_config['callback']

  ?

Here is where i found the configured API : https://docs.dagster.io/concepts/configuration/configured#reusing-an-op-definition-with-configured but the only example is an example where they pass the config in the callback in hard from the code

Maybe Op Factory was what i was searching for 🙂 https://docs.dagster.io/concepts/ops-jobs-graphs/ops#op-factory

Hi! I am new to Dagster and am exploring ways of running it within our K8 cluster. One thing that I would love to do is to be able to use different images per op in a Dagster job. There seems to have been an PR here and some discussion about the feature here but I'm having a hard time figuring out how to implement it...Is there an example of how that above PR should be used when defining an op and using the

K8JobExecutor

?

Is there a way to create an AssetMaterialization based on job success (run_status_sensor)? I've tried below but it doesn't seem to publish anything. The use case I have is a I want to use multi-asset sensor but have it based on when the upstream jobs successfully finish (was originally doing the AssetMaterialization in a publish op but since the upstream jobs have DynamicOutput root op, there's no guarantee everything has been published by the time the AssetMaterialization is created).

@run_status_sensor(pipeline_run_status=PipelineRunStatus.SUCCESS)
def ingest_asset_materialization(context: RunStatusSensorContext):

    job_name = context.pipeline_run.pipeline_name

    metadata = {
        "run_id": context.pipeline_run.run_id,
        "pipeline_name": job_name,
    }

    yield AssetMaterialization(
        asset_key=job_name,
        description="Job successfully finished",
        metadata=metadata,
    )

Is anyone using Taint and Toleration with dagster-k8s? We're having an issue getting the tolerations not passing to the job/pod for our dagster run, so it is not getting scheduled to the instances with the "transient" Taint.

Hi there, how can I get dagit to be reachable by other IP addresses? It seems to be only reachable from 127.0.0.1

Hi guys 🙂

I've this error coming from my repository file