Hey team, I've run into the following error when trying to run a pipeline after deploying with docker-compose:

Traceback (most recent call last):
  File "/usr/local/bin/dagster", line 8, in <module>
    sys.exit(main())
  File "/usr/local/lib/python3.8/site-packages/dagster/cli/__init__.py", line 41, in main
    cli(auto_envvar_prefix=ENV_PREFIX)  # pylint:disable=E1123
  File "/usr/local/lib/python3.8/site-packages/click/core.py", line 829, in __call__
    return self.main(*args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/click/core.py", line 782, in main
    rv = self.invoke(ctx)
  File "/usr/local/lib/python3.8/site-packages/click/core.py", line 1259, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/usr/local/lib/python3.8/site-packages/click/core.py", line 1259, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/usr/local/lib/python3.8/site-packages/click/core.py", line 1066, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/usr/local/lib/python3.8/site-packages/click/core.py", line 610, in invoke
    return callback(*args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/dagster/cli/api.py", line 86, in execute_run_with_structured_logs_command
    DagsterInstance.from_ref(args.instance_ref) if args.instance_ref else DagsterInstance.get()
  File "/usr/local/lib/python3.8/site-packages/dagster/core/instance/__init__.py", line 311, in from_ref
    run_storage=instance_ref.run_storage,
  File "/usr/local/lib/python3.8/site-packages/dagster/core/instance/ref.py", line 196, in run_storage
    return self.run_storage_data.rehydrate()
  File "/usr/local/lib/python3.8/site-packages/dagster/serdes/__init__.py", line 342, in rehydrate
    module = importlib.import_module(self.module_name)
  File "/usr/local/lib/python3.8/importlib/__init__.py", line 127, in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
  File "<frozen importlib._bootstrap>", line 1014, in _gcd_import
  File "<frozen importlib._bootstrap>", line 991, in _find_and_load
  File "<frozen importlib._bootstrap>", line 961, in _find_and_load_unlocked
  File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
  File "<frozen importlib._bootstrap>", line 1014, in _gcd_import
  File "<frozen importlib._bootstrap>", line 991, in _find_and_load
  File "<frozen importlib._bootstrap>", line 975, in _find_and_load_unlocked
  File "<frozen importlib._bootstrap>", line 671, in _load_unlocked
  File "<frozen importlib._bootstrap_external>", line 783, in exec_module
  File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
  File "/usr/local/lib/python3.8/site-packages/dagster_postgres/__init__.py", line 4, in <module>
    from .run_storage import PostgresRunStorage
  File "/usr/local/lib/python3.8/site-packages/dagster_postgres/run_storage/__init__.py", line 1, in <module>
    from .run_storage import PostgresRunStorage
  File "/usr/local/lib/python3.8/site-packages/dagster_postgres/run_storage/run_storage.py", line 3, in <module>
    from dagster.core.storage.runs import DaemonHeartbeatsTable, RunStorageSqlMetadata, SqlRunStorage
ImportError: cannot import name 'DaemonHeartbeatsTable' from 'dagster.core.storage.runs' (/usr/local/lib/python3.8/site-packages/dagster/core/storage/runs/__init__.py)

I've followed the deploy docker example and the only difference is the docker_example pipelines dockerfile (where I use my own app). All the containers are up and running but the error rises when a new container is spawned after trying to run a pipeline. Anybody knows why does this happen? Edit: The log error comes from the spawned container (which stops immediately)

Hi Dagster Team dagsir! I want to integrate my dagster pipeline with amazon SQS Queue such that, whenever i receive a message in SQS queue, it triggers my dagster pipeline. Is this even possible through dagster or not ?!!

Hi team, where can I find / how can I setup the logging for sensors?

Hi All! Thanks for this amazing tool. I am working on a proof of concept for my team where we want to dynamically create the graph and distribute the execution over a dask cluster. Does dagster-dask currently support the experimental DynamicOutput? My pipeline fails at the moment the dynamic part should be executed, but I am not sure whether this is an error from my side or that it is just not supported yet.

Related to this query: https://dagster.slack.com/archives/CCCR6P2UR/p1614166979102300 - how can one terminate in bulk all queued runs? from Dagit it seems only 25 queries displayed within the page can be terminated - what if the queued runs fits into several pages?

Hello Dagster folks! My team is building out our first bigger piece of software that uses Dagster (loving it so far). I have a question that involves modes and resources vs. configs. We have defined modes for “prod”, “dev”, “local”, and “test”, and have defined resources to use in each of these. We are now adding a solid that needs to know what environment we’re in, so for the “prod” mode we want a “prod” environment, but for “dev” and “local” we want “dev” as the env. Is it antipattern to simply add a resource that is a string for “prod” vs “dev”? Is this a case of using configs instead? If it is, is there a common pattern for tying it in with our modes? Thanks!

Question about dagster grpc workspaces

Hey! I’m trying to pass a BigQuery table schema into the config for the import_gcs_paths_to_bq solid in

dagster_gcp

and had a question - I’d like to avoid typing out the entire schema in a YAML file and it seems my best bet is to use the

configured

API and write Python to load the schema from a JSON file. Is there any way I can instead specify the path to the file in a YAML file and write a custom type with a corresponding type loader to pass the schema to the solid? My concern then is that the solid config expects a schema of type

Array(inner_type=dict)

but this would then be some other custom type. Also, are custom types even supported for config, or is that only for inputs?

Hey everyone! Does anybody have some experience with the newly-added dynamic orchestration feature? I’m currently stuggling with

DynamicOutput

assignation and I find the example in the documentation a bit vague in this regard. Quick example from the documentation:

@pipeline
def process_directory():
    files_in_directory().map(process_file)

How would one pass the

DynamicOutput

from

files_in_directory

as a specific keyword parameter to

process_file

, as in this:

@pipeline
def process_directory():
    files_in_directory().map(process_file(file_path=mapped_value, other_arg="fixed value"))

Does the mapping rely on the name given to the

DynamicOutput

? Is something like this even possible at the moment?

Sandy Ryza just published a new version: 0.10.7. See https://github.com/dagster-io/dagster/releases/tag/0.10.7 for the release notes.

Hi!

Is there any possibility to debug sensor? I can run pipe_line by

if __name__ == "__main__":
execute_pipeline(
        pipe_line_process_data_exch,
        mode="some_mode"
        , run_config={"loggers": {"my_logger": {}}}

    )

but for sensor I didn't find any example.

Hello friends. I am attempting to learn wonderful Dagster but am having some troubles with the

run_config

passing to

execute_pipeline

. The error I am getting i s:

raise DagsterInvalidConfigError(

dagster.core.errors.DagsterInvalidConfigError: Error in config for pipeline run_pipeline

Error 1: Missing required config entry "solids" at the root.

The problem appears to be that the config that is being passed to the

input_code

function is resolving as NoneType. I've read elsewhere that malformed returns cause the above error, and so it may be that

context.solid_config

is the cause, however this is a fairly minimal, modified from https://docs.dagster.io/tutorial/basics_solids. Any idea what I am doing wrong here?

from dagster import (
    execute_pipeline,
    pipeline,
    solid,
    String,
)



@solid
def market_iterator(context, market_now_is):

    market_mod = market_now_is + '_modified'

    <http://context.log.info|context.log.info>(
        f"Processing {market_mod}"
    )

    return market_mod


@solid(config_schema={"market_id": String})
def input_code(context):
    market_now_is = context.solid_config["market_id"]
    return market_now_is


@pipeline
def run_pipeline():
    market_mod = market_iterator(input_code())


config_dict = {
    "solids": {
        "input_code": {
            "config": {"market_id": "DATA"}}
    }
}


def batch_markets():
    execute_pipeline(run_pipeline, run_config=config_dict)


if __name__ == "__main__":
    batch_markets()

Hey guys I got a question regarding solid naming conventions. Let's say I have a repo where I have one pipeline for foods and another one for drinks. Ideally, I'd like to have my pipelines consist on just 3 composite solids (extract, transform, load) and then inside those composites just call other solids. The problem is that my ETL composite solids are different for foods and drinks, but still I would like to name the composite solids in the same way. Problem: Dagster does not allow me to have solids with the same names even though they are imported from different files. How should I solve this? I'd rather not have solids named c_solid_food_extract, c_solid_drink_extract, c_solid_food_transform, c_solid_drink_transform, c_solid_food_load and c_solid_drink_load.

Hey, Is there a way to handle error conditions for a pipeline? Think about how exceptions are handled in regular code. Similarly are there any ways to handle errors, if something goes wrong anywhere in a pipeline?

Hello everyone, I just started trying out Dagster 2 weeks ago and very happy how I can present the data flow logic to non-technical executives with dagit! is there a workaround to get schedule and/or sensor working on Windows laptop? or better just to get a linux server or container?

Hello all, I'm having a bit of trouble getting a solid factory to work. The solid runs as expected when it is not encapsulated in a wrapper function, however I must not be passing the correct information to the decorator. If anyone could look at the examples below and let me know if it looks off. Thanks for your help. This is the factory, and does not get invoked by the pipeline.

def get_market_ohlc_factory(name, market):
    @solid(name=name, required_resource_keys={"api"})
    def _get_market_ohlc(context):

        ts_ohlc = context.resources.api.get_ohlc(market)

        return ts_ohlc

    return _get_market_ohlc

This runs as expected

@solid(required_resource_keys={"api"})
def _get_market_ohlc(context):

    market = 'BTC/USDT'
    
    ts_ohlc = context.resources.api.get_ohlc(market)

    return ts_ohlc

The purpose of the factory is to pass parameters form a list to create solids. The pipeline func looks like this. The

get_market_list()

returns a python list from a file.

@pipeline(
        mode_defs=[
            ModeDefinition(
                resource_defs={"postgres": postgres_resource,
                               "api": api_resource,
                               "redis": redis_resource})
        ])
def run_pipeline():
    market_list = get_market_list()
    for market in market_list:
      ts_ohlc = get_market_ohlc_factory(name=f'_get_market_ohlc', market=market)

Fun weekend with dagster and mlflow I try to integrate mlflow with dagster pipelines. I will publish an architecture of it. I am mixing: • mlflow + minio (S3) + postgres for registering the model • dagster with daemon launching specific mlflow image with a scheduler and amazing sensors (simple but the idea is here). If you have any improvements tell me 🙂 https://github.com/slamer59/dagster-mlflow.git

hey everyone! i'm trying to use dagster to poll another service of ours to kick off a pipeline when a pipeline in the other service completes. this seems like what sensors are designed for, but i'd like to set things up such that our production dagster only polls the production env of the other service, and likewise for our dev environment for dagster. sensors don't seem to support environment config, based on the docs. am i missing something, or is this just not what sensors are meant to do?

Hi there! Is there a way to reset the sensor to rerun all runs? I tried deleting all the runs on the dagit UI and now the sensor is no longer running at all for some reason.

Hello there! I’m trying to use

dagster-k8s/config

tag to apply tags to the k8s job that’s created by dagster I’m doing it like this:

@solid(
    tags={
      'dagster-k8s/config':{
        'pod_template_spec_metadata': {
          'annotations': { "<http://iam.amazonaws.com/role|iam.amazonaws.com/role>": "Dist"}
        },
      }
    },
    required_resource_keys={"data_lake", "dataset_name"},
    config_schema=String,
)

the created job is being created without the above tags. I downloaded the run’s debug log from the dagster UI and i see that the tags do appear in the debug log. just not on the k8s job. am i doing something wrong? thank you!

Hi, I wanted to ask two questions: 1. Is it possible to run a pipeline when another pipeline finishes it’s task or run? 2. Is it possible for a single solid in a pipeline to have a tag based run concurrency so that if the concurrency is set to 4 and 10 pipelines are running concurrently at a time, then at a point where all these 10 reaches that specific solid, 6 out of 10 waits for 4 solids of different instance of the pipeline to complete?

Hi there, a question to sensors where

run_key

isn’t a filename, but a list of filenames. Let’s say our pipeline will process a list of files rather than file-by-file, therefore we pass a list along. The sensor will check if there are new files and package them into a list of files around 500MB. Now the challenge, a file could be contained in different lists in case of re-processing. Is there a possible way the sensor could check if a “file in a list” has been processed? 👉 [more in thread]

hello, i have seen this issue: https://github.com/dagster-io/dagster/issues/3018 .. we would have more or less the same thing: we would like to have one dagster instance, leave it running and deploy new repo's without touching the dagster instance (created from helm). the approach mentioned in the issue (subcharts) would not allow for this, because this way the usercode deployments need to be known at installation time (look at the current code in the helm chart: it uses for-loops in the template). i think there should be a way (maybe even using a custom resource DagsterRepository) to deploy new repo's to a dagster instance

what's the behavior of sensors and schedules when execution gets stalled? like if you have a sensor running every 30 seconds and execution takes >30secs? i can't find this covered in the docs

👋 I have a scheduled pipeline which I cannot rerun after it failed.

xxxx_pipeline cannot be executed with the provided config.

If I copy the schedule config runs fine.

Is there a way to have dagit/dagster pickup logging from other python modules that are being output using standard python logger?

I'm using Google Cloud to run dagster daemon and dagit. For the moment, I'm running it inside a tmux to analyze how everything is running. In my pipeline, we're getting some JSON data and posting it to Big Query every minute. Usually, the whole process takes 4s. However, analyzing my CPU utilization on Google Monitoring it keeps increasing in a constant rate. What can we do to keep CPU down? My impression is that dagster is open more process without closing old ones or something like this.

Hi! Is there some way to stop pipeline execution, if on some step we decide this. f.e. no data in table.

@solid(output_defs=[OutputDefinition(Dict, 'data', is_required=False)])
def solid_1(context):
	#read data code
	if data:
		return dict()
	else:
		return None #how stop pipeline here?
	
@solid(input_defs=[InputDefinition('data', Dict)])
def solid_2(context,data):
	return dict()

Thanks!