set the channel description: Discussions on best practices for using dagster to build an experimentation platform, ml training platform, ml ops platform, etl platform, etc

great channel idea!

Does anyone have thoughts to share on documenting the “transformation rules” that are implements (and tested!) in a Dagster pipeline that are useful to multiple stakeholders? Ideally in a way that DRYs up describing the rules across (a) the requirements specification, (b) the tests and (c) the documentation The BDD community has some ideas around “executable specifications” / “living documentation”; but all the tools / examples are pretty “website workflow” orientated.

Starting to develop my first pipeline locally on a Macbook. For production plan to deploy the dagster services to a cluster to spawn workers as needed. We don’t need all the capabilities of Kubernetes so we’d like to use Hashicorp’s Nomad system. Has anyone done this/any thoughts/advice? I’m thinking this is probably a good place to start?

Hello! I've tried Dagster so far on a (simple) Data Prep / ML pipeline and I've decided to go further with it. Something that I find really interesting is developed in . Is there any repo with the code?

Hello! I am trying to migrate to the new experimental API. I want to essentially execute the same pipeline/graph multiple times based on a list of string values. The string values in my case are 12 manufacturing plant codes. Example: "Plant1", "Plant2", ... , "Plant12". I also need to pass in the plant code to each and every solid/op, so I have a solid/op config. So I want to schedule 12 jobs and have them execute weekly every Monday at 10am. The naïve approach which violates DRY (Don't Repeat Yourself) would be to do the following:

plant1_schedule = ScheduleDefinition(
    job=my_graph.to_job(
        name='plant1_job',
        config={
            "ops": {
                "do_some_task1": {"config": {"plant_code": "Plant1"}},
                "do_some_task2": {"config": {"plant_code": "Plant1"}},
                "do_some_task3": {"config": {"plant_code": "Plant1"}},
            }
        }
    ),
    cron_schedule="0 10 * * 1"
)

plant2_schedule = ScheduleDefinition(
    job=my_graph.to_job(
        name='plant2_job',
        config={
            "ops": {
                "do_some_task1": {"config": {"plant_code": "Plant2"}},
                "do_some_task2": {"config": {"plant_code": "Plant2"}},
                "do_some_task3": {"config": {"plant_code": "Plant2"}},
            }
        }
    ),
    cron_schedule="0 10 * * 1"
)

...

plant12_schedule = ScheduleDefinition(
    job=my_graph.to_job(
        name='plant12_job',
        config={
            "ops": {
                "do_some_task1": {"config": {"plant_code": "Plant12"}},
                "do_some_task2": {"config": {"plant_code": "Plant12"}},
                "do_some_task3": {"config": {"plant_code": "Plant12"}},
            }
        }
    ),
    cron_schedule="0 10 * * 1"
)

Piggybacking off of @Daniel Kim and @sandy. For the below code, is there a way to notify when all 12 runs completed successfully?

@schedule(job=my_graph, cron_schedule="0 10 * * 1")
def plant_schedule(_):
    for i in range(12):
        config={
            "ops": {
                "do_some_task1": {"config": {"plant_code": f"Plant{i}"}},
                "do_some_task2": {"config": {"plant_code": f"Plant{i}"}},
                "do_some_task3": {"config": {"plant_code": f"Plant{i}"}},
            }
        }
        yield RunRequest(run_key=None, run_config=config)

I've looked into asset-based sensors which would work for each launched run and multi-asset sensors which would involve writing out each plant code again (code example below from docs):

@sensor(pipeline_name="my_pipeline")
def multi_asset_sensor(context):
    cursor_dict = json.loads(context.cursor) if context.cursor else {}
    a_cursor = cursor_dict.get("a")
    b_cursor = cursor_dict.get("b")

    a_event_records = context.instance.get_event_records(
        EventRecordsFilter(
            event_type=DagsterEventType.ASSET_MATERIALIZATION,
            asset_key=AssetKey("table_a"),
            after_cursor=a_cursor,
        ),
        ascending=False,
        limit=1,
    )
    b_event_records = context.instance.get_event_records(
        EventRecordsFilter(
            event_type=DagsterEventType.ASSET_MATERIALIZATION,
            asset_key=AssetKey("table_a"),
            after_cursor=b_cursor,
        ),
        ascending=False,
        limit=1,
    )

Not sure if more appropriate for this channel or #dagster-support, but for engineers using Dagster with EMR, are you spinning up/terminating clusters in a Dagster solid or using external resources to handle those pieces? My use case: a pipeline that I want to spin up a cluster, run some PySpark jobs on (and ensuring those jobs are successful before moving on to the next solid), maybe loading data into Snowflake, then terminating the cluster. Running the Pyspark job and loading the results into Snowflake would both be solids. Wondering if Dagster is appropriate to use to orchestrate the cluster itself

A very similar question here: what’s the go-to deployment model when dealing with multiple teams? Do you centralize the dagit instance and let teams deploy their pipelines against that central unit? Do you let each teach deploy everything for each project? How do you deal with the repositories being managed by another team?

I'm wondering how folks are thinking about monitoring in a “Software Defined Asset“ world. Has anyone done any thinking / work around what the correct “locations” might be to monitor; and how these might be integrated into common monitoring / alerting tool chains like Prometheus/PagerDuty?

Good morning everyone 👋 I'm new to Dagster and am currently trying to understand whether it is an appropriate choice for our use case. The one question I can't seem to find the answer to is about how to manage jobs across multiple Git repositories (stored in GitLab in our case). We have a number of separate Python applications each running regular batch jobs (currently using scheduled ECS tasks). We're not going to put all this batch job code in a single Git repository as these jobs are part of larger applications (e.g. also providing an API). So, if we are running Dagit and the Dagster daemon in a separate service, what's the best way to pull in the relevant job code from each application to be executed? The best option I can think of would be for each application to publish a Python package for the main Dagster service to import, but I can see that leading to dependency issues. Any suggestions would be greatly appreciated 😄

Is it reasonable to use a persistent volume in kubernetes with SQLite for change data capture? Specifically, with a User Code deployment Context: I have a simple slack-bot pipeline I created that hits an API to retrieve new content, filters results, and posts a message. I’d like to retrieve the new content to see if it’s content that’s actually new or just updated. Naturally, I could create a table and do some select/upsert fun, but this is very small-scale at the moment and it’d be nice if my User Code deployment could just use a PV with SQLite (perhaps with an occasional backup to a bucket). Is there a better way to go about this? (e.g. read more on asset materialization haha)

Hi all 👋 we’re overhauling our data collection & ML pipeline and Dagster came up as an alternative to Argo, which we use as a Kubernetes-native workflow scheduler. Unfortunately we’ve found the patterns Argo offers to be too blunt and confusing for our Data Scientists to daily drive. The programming model in Dagster seems to be a much better fit for us. If we use it, we’ll use it for everything, but on the data collection side of things it isn’t clear is how to structure high volume, highly parallel tasks. Say we’re collecting data from an API, and we want to limit our 100k scheduled queries to 100 in parallel. With Argo we create batches of 1,000 requests, set

paralellism

to 100 and it would only spin up pods as necessary. After collection a quasi-reduce stage would take the outputs from those collection tasks and output them as a dataset/publish to Kafka. My question is, can I get better granularity with Dagster in an idiomatic way? Can I spin up 100K tasks and have the Run Coordinator limit concurrency to 100, or will I need to batch the tasks together? Will I need Celery to handle this load? I’d prefer not to have another task queue to manage!

A second question - we lean on CUElang for all our config, and it would be good to use it in Dagster. We can manage resource connection strings, node names etc in production. Our pipeline for CUE is to compile the config down to a YAML file in CD and deploy that as a K8s secret. We’d do this for each repository. The secret is mounted into the pod using volumes like:

spec:
      volumes:
      - name: config
        secret:
          secretName: my-service-config
      containers:
        volumeMounts:
          - mountPath: "/etc/my-service-config"
            readOnly: true
            name: "config"

From what I can see on the docs we can import that into the container using

K8sRunLauncher

. Is that right? The main question is, can I merge configs? It would be good for all the production string values for resources to come from this precompiled YAML file, but op configuration we might want to handle in the launchpad - this would be great DX for our Data Scientists.

Hello everybody, I just finished playing around with the different tutorials and got a newb question : Is it recommended to run dagster on a single ubuntu vps, we just need to orchestrate 10 ish jobs ( ELT mainly nothing really heavy ) once a day ? And if it s sounds ok do you have link with some tips on how to do such a simple deployment, I have gone through the documentation and havent found a step by step configuration guide ( how to open the right ports, for conda users installing package with conda forge/ where the config yaml files are supposed to be ... ) . I ll be happy to contribute to such a tutorial for the slow learners like me . What do you think ?

How should I materialize partitioned assets when using Software Defined Assets, particularly when working with SourceAssets? I’m reading over the new Software Defined Asset documentation for 0.14.1 and it’s unclear to me how to go about this. e.g. If I have a partitioned, daily job that hits an API and yields a (daily) partitioned AssetMaterialization, how could that be refactored to be an SDA?

Im using DBT and it works great so far. How do you make sure you only refresh one model if you have multiple? Create a dbt project for the each or modify the dagster operator?

Hello, everybody! I'm new to Dagster - could you please help me with an advise - I need to backfill my partitions and I need to prevent skipping a partition if it was failed. How can I do it? At this point Dagster queues tasks for all partitions and it starts a next task from a queue if previous was failed.

This message was deleted.

Hello ! I'm found the blog post by @sandy really great (actually, all are great, but this one is so enlightening for me!). I have many questions, but there's a point I'm really not comfortable with. There's a reference to Maxime Beauchemin's Functional Data Engineering and in this article, about "late arriving facts" it insists on the fact that time partitioning should always be done on event processing time. But most likely, produced assets are partitioned on the data dates. So how do we reconcile the 2 dimensions? There is no longer obvious lineage link between a produced asset (a partition based on data date) and the input (potentially multiple partitions containing upstream input data). And regarding non-mutability of data "units", combined with the need for reproducibility, don't you think that materialization layer must take care of this and be able to serve the possible versions to Dagster? Or else, it's the processing/ingestion partitioning from the first question that is the only enabler for reproducibility?

Curious about Software Defined Assets to represent files that are dropped into S3 from external sources, at what level to consider the 'first' asset and trying to make as many downstream jobs idempotent, as someone new to Dagster --- Looking into some of the past comments about using sensors for Message queues, interested in jobs that would be triggered off Cloud Storage object creation events. Any suggestions to at what point you would materialize an asset to represent the messages from the queue? The concern is since queue consumption isn't idempotent as the messages are removed. Seems like a really small job that would sense a single message on the change notification topic, materialize an asset that represents the File object, and only ack the message once the materialization was successful. So everything downstream would be based off the Object Asset. Does that sound reasonable or how would you look to structure a workflow like this?

Hi everyone ! I would like to have your thoughts about how to implement Dagster and run jobs on several local machines (I'm a week old dagster newbie) My use case : I have an already existing data pipeline written and ran only manually (step by step) that I want to transfer to Dagster (because it's currently a huge pain to operate and adapt to other use cases without a proper orchestrator). As two external paying APIs are used, I'm pretty cautious about task executions and duplicates avoidance. Details : The pipeline is seeded with quite extensive configurations (semantics, urls, numeric tresholds, etc.), it begins with composing first API queries, then runs them in batches, then process them (with a changing IO json schema), then evaluates data and compose the second API targets, etc. until processing a custom report. It's currently built on top of a local Mongo DB (which was better for prototyping) and ran locally. Current goal : As I need to prototype quite a few things to improve and extend the existing pipeline and make it run time consuming tasks (selenium automation tasks), I want to use some local computers that I own to operate those tasks (a Raspberry Pi 4, an old 32bit linux PC, etc.) My question : I want to know what would be the Dagster way of using those machines to run jobs and still monitoring them on my main PC dagit instance ? (I'm not so good at Docker and absolutely noob in K8S) My ideas : • Joining dagit instances on multiple PC with specific ops making graphQL calls, maybe with a Fastapi "hub". My main PC would send runjob calls and receive AssetMaterialization events for example. • Maybe run celery servers on those machines and configure dagit to interact with them on specific jobs (but I'm not sure to understand how it's implemented) • Else ? Sometimes when you learn new concepts, the problem or the solution you focus is wrong, so feel free to correct me in any level you see necessary 🙂 I may also have missed a doc or example dealing with this kind of use case Thanks a lot !

:dagsir: Tell us about how you organize your Dagster code Hi everyone, I’m looking to learn more about how you organize your complex Dagster projects and the troubles you are facing when organizing the code. Besides, when structuring your projects, are there any goals you’re trying to optimize? We’re working on a more comprehensive recommendation for Dagster project structure. In the past, we’ve come up with the experimental Create a New Project which comes with a CLI command that scaffolds a project skeleton, but we’ve heard great feedback about it not being well suited for larger complex and prod-scale projects. So I’d love to learn from you about any good practices when you design and structure your data projects. Here’s a GitHub discussion for collecting feedback and scenarios.

Are many people using Dagster to coordinate Deep Learning training runs on cloud GPUs? If so, how's it going for you? The use of sensors appeals to me, and a kind of large scale control flow. Like if this training run goes well, finetune this other thing. If training goes well enough on a small dataset, evaluate on a larger dataset and dump some report somewhere etc. And for maintaining an augmented image set (maybe just for the slowest augmentations) to speed up individual experiment time

New blog post: Orchestrating Machine Learning Pipelines with Dagster by @sandy https://dagster.io/blog/dagster-ml-pipelines

This looks like a better place to ask :daggy-3d: https://dagster.slack.com/archives/C01U954MEER/p1667513159491489

We’re looking at wandb.ai for our future MLOps, any dagster companies here using it and can you provide pros and cons?

How do people handle dynamic partitioning (each new file coming in causes a sensor to fire, and then produce a graph, ultimately resulting in a number of assets getting updated)

have anyone tried this with Dagster? https://prql-lang.org/ dbt is powerful and we use it heavily now. With Dagster, I could see a future without the need of dbt DAG. one other benefit of dbt is the SQL + jinja, just curious if there is a nice way to write those super-charged SQL transformation easily in Dagster without dbt.

How do you source your data in pre-production environments? For faster development and increased security we want to have a fully separate pre-production cloud environment. Ideally, it would have an exact copy of the data present in our production warehouse or lake. Problem is that this poses a security risk and is not cost effective. How are you dealing with this issue? Do you have an anonymization pipeline from prod into staging? Do you generate fakes? It seems complicated for us since our dbt project is very large and some raw sources can’t be easily reproduced. Especially SaaS sources that rarely have a non-prod equivalent as compared to an operational database which lives by default in pre-production.