I ran into this error when using the dagster with the DockerRunLauncher. It occurred because I had hundreds of stopped containers from past pipeline runs. I was able to fix it by running

docker rm $(docker ps -aq)

to remove all the stopped containers. Really you'd want the dagster daemon to be removing these itself, any chance this be implemented?

Hello, I'm trying to setup a pipeline that is partitioned and has a schedule. I ended up making this stackoverflow question. Was wondering if anyone would be able to help. https://stackoverflow.com/questions/67928569/how-to-create-partitions-with-a-schedule-in-dagster

is the code shown here available somewhere on Github?

Hi team, I am trying to link my solid's output definition to an asset as explained here https://docs.dagster.io/concepts/assets/asset-materializations#linking-assets-to-an-output-definition. In my case the asset key depends on the solid input passed from the upstream solid's output. Does the

OutputContext

that I can pass to the get_asset_key() have the solids input information?

Hello, I was hoping someone could offer some advice on how to implement this pipeline in dagster such that the API request sections are run concurrently as much as possible (or a simple compromise). The pipeline will operate for multiple files in a directory and the order of the files is not important so if possible the pipeline could run for multiple files concurrently. I could not find much about asynchronous solids in the docs and I'm quite new to all of this. Any advice would be appreciated. Thanks!

Hey 👋 Curious if there’s a good way to access the start time of

PipelineRun

objects from within a sensor? For context, I would like to get alerted when a pipeline has been in

STARTING

/

STARTED

status for more than X hours since pipeline runs occasionally become zombies with our k8s deployment. I can get

PipelineRun

objects from

context.instance.get_runs()

but it’s not clear to me from looking at the class definition that status timestamps are available — should I instead look for the event stream from that

PipelineRun

, if so, is that easily query-able? Thanks!

Hi team, I have just started to using partition sets and wondering how to add other solid specific config in the

run_config_fn_for_partition

in addition to the partition config?

greetings everyone, would appreciate some replies on how is it possible to achieve run-once-per-partition solids - https://github.com/dagster-io/dagster/discussions/4287

Hi Dagster team! I am testing the backfill feature but I am confusing because when I try to launch the backfill from dagit I got this error:

Traceback (most recent call last):

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\site-packages\dagster\core\execution\poll_compute_logs.py", line 61, in <module>

execute_polling(sys.argv[1:])

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\site-packages\dagster\core\execution\poll_compute_logs.py", line 53, in execute_polling

with open(ipc_output_file, "w"):

FileNotFoundError: [Errno 2] No such file or directory: 'C:\\Users\\s4957336\\AppData\\Local\\Temp\\1\\tmpo_zr47w6\\execute-windows-tail-15bc135d998546caa23116bbff00ceb1'

2021-06-14 02:33:51 - dagster - ERROR - Mexico_CTMS_Corp - a1cb3fdf-19ef-4676-a6b0-b3c7610ec0f9 - 22036 - PIPELINE_FAILURE - Execution of pipeline "Mexico_CTMS_Corp" failed. An exception was thrown during execution.

Exception: Timed out waiting for tail process to start

Stack Trace:

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\site-packages\dagster\core\execution\api.py", line 762, in pipeline_execution_iterator

for event in pipeline_context.executor.execute(pipeline_context, execution_plan):

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\site-packages\dagster\core\executor\in_process.py", line 38, in execute

yield from iter(

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\site-packages\dagster\core\execution\api.py", line 841, in __iter__

yield from self.iterator(

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\site-packages\dagster\core\execution\plan\execute_plan.py", line 72, in inner_plan_execution_iterator

active_execution.verify_complete(pipeline_context, step.key)

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\contextlib.py", line 120, in __exit__

next(self.gen)

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\site-packages\dagster\core\storage\compute_log_manager.py", line 56, in watch

yield

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\contextlib.py", line 120, in __exit__

next(self.gen)

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\site-packages\dagster\core\storage\local_compute_log_manager.py", line 51, in _watch_logs

yield

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\contextlib.py", line 120, in __exit__

next(self.gen)

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\site-packages\dagster\core\execution\compute_logs.py", line 31, in mirror_stream_to_file

yield pids

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\contextlib.py", line 120, in __exit__

next(self.gen)

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\site-packages\dagster\core\execution\compute_logs.py", line 75, in tail_to_stream

yield pids

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\contextlib.py", line 120, in __exit__

next(self.gen)

File "C:\Users\s4957336\Anaconda3\envs\borrar\lib\site-packages\dagster\core\execution\compute_logs.py", line 104, in execute_windows_tail

raise Exception("Timed out waiting for tail process to start")

I am not sure why dagster is looking into this path

FileNotFoundError: [Errno 2] No such file or directory: 'C:\\Users\\s4957336\\AppData\\Local\\Temp\\1\\tmpo_zr47w6\\execute-windows-tail-15bc135d998546caa23116bbff00ceb1'

if I configured the dagster.yaml in this way:

scheduler:
  module: dagster.core.scheduler
  class: DagsterDaemonScheduler
run_launcher:
  module: dagster.core.launcher
  class: DefaultRunLauncher
run_coordinator:
  module: dagster.core.run_coordinator
  class: QueuedRunCoordinator
  config:
    max_concurrent_runs: 25
run_storage:
  module: dagster_postgres.run_storage
  class: PostgresRunStorage
  config:
    postgres_url: "<postgresql://user:pwd@hostname>:port/dagster"

local_artifact_storage:
  module: dagster.core.storage.root
  class: LocalArtifactStorage
  config:
    base_dir: C:\Users\s4957336\Documents\caps\local_artifact_storage

compute_logs:
  module: dagster.core.storage.local_compute_log_manager
  class: LocalComputeLogManager
  config:
    base_dir: C:\Users\username\Documents\caps\logs

The folder logs is storing the compute logs

Hello, is it possible to implement the following DAG with dagster such that solids A,B,C,D all operate together for each file in a directory using dynamic outputs? I couldn't seem to find similar examples in previous slack messages or in the docs. Any responses would be greatly appreciated! Thanks

Hello channel. I ran into some major technical issues today after running Dagster for a while. My EC2 instance was spawning a high number of dagster-related processes (7k or so) to the point that it was not accessible via ssh anymore. This happened after I moved to a bigger machine: there is a suspicion that higher capacity meant higher number of spawn-able threads and eventually some hard limit was hit. For context: I have been experiencing UI performance issues that seem to get worse over time as the number of runs increases. Anyways, here’s a couple of questions for you: 1. Does it make sense that the high number of spawned threads and sluggish UI  are the manifestation of the same cause (high number of runs accrued) 2. is there any new feature or plans around time-to-live in the sqlite DB? 3. I am planning to start to rely on Dagster for critical production processes soon. To avoid running into performance issues my plan is to have the live processes to run on a dagster instance and leave all backfills to another machine, running a separate Dagster instance. The idea here is that this latter instance is more easily purge-able (in the absence of a better TTL-related approach). What do you think about it? Is there any recommended way to run such a ‘secondary’ instance?

Hi team, I want to add some vault specific

annotations

to all the pipeline jobs. Is there any way to specify this in the helm for all the jobs, or should I add them to all the pipeline definitions

dagster-k8s/config

?

Hi there! Is there a timeout after which a solid/pipeline gets killed? For context, I'm trying to run a simple pipeline with just 1 solid with dagster running through docker-compose. After about 45 mins the output of the run hangs and processing stops. Also, the temporary docker container that is supposed to execute the job dies without producing any output information. I can see a bunch of messages similar to: https://dagster.slack.com/archives/C01U954MEER/p1621007250146700 so I bumped the

--db-statement-timeout

but it did not help. I'm running v0.11.12. Any ideas?

Hi Dagster Team! Currently ramping up our usage of Dagster after a successful qualification. We ran into an interesting issue: the Dagster daemon got killed due to hitting (misconfigured) memory limits. When the daemon restarted, I would have expected it to fail all the currently running pipelines, but that was not the case: the previously-started pipelines were still showing up as STARTED in dagit, but they were not progressing. I tried to restart our Dagster celery workers then the daemon again, hoping that would unwedge things, to no avail. I had to manually cancel the pipelines from the UI. Is the daemon expected to be able to recover running pipelines on restart?

Hi I want to use memoization for my pipeline My versioned io manager has dependency on pyspark resource And i see that for each solid in the pipeline new pyspark resource is being created. Is there any way to avoid this? Thanks!

Regarding the Dagster scheduler decorator: is it possible to, for example, create a schedule which runs once per month on the last weekday of the month? Typically this sort of thing is done with a script in conjunction with cron - is this functionality supported in any way through Dagster? Thanks!

Can Dagster help with mutual exclusivity across solids and workflows? My use case is preventing DBT from running concurrently with itself. e.g. Both in response to a regular schedule and to a signal.

hello again D friends! i ran into another error using the python graphql client. the connection is now working with the

use_https

flag (thanks again!) but we are now running into this error while trying to launch a pipeline execution. we could see the failed run in the UI, so it did try to execute, but this error does not occur when trying to launch within the UI itself. im not really sure where this is coming from. (code run in thread)

---------------------------------------------------------------------------
DagsterGraphQLClientError                 Traceback (most recent call last)
<ipython-input-13-0164b5809f7d> in <module>
      2         "tatari_pipeline",
      3         run_config=dct,
----> 4         mode="local",
      5     )

/usr/local/lib/python3.7/site-packages/dagster_graphql/client/client.py in submit_pipeline_execution(self, pipeline_name, repository_location_name, repository_name, run_config, mode, preset, tags)
    230             # query_result_type is a ConflictingExecutionParamsError, a PresetNotFoundError
    231             # a PipelineNotFoundError, a PipelineRunConflict, or a PythonError
--> 232             raise DagsterGraphQLClientError(query_result_type, query_result["message"])
    233 
    234     def get_run_status(self, run_id: str) -> PipelineRunStatus:

DagsterGraphQLClientError: ('PythonError', 'sqlalchemy.exc.OperationalError: (psycopg2.OperationalError) SSL connection has been closed unexpectedly\n\n[SQL: INSERT INTO event_logs (run_id, event, dagster_event_type, timestamp, step_key, asset_key, partition) VALUES (%(run_id)s, %(event)s, %(dagster_event_type)s, %(timestamp)s, NULL, NULL, NULL) RETURNING event_logs.run_id, event_logs.id]\n[parameters: {\'run_id\': \'5c42b0e4-7366-4bc4-8d70-cce9a81269ee\', \'event\': \'{"__class__": "EventRecord", "dagster_event": {"__class__": "DagsterEvent", "event_specific_data": null, "event_type_value": "PIPELINE_ENQUEUED", "lo ... (225 characters truncated) ... ne_name": "tatari_pipeline", "run_id": "5c42b0e4-7366-4bc4-8d70-cce9a81269ee", "step_key": null, "timestamp": 1623773911.8121505, "user_message": ""}\', \'dagster_event_type\': \'PIPELINE_ENQUEUED\', \'timestamp\': datetime.datetime(2021, 6, 15, 16, 18, 31, 812150)}]\n(Background on this error at: <http://sqlalche.me/e/14/e3q8>)\n')

Another quick capabilities question: Is there any in-built support for timeouts at either the pipeline or solid level in D ? I see

RetryRequested

for solid retries, but can't find anything for timeouts.

Hey all... super newbie to dagster, but I'm running into a strange issue... I want to load multiple pipelines from multiple repos, but each time I add a new pipeline to my

workspace.yaml

,

dagit

only shows 1 pipeline:

hello_world.py:

from dagster import pipeline, solid

@solid
def get_name():
    return 'dagster'

@solid
def hello(context, name: str):
    <http://context.log.info|context.log.info>(f"Hello, {name}!")

@pipeline
def hello_pipeline():
    hello(get_name())

hello_cereal.py

import requests
import csv
from dagster import pipeline, solid

@solid
def hello_cereal(context):

    repsonse = requests.get("<https://docs.dagster.io/assets/cereal.csv>")
    lines = repsonse.text.split("\n")
    cereals = [row for row in csv.DictReader(lines)]
    <http://context.log.info|context.log.info>(f"Found {len(cereals)} cereals")

    return cereals

@pipeline
def hello_cereal_pipeline():
    hello_cereal()

workspace.yaml:

load_from:
  - python_file: hello_cereal.py
load_from:
  - python_file: hello_world.py

seems pretty straight forward... but I only get the one pipeline showing up in dagit

@daniel Thanks for your help!

Hey all, I'm back. One last question. If I want to use the

python api

to execute pipelines, how do I send them to the current instance of

dagit

?

it seems that when I run a given pipeline via

python api

, it is running on an ephemeral instance instead of my locally hosted instance

127.0.0.2:3000

Please tell me this is a horrible way of executing pipelines:

from dagster import execute_pipeline

import sys

sys.path.append('.')

from hello_cereal import hello_cereal_pipeline

if __name__ == '__main__':



    execute_pipeline(hello_cereal_pipeline)

nevermind, found the solution here: https://github.com/dagster-io/dagster/discussions/3048 seems like a common use case for local development?