I'm running a composite_solid to process a DynamicOutput in several steps, much like this example. However, when I try to define some inputs in the configuration file, I receive this error in dagit:

dagster.check.CheckError: Failure condition: Unexpected dynamic output dependency in regular fan in, should have been caught at definition time

are there any plans to make dagit compatible with graphql-core 3?

Hey all! The scheduler has recently been catching this exception but the schedule runs fine. Any idea what might be causing it? P.S. sorry for the picture. Don't have access to slack on my work PC.

Hey, what's the recommended way for doing CI/CD on a git repo of pipelines? For example, I want my team to be able to simply push a new pipeline definition and have it "picked up" by dagster and viewable in dagit etc. My main concern is do I need to restart dagit for changes to my pipelines to be noticed?

Hey all! I'm getting a "No space left on device" error, but the machine has plenty of space. OSError: [Errno 28] No space left on device: '/opt/dagster/dagster_home/storage/5cb15f29-b5fd-403e-9cda-632d8fd7112c' Our storage driver configuration is this: Storage Driver: overlay2 Backing Filesystem: extfs Supports d_type: true Native Overlay Diff: true I read that the this filesystem has a limited number of inodes. Could that be causing this error? What should I do?

are resources cached in any way in dagit (or dagster in general)? e.g. if we're using a resource with no config schema, will it spin up a new instance of that resource for each pipeline run, or use the same instance for each run?

Hi everyone! Does anyone know if there is an option to filter by run timestamp in the daggit "Runs" page?

Hi all, Does anyone know if there is a way to "side load" inputs to dynamically-generated (via .map) solids? My use case loosely resembles the following:

@pipeline
def my_pipeline():
  features = ...
  untrained_models = generate_models()
  
  trained_models = untrained_models.map(
    train_model  # I want to also pass in "features" here
  )
  
  best_model = pick_best(trained_models.collect())
  ...

The

generate_models

solid yields a variable number of untrained model instances, driven by config. Is there another obvious way to do this that I'm missing, besides feeding the

features

output through

generate_models

?

Hello all, is there a way to uniquely identify a pipeline in combination with the provided input configuration? For example, if I run a pipeline with the same code and the same parameters a second time, I only want to process data that has not yet been processed by this particular pipeline / input parameter combination. But for this I have to identify wether a specific pipeline version with specific parameters already has been executed in the past. My recent idea is to compare the git commit of the pipeline file and the current input config with previous executions which are stored in run storage. Is there maybe a better / easier way or a best practice ? Thank you.

Hi All, I'm a newbie in dagster and we are trying to do a PoC to see if dagster is right fit for us. We have spark (scala) and hive jobs running on a internal thousand node big data cluster. The way we are planning to deploy dagster is to have the dagit installed on the internal cloud environment using kubernetes and docker and have the dagster as service running on the edge nodes of the cluster. Dagit and dagster should then communicate via network where we would need to open some ports between them. Question - 1. Is this the right approach we are moving forward with? 2. What ports should be opened between the internal cloud and bigdata cluster?

Hey all, I'm getting some errors when trying to run tests and when loading a user code repository:

dagster.core.errors.DagsterImportError: Encountered ImportError: `cannot import name 'DOCKER_IMAGE_TAG' from 'dagster.core.storage.tags'

Any idea what this comes from? It appears to be around the import for

from dagster_celery_k8s import celery_k8s_job_executor

Hello Dagster community. I have been experiencing performance issues with the Dagit UI (0.11.6) in the last few days. Since yesterday it became completely unusable - all tabs were either unresponsive or would not load at all. Upon checking the logs I see the same BlockingIOError reported here https://github.com/dagster-io/dagster/issues/1652. I have then purged the Dagster storage, history and schedules folder and the UI became responsive and fast again. What concerns me is that I am still running at a small fraction of what will be the full extent of the project. How can I avoid these performance problems going forward? • I am ok with purging/archiving old runs - how can I do that programatically and what should I purge, specifically? What affects the UI performance the most? Is there any way to later access archived runs? • I could potentially split up the project so that it hits multiple instances and corresponding DBs? At the moment all repos runs on the same instance, on a single EC2 machine. Computing resources have not been an issue so far so I am reluctant to scale up. Only problem so far is the Dagit UI speed / responsiveness. I am attaching my config in case you can think of any easy wins.

Hey 👋 Is it possible to trigger multiple pipeline runs from a single

@daily_schedule

partitioned-based schedule? The idea being that I would like to refresh the last N partitions every day. I saw the

partition_days_offset

parameter, which sounds like it would result in the expected behaviour if I created N schedules and incremented the

partition_days_offset

for each — but I’m curious if there’s a way to directly do this within a single schedule.

Hi I am trying to setup reverse proxy in front of dagit. The page opens but is blank.

Hi Everyone; I’m getting an exception when automatically running a pipeline: How would I resolve this?

Hey, I'm getting:

dagster.core.errors.DagsterUnknownResourceError: Unknown resource `custom_resource`. Specify `custom_resource` as a required resource on the compute / config function that accessed it.

I have the resource decorated in it's own file with

@resource

and it's loaded in the mode definition as:

resource_defs={ "custom_resource": custom_resource }

Am I missing something obvious with how the resource needs to be loaded?

Hi, is there a way to run a schedule programmatically from user-code (that defines the schedule)? I read here (https://docs.dagster.io/concepts/partitions-schedules-sensors/schedules#running-the-scheduler) that the schedule needs to be started/stopped from dagit or via dagster schedule start/stop commands

does dagster have utilities to translate python types to the equivalent dagster config type (if possible)? i'm trying to write a decorator to convert a dataclass into a resource and am having difficulty converting the dataclass field list into a config schema

I am having troubles defining a partition set with solid_selection. Specifically, if I do not pass the config of the left out solids I have an error ‘Minimum acceptable config for this path..’ saying that i should include the left out solid configs via run_config_fn_for_partition. If I pass them, on the other hand, I get an error saying that the pipeline (with sub selected solids) does not have those solids. How do I get it to work?

Hey 👋 I’m wondering if this is a bug or something incorrect on my end? I started getting this error in the Playground in Dagit after I upgraded from

0.11.0

to

0.11.6

today, while nothing changed in the code.

AttributeError: 'ConfigEnumValueSnap' object has no attribute 'config_value'

I have an

Enum

field with `EnumValue`s in the config schema for one of my solids, and I get this as a

PythonError

when I open the Playground tab without the config. This gets resolved when I do add a valid config for that field. It initially made me think something was wrong in the code and so it took quite a while before I realized it was simply because I was missing the config value in the Playground config.

Hi, we have a set of pipelines and let's say a particular table is being used in each of those pipelines. If that table got messed up or failed to update for a particular day, we might have to launch backfills on all the pipelines that depend on it. Question: Is there any way to launch backfills easily for a set of pipelines for a particular date?

Does dagit provide something like a read-only UI? I basically want my Dagit users to have visibility into the pipelines execution, without them being able to take any actions. Actions could be triggering a Pipeline execution, toggling a schedule, etc (any writes basically)

Hi there, I have a couple (quite dense) questions to do with project structure and docker deployment on ECS. I have two projects I want to deploy in their own container (from this idea but extending docker_example_pipelines service in docker-compose to maybe docker_project_A, docker_project_B). The logic here would be to have each project (which would be its own package like here) in its own running container as a gRPC server: Root/ src/ project_A/ my_packageA/ init.py repository.py solids/ pipelines/ … my_package_testA/ setup.py requirements.txt  workspace-projectA.yaml # For local development purposes project_B/ my_packageB/ init.py repository.py solids/ pipelines/ … my_package_testB/ setup.py requirements.txt  workspace-projectB.yaml # For local development purposes dagster.yaml # This will be copied over in the docker images docker-compose.yaml Dockerfile_dagster Dockerfile_projectA Dockerfile_projectB workspace.yaml # This will be copied over in the docker images I suppose my argument here is if we containerise both project_A (Dockerfile_projectA), project_B (Dockerfile_projectB) separately (in ECS say) if someone does a blunder and pushes broken code in project_B my project_A repository with all its pipeline code will still run and not fall apart? Is this a more standard approach to a multiple project based workflow for dagster? I watched a video of your talk (33:40) about reloading repository but I felt the example was too simple having two separate repository modules with all logic within them, doesn’t really extend to larger projects so well. Question 2, extending the above, if we make a change to project_A and add another pipeline say I would have to rebuild the image and push that. I’ve read about some git-sync options with Kubernetes (but not really an option with Kubernetes for us), would an alternative method be to set up a volume like EFS to both containers holding project_A, project_B (read only) and then just have some CI/CD setup to push to EFS for every change? I’m hoping this might solve the problem of having to rebuild the image every time there is a change? Sorry for the longwinded questions. Would really appreciate the feedback. Thanks.

Hi all, I'm working on a project where we want to run Spark jobs on Kubernetes, read from S3 and save intermediate results there (especially parquet) and orchestrate via Dagster. My experience with all these tools is still fairly basic and I'm trying to figure out the best way to combine them effectively while avoiding overcomplication. So far I managed to run Spark on a local Kubernetes cluster (Microk8s) and read a csv file from S3 into a Spark Dataframe (with Dagster). In general I also managed to deploy Dagster to Kubernetes and run pipelines from there. I am currently planning my next steps and I was wondering if anyone could give me feedback on my planned approaches: 1. As a next step I would like to combine what I have so far and read the csv from S3 from within the Dagster deployment on Kubernetes. Locally I manually login via

aws sso login

which creates temporary credentials. I created a custom

pyspark_s3_resource

which accesses these credentials via

boto3.Session().get_credentials()

and adjusts the `pyspark_resource`'s hadoop config, so it can read from S3. However, I am unsure how I can access these temporary credentials from within the Dagster Pod on Kubernetes. I was suggested to mount the folder with the credentials into the Dagster user code Pod via hostPath but I'm unsure how to do that and if it's a valid solution. Any thoughts on that? (I'm only interested in a quick workaround solution for my local cluster as the AWS authentication will be solved differently in our Production cluster on EKS.) 2. Are there any best practices on how to run Spark jobs efficently with Dagster? A naive approach would be to save all intermediate results of each solid (especially Dataframes as parquet) on S3, however saving ALL intermediates and starting new Spark sessions in every solid effectively negates the advantages of Spark, i.e. lazy evaluation, caching, etc. This could be avoided by combining several solids into a monolith solid but this would contradict the single-responsibility principle (each step only does one thing). Is it possible to share a single Spark session in several consecutive solids within a pipeline, and e.g. pass the results from one solid to another via a custom IOManager that caches the results instead of saving them, or only passes them without doing anything? 3. As my pipeline will contain several steps that don't depend on one another, I would like to run solids in parallel as well. This means running independent Spark jobs in parallel. As I'm not a Spark expert, I don't know what's the best approach to do so, especially on Dagster and on Kubernetes. Is this something the celery-kubernetes executor can solve? (is it recommended to combine Spark and Celery at all)? I know this is a lot at once but even partial help on any of these questions would be greatly appreciated as the entire framework is starting to get a bit overwhelming, based on the more and more complex infrastructure requirements from our core developer and DevOps team.

I'm experiencing an issue where at least one schedule fails to start each evening. Has been the same schedule last couple days but on previous days it was a different schedule. (Dagster 0.11.6 running on k8s) The daemon is crashing shortly after it evaluates the schedule and when it restarts I can see a job within Dagit in (started state) but no actual steps have started nor has a k8s job begun Edit: was actually STARTING state not STARTED

2021-04-30 03:00:17 - SchedulerDaemon - INFO - Evaluating schedule `load_all_tables_schedule` at 2021-04-30 03:00:00+0000 
2021-04-30 03:00:18 - SensorDaemon - INFO - Not checking for any runs since no sensors have been started. 
2021-04-30 03:00:20 - BackfillDaemon - INFO - No backfill jobs requested. 
2021-04-30 03:00:23 - SensorDaemon - INFO - Not checking for any runs since no sensors have been started. 
2021-04-30 03:00:28 - SensorDaemon - INFO - Not checking for any runs since no sensors have been started. 
2021-04-30 03:00:33 - SensorDaemon - INFO - Not checking for any runs since no sensors have been started. 
2021-04-30 03:00:38 - SensorDaemon - INFO - Not checking for any runs since no sensors have been started. 
2021-04-30 03:00:43 - SensorDaemon - INFO - Not checking for any runs since no sensors have been started. 
2021-04-30 03:00:48 - SensorDaemon - INFO - Not checking for any runs since no sensors have been started. 
2021-04-30 03:00:50 - BackfillDaemon - INFO - No backfill jobs requested. 
2021-04-30 03:00:53 - SensorDaemon - INFO - Not checking for any runs since no sensors have been started. 
2021-04-30 03:00:58 - SensorDaemon - INFO - Not checking for any runs since no sensors have been started. 
2021-04-30 03:01:03 - SensorDaemon - INFO - Not checking for any runs since no sensors have been started. 
2021-04-30 03:01:08 - SensorDaemon - INFO - Not checking for any runs since no sensors have been started. 
2021-04-30 03:01:13 - SensorDaemon - INFO - Not checking for any runs since no sensors have been started. 
2021-04-30 03:01:18 - SensorDaemon - INFO - Not checking for any runs since no sensors have been started. 
2021-04-30 03:01:20 - BackfillDaemon - INFO - No backfill jobs requested. 
2021-04-30 03:01:23 - SensorDaemon - INFO - Not checking for any runs since no sensors have been started. 
2021-04-30 03:01:58 - dagster-daemon - ERROR - Thread for SCHEDULER did not shut down gracefully 
    self.check_daemon_heartbeats() 
    sys.exit(main()) 
Exception: Stopping dagster-daemon process since the following threads are no longer sending heartbeats: ['SCHEDULER'] 
  File "/usr/local/lib/python3.7/site-packages/click/core.py", line 1066, in invoke 
  File "/usr/local/lib/python3.7/site-packages/dagster/daemon/controller.py", line 192, in check_daemon_loop 
  File "/usr/local/bin/dagster-daemon", line 8, in <module> 
  File "/usr/local/lib/python3.7/site-packages/click/core.py", line 829, in __call__ 
    return self.main(*args, **kwargs) 
Traceback (most recent call last): 
  File "/usr/local/lib/python3.7/site-packages/dagster/daemon/controller.py", line 175, in check_daemon_heartbeats 
    return _process_result(sub_ctx.command.invoke(sub_ctx)) 
  File "/usr/local/lib/python3.7/site-packages/dagster/daemon/cli/__init__.py", line 39, in run_command 
    return ctx.invoke(self.callback, **ctx.params) 
    controller.check_daemon_loop() 
    failed_daemons=failed_daemons 
  File "/usr/local/lib/python3.7/site-packages/click/core.py", line 610, in invoke 
  File "/usr/local/lib/python3.7/site-packages/click/core.py", line 1259, in invoke 
    cli(obj={})  # pylint:disable=E1123 
    return callback(*args, **kwargs) 
  File "/usr/local/lib/python3.7/site-packages/click/core.py", line 782, in main 
    rv = self.invoke(ctx) 
  File "/usr/local/lib/python3.7/site-packages/dagster/daemon/cli/__init__.py", line 126, in main

Hey, I couldn't find anything on this but what's the easiest ways to use k8s secrets as env vars? I've tried adding it to the user code repository and the celeryk8s run launcher with no luck under the

envSecrets

section. The creds appear in the pod but not in the run pod that gets launched by celery. Is my best bet adding it as container config on the pipeline or am I missing something obvious? I should add, I'd prefer not to use creds in a pipeline/config yaml if at all possible. I don't want other people who might be running the pipeline to see the credentials for apps

Deploy on a dedicated ressource (V100) Hello, I work on a local PC (mainly window) with a team on different pipelines. We have V100 for stronger computational ressoirces and openshift platform accessible. How would you setup a simple architecture to dev locally, run on V100 and monitor centrally ? Let's say we can have different repo for each pipelines. Openshift can be optional. I try to look at docker exemple but fail too see how to do that ? What should go where ? How to auto update if git repo update etc? I think about a sensor that check change in hash to trigger. Any advises appreciated :)

Hey! Is there a way to use a resource inside a pipeline itself? Something like this:

@pipeline(
   mode_defs = [basic_mode]
)
def test_pipeline(context, input1, input2):
   resources.basic_mode.run()

Hi All, if the "dagster-daemon run" is executing on a different node compared to gRPC server, where can I provide the host:port details for the same in the daemon?

Good day, everyone! When I use postponed evaluation of annotations (Python 3.8) in dagster code, DagsterInvalidDefinitionError is thrown. E.g.

from __future__ import annotations

@solid
def dmt_verify_solid(context: Any, data_release_directory: str) -> Output:
   ...

I get:

dagster.core.errors.DagsterUserCodeProcessError: dagster.core.errors.DagsterInvalidDefinitionError: Invalid type: dagster_type must be DagsterType, a python scalar, or a python type that has been marked usable as a dagster type via @usable_dagster_type or make_python_type_usable_as_dagster_type: got str.

What's the reason? It there a way to resolve it while keeping the annotations feature?