I am facing issues in logging the dagster stdout locally in a text file. As per tutorial, trying '*LocalComputeLogManager*' class, I have tried using the same as a run-config in a pipeline(getting error 'Cannot pickle module'). Currently, I am using dagster version 0.11.7 Following is the code for the same: path = r'C:\Desktop\Log1.txt' execute_pipeline(pipeline=demo_pipeline ,run_config={'compute_logs':{'module':dagster.core.storage.local_compute_log_manager, 'class': LocalComputeLogManager, 'config' : {'base_dir': path} } })

Hi, I was wondering if anyone has a template of deploying their multi container dagster instance (like this) on ECS? I tried using docker compose which now generates a ecs fargate behind the scenes but this

volumes: # Make docker client accessible so we can terminate containers from dagit
      - /var/run/docker.sock:/var/run/docker.sock

fails due to not being able to mount to a host?

I am facing errors in using 'compute_logs' in dagster.yaml file, on executing the pipeline it throws me this error: raise DagsterInvalidConfigError( dagster.core.errors.DagsterInvalidConfigError: Error in config for pipeline demo_pipeline Error 1: Received unexpected config entries "['compute_logs', 'run_storage']" at the root. Expected: "['execution', 'intermediate_storage', 'loggers', 'resourc es', 'solids', 'storage']." This is my Yaml file compute_logs: module: dagster.core.storage.local_compute_log_manager class: LocalComputeLogManager config: base_dir: 'C:\Desktop' Currently using dagster version 0.11.7

Hello! I have a pipeline right now that takes a single image as input (from s3) and runs it through a bunch of solids which call external services. A single run for a single image usually takes approximately 30 seconds. We have it deployed on k8s with a QueuedRunCoordinator which spins up one pod per run (so one pod per image). We're trying to call this pipeline for 5000 images at once. We have tried having an external python script that uses the graphql api to submit the 5000 runs with a loop. This approach crashed our dagit/graphql service pods more often than not. I'm currently testing a sensor approach with the sensor looking at the s3 bucket for new files (similarly to the documented examples). This approach is much more reliable (doesn't crash dagit/graphql) but the queue is never being fully utilized. Our queue is configured with 100 max concurrent runs but we only ever reach about 60-70 running with about 40-50 in the queue. It seems like the daemon running the sensor is throttled so the queue never fills up. This is causing our overall process to slow down quite a bit. Are there ways that I could increase the throughput of the sensor? I have tried increase the k8s resource allocation to the dagster-daemon but it doesn't seem to do much. Or alternately, is this approach of a single image pipeline actually feasible? Should I instead be looking at the pipeline input being the whole s3 bucket (possibly using dynamic mapping)? Currently using version 0.11.4 but can easily upgrade if needed. Thanks in advance for help and suggestions.

Hey 👋 We’re running into issues trying to use

multiprocess

execution — our pipeline stalls when Dagster starts multiple threads as if it’s waiting for a signal which it never receives. This only happens in production in our kubernetes deployment with version 0.11.6 — things work fine locally with the same config though. For reference, we have the following run config:

execution:
  multiprocess:
    config:
      max_concurrent: 8
resources:
  io_manager:
    config:
      s3_bucket: <BUCKEt>
      s3_prefix: <PREFIX>
  s3:
    config: {}

With the

io_manager

being

s3_pickle_io_manager

. I can confirm the AWS credentials used by the cluster have access to the bucket as they’re using the same credentials/bucket for other pipelines that work fine when not using the multiprocess execution. Any ideas why that could be?

is there a way to access the mode a pipeline is running in within a pipeline definition? my team follows a pattern of loading mode-specific config for solids and resources using a wrapper to preconfigure the object with said config, e.g.

preconfigure_for_mode(resource, "dev")

, but this doesn't seem to be doable for solids in the context of a pipeline, since pipelines aren't passed a context argument.

Good afternoon. Where can I find some examples of Dagster running with Dataframe? thank you.

I’m interested in using sensors and schedules for automating my runs but my understanding of daemons is quite limited. Is this feature more aimed at a cloud app that never goes down or is it usable with local machines that eventually get turned off as well? Do I have to start the process every time I turn the computer on? Does it check for changes that happened while it wasn’t running (i.e the addition of a file that would trigger a sensor)?

Hi all! Is there a way to generate a template run_config dictionary or yaml file for a pipeline in a python file (not within the dagit ui)? I've been putting a bad config function into the pipeline the first time and using the dictionary provided in the error message for the expected configuration dictionary to give me all the right keys. Thanks!

Hello everyone! I am trying to implement an example given here https://docs.dagster.io/concepts/partitions-schedules-sensors/schedules but I am having troubles making it work. The specific example on the page is this one:

@daily_schedule(
    start_date=datetime.datetime(2020, 1, 1),
    pipeline_name="my_pipeline",
    solid_selection=preset.solid_selection,
    mode=preset.mode,
    tags_fn_for_date=lambda _: preset.tags,
)
def my_modified_preset_schedule(date):
    modified_run_config = preset.run_config.copy()
    modified_run_config["date"] = date.strftime("%Y-%m-%d")
    return modified_run_config

When I am selecting the partition in dagit's playground, the config is updated and the right date is picked up. However, when I try to run a backfill, the config is not updated for each run. Instead, it seems to take only the config of the last partition selected in dagit. Here is the exact code for the schedule I am trying to run, I created it in isolation to make sure it wasn't something else in my code causing the issue:

@solid(config_schema={"test_in": str})
def backfill_solid(context):
    test_in = context.solid_config["test_in"]
    <http://context.log.info|context.log.info>(f"test_in is {test_in}")



MODE_BUG = ModeDefinition(name="MODE_BACKFILL_BUG", resource_defs={"io_manager": fs_io_manager})

PRESET_BACKFILL_BUG = PresetDefinition.from_files(
    name="PRESET_BACKFILL_BUG",
    mode="MODE_BACKFILL_BUG",
    config_files=[
        file_relative_path(__file__, "./presets_backfill_bug.yaml")
    ])



@pipeline(mode_defs=[MODE_BUG],
          preset_defs=[PRESET_BACKFILL_BUG])
def backfill_pipeline():
    backfill_solid()



@daily_schedule(
    pipeline_name="backfill_pipeline",
    start_date=datetime.datetime(2021, 5, 1),
    execution_time=datetime.time(9, 0, 0),
    solid_selection=PRESET_BACKFILL_BUG.solid_selection,
    mode=PRESET_BACKFILL_BUG.mode,
)
def backfill_schedule(date):
    modified_run_config = PRESET_BACKFILL_BUG.run_config.copy()
    modified_run_config["solids"]["backfill_solid"]["config"]["test_in"] = date.strftime("%Y-%m-%d")
    return modified_run_config

I attached the run configuration I see in dagit for one of the runs I use Dagster 0.11.8 and my Python version is 3.8.5

Hi, we have our Dagster

pipeline

that contains several

composite solids

, is there a way to manage the order of execution if there are no dependencies. (e.g. we want some

composite_solid

to be executed at the beginning before the other composite solids are executed). Any ideas?

Hi Is it possible to create custom config types(eg, for dates)? Thanks

Hello! I'm trying to use IOManager to save files into folders. Is there anyway I could send some dynamic filename into IOManager? They are only defined at runtime (for example, the filename should be the result of some calculation), so I can't define it in the pipeline resources.

In dagit is there a way to select a solid that starts with similar name as another solid without selecting both? I think previously I could put a comma after the name and it wouldn't choose the other. Example: load_table_Contact, load_table_ContactHistory If I put a solid selection of load_table_Contact dagit wants to run both of them, even if I put a trailing comma

Hi people, I have just started with the

hello_cereal.py

tutorial and I encountered this error. Can anyone help me with this? I am running in conda environment. (MacOS)

➜  dagster-tutorial dagit -f hello_cereal.py
Using temporary directory /var/folders/ml/cyrfcd_d5l1fn6nvkb4d3y14kdlv1s/T/tmp8gnm3ukt for storage. This will be removed when dagit exits.
To persist information across sessions, set the environment variable DAGSTER_HOME to a directory to use.

Loading repository...
Serving on <http://127.0.0.1:3000> in process 95032
Traceback (most recent call last):
  File "/opt/anaconda3/envs/dagster-papermill/lib/python3.8/site-packages/gevent/pywsgi.py", line 999, in handle_one_response
    self.run_application()
  File "/opt/anaconda3/envs/dagster-papermill/lib/python3.8/site-packages/geventwebsocket/handler.py", line 75, in run_application
    self.run_websocket()
  File "/opt/anaconda3/envs/dagster-papermill/lib/python3.8/site-packages/geventwebsocket/handler.py", line 52, in run_websocket
    list(self.application(self.environ, lambda s, h, e=None: []))
  File "/opt/anaconda3/envs/dagster-papermill/lib/python3.8/site-packages/flask/app.py", line 2069, in __call__
    return self.wsgi_app(environ, start_response)
  File "/opt/anaconda3/envs/dagster-papermill/lib/python3.8/site-packages/flask_sockets.py", line 40, in __call__
    handler, values = adapter.match()
  File "/opt/anaconda3/envs/dagster-papermill/lib/python3.8/site-packages/werkzeug/routing.py", line 2026, in match
    raise WebsocketMismatch()
werkzeug.routing.WebsocketMismatch: 400 Bad Request: The browser (or proxy) sent a request that this server could not understand.
2021-05-11T23:57:14Z {'REMOTE_ADDR': '127.0.0.1', 'REMOTE_PORT': '62035', 'HTTP_HOST': '127.0.0.1:3000', (hidden keys: 32)} failed with WebsocketMismatch

Hi all, For years I've worked with apache airflow and got used to it. Recently a guy I interviewed mentioned they are using dagster, which was the first time I've heard about it. Knowing airflow is not perfect, I'm wondering for a new project I'm now starting to work on, why should I choose dagster over airflow? Sorry if it was asked already, I couldn't find this question online or here on slack.

Hello world! I have a solid that always takes an input from config. It can be a dict, int, or in case it's not passed I want it to default to None. I've tried to define the solid variable as

@solid
def my_solid(
	context,
	var1: dagster.Optional[dagster.Any] = None
	...
):

but when I try to set a value for the input in dagit config like:

solids:
	my_solid:
		inputs:    
			var1: 42

or

solids:
	my_solid:
		inputs:    
			var1: {"ke1":"value1"}

I get:

Expected: "{ json: { path: String } pickle: { path: String } value: Any }"

I've also tried to define a dagster.DagsterTypeLoader to accept any python Object like:

@dagster_type_loader(Permissive())
def load_object(_context, value):
    return value

and then define the solid input as

@solid
def my_solid(
	context,
	var1: dagster.Optional[dagster.PythonObjectDagsterType(python_type=object, loader=load_object)] = None

	...
):

but then dagit insist that the input should be a dict. How can I admit dict/int/None as an input that will only be defined via config? Thanks!!!

Hi again, i am trying to figure out how to start a scheduled pipeline based on a PresetDefinition or even better based on a ModeDefinition. I found this https://github.com/dagster-io/dagster/issues/2056 but it seems to be outdated? What is the current way to schedule a pipeline based on on of its mode_defs and/or preset_defs?

is it possible to use two environment variables as config for a solid? I tried a few variations of the

StringSource

documentation’s example without success

Hi all! I'm running dagster in docker (pretty much as in your docs) and it works fine. But now I need to run some docker containers from within a solid. Can you please suggest the best way to do it?

Hello!! I'm trying to modify pipeline priorities, but I'm getting the error: dagster.check.CheckError: Invariant failed. Description: User attempted to set tag with reserved system prefix: dagster/priority I'm on dagster 0.11.8 and I followed these instructions: https://docs.dagster.io/deployment/run-coordinator#priorities I'm setting the tags directly to the pipeline as in the example

anyone else having trouble with

pipenv install dagit

? I get

ERROR: Could not find a version that matches graphql-core<3,<4,>=2.0,>=2.1,>=2.3,>=2.3.2,>=3.1.2

but dagster installs fine

I tried looking in tickets but I don't see anything, have there been any changes towards letting existing pipeline runs restart where the runner crashes? Like not a solid has failed, but when the runner itself is interrupted.

Hi, we have a pipeline that runs daily (on a schedule), The pipeline run config gets an effective date which drives the logic for its data extract. A by-product of the pipeline run is that certain conditions might be met that necessitate the same pipeline (or similar) to be run with a range of backdated effective_dates, a backfill of sorts. The backfill would need a

PartitionSet

with a

partition_fn

that works out the dates based on something we can access at the time it is called, so it might need access to a Resource or at least an asset for which we've created an

AssetMaterialization

event. But I don't have access to a context with a run_id, so not sure how I can best achieve this. Even if I can work out how to create the PartitionSet, how do I kick off the backfill? It seems I can only do it via CLI or Dagit UI. I would like to be able to do it via the python or GraphQL API. What I want to do seems like a reasonably commonplace use case, so I feel I must be missing something obvious about the way Dagster backfills work. Can someone help? Thanks Paul

When using a local ephemeral dagit (v0.11.8) instance, the Gantt & Log views don't update during a pipeline run (see attached video). I have refresh the full browser window to get them to update. Since the pipeline status icon at the top updates without a refresh, I'm wondering if this a bug? Or by design due to some limitation of running an ephemeral instance of dagit?

$ dagit -p 3000 -w dagster_workspace.yaml
Using temporary directory /var/folders/cx/c_41bpzs1jlc465fx_k_9qfh0000gn/T/tmpe442ol4u for storage. This will be removed when dagit exits.
To persist information across sessions, set the environment variable DAGSTER_HOME to a directory to use.

Loading repository...
Serving on <http://127.0.0.1:3000> in process 10619

Any debugging thoughts on this error message when trying to run Dagit locally? (Suspect it might be due to an adblock / network lockdown extension on a corporate PC; just not sure where to start looking) Most of Dagit loads ok; but the

/graphql

queries seem to fail and cause Dagit to emit error logs like the below:

Traceback (most recent call last):
  File "/Users/dbasner/.local/share/virtualenvs/tanzu-dm-Qg2J58No/lib/python3.7/site-packages/gevent/pywsgi.py", line 999, in handle_one_response
    self.run_application()
  File "/Users/dbasner/.local/share/virtualenvs/tanzu-dm-Qg2J58No/lib/python3.7/site-packages/geventwebsocket/handler.py", line 75, in run_application
    self.run_websocket()
  File "/Users/dbasner/.local/share/virtualenvs/tanzu-dm-Qg2J58No/lib/python3.7/site-packages/geventwebsocket/handler.py", line 52, in run_websocket
    list(self.application(self.environ, lambda s, h, e=None: []))
  File "/Users/dbasner/.local/share/virtualenvs/tanzu-dm-Qg2J58No/lib/python3.7/site-packages/flask/app.py", line 2069, in __call__
    return self.wsgi_app(environ, start_response)
  File "/Users/dbasner/.local/share/virtualenvs/tanzu-dm-Qg2J58No/lib/python3.7/site-packages/flask_sockets.py", line 40, in __call__
    handler, values = adapter.match()
  File "/Users/dbasner/.local/share/virtualenvs/tanzu-dm-Qg2J58No/lib/python3.7/site-packages/werkzeug/routing.py", line 2026, in match
    raise WebsocketMismatch()
werkzeug.routing.WebsocketMismatch: 400 Bad Request: The browser (or proxy) sent a request that this server could not understand.
2021-05-13T14:39:37Z {'REMOTE_ADDR': '127.0.0.1', 'REMOTE_PORT': '57343', 'HTTP_HOST': '127.0.0.1:3000', (hidden keys: 31)} failed with WebsocketMismatch

Hi , I am running

dagit -w workspace.yaml

but getting following error

Error loading repository location repo.py:dagster.core.errors.DagsterUserCodeProcessError: dagster.core.errors.DagsterInvariantViolationError: File or glob pattern "config.yaml" for "config_files"produced no results.

This is my workspace.yaml:

load_from:

- python_file:

relative_path: test_project/scripts/repo.py

working_directory: C:/Users/Deveshi/source/test_project/scripts

If I just move to folder containing repo.py and run dagit -f repo.py, it works I am using a preset that loads config from config.yaml my dagster version is 11.0 Thanks!

Hey is there any way to view the configuration that a schedule will use on it's next run in Dagit? I've updated the configuration with the preset definition, but I can't really tell if I did it right or not

Hey! Has anyone been able to deploy the Dagster docker compose arrangement onto AWS? I'm looking to do this, but I'm fairly new to deploying a compose file onto AWS. What modifications must be made to the

docker-compose.yaml

?

Hi. What’s an easy way to halt the execution of a pipeline run when a solid raises

Failure

? I naively thought that would stop it, but it seems like I need to use conditional branching to achieve it? Is there an example somewhere? I am having a bit of hard time grasping how I can apply it.