is there a way to combine presets with e.g. the config generated from a schedule or partition, so you don't have to define secret envs in multiple places for the same pipeline?

Hey folks 👋 Does Dagster support partitions that have more than 1 dimension, or pipeline runs with multiple partitions? For example, I have a pipeline that pulls data from multiple endpoints of an API every day — therefore, hoping to partition that pipeline along “days” and “endpoints”. If for a given day, the pipeline fails for a given endpoint, it feels optimal to only re-run this specific day/endpoint, instead of re-running the pipeline for this given day on all endpoints.

hi all . anyone has the experience of comparing building a ML pipeline using serverless architecture vs using dagster. my company is using aws lambda extensively in ETL jobs but when coming to training ML models, aws lambda seem to be less attractive

Hi all 👋🏼 Anyone have experiences to share regarding benchmarking pipeline / solid run durations over time, and under different execution scenarios? E.g. sending through 1 item vs 10k items; 1 running pipeline on the system vs 5 running simultaneously; day 1 of the project vs day 30, etc. I know a lot of good per-run metrics are available via the API, but to get those trends and comparisons we think we'll need to aggregate in a separate system. Interested what other folks are doing for this 🙂

Hi everyone, I am attempting to pass a config schema into a solid that takes a input definition. However, whenever I pass the config schema in I get the following error:

Must pass the output from previous solid invocations or inputs to the composition function as inputs when invoking solids during composition.

Does anyone have any ideas on why this might be happening. Thanks 🙂 Code example

config = {
    "solids": {
        "solid_b": {
            "config": {'path': 'test.txt'}   
        }
    }
}

@pipeline
def test_pipeline():
  test_solid = solid_b(solid_a, config_schema=config)

Hi all, I’m hitting an error with dependencies and I was wondering if anyone knew a better way to dive deeper into debugging to figure out what dependency it is failing at? My solids are pretty simple:

@solid
def hello(_):
	return json.loads("{"msg":"hello"}")

@solid
def send_http_call(context, results):
	// Does an HTTP POST using the value from hello()

@pipeline
Def hello_pipeline():
	send_http_call(hello())

The actual error message is

Dependencies for hello_pipeline().compute failed

so trying to figure out how I can dive deeper to where .compute is occurring. Currently been reading up on the docs in case there’s a concept I missed

What is the best way to work with Spark `DataFrame`’s in

0.10.0

. In earlier versions we returned a DataFrame as in the airline demo (see also image attached). I understand from the docs that

PySpark DataFrames cannot be pickled, which means that IO Managers like the fs_io_manager won’t work for them.

If we use the

LocalParquetStore

as illustrated above, do we need to add this IO-Manager to Pipeline. Does it mean that other outputs in these pipeline must be parquets as well? Or can there be multiple? Or how does default pickle work along side with such LocalParquetStore? Haven’t gotten my hands on that, but my co-worker is struggling with it. Thought I will ask quickly for a guideline before he starts adding stuff to io_manager.py in dagster-aws. In my understanding was, that with io-manger not much to the solids need to be changed. But for DataFrames and Spark, we need to change our way we output data frames in our solids? Thanks for your help. Not sure if others have already update to 0.10 with spark?

Hey Dagsters and Early Adopters, Thank you for the awesome work - I am in the early stages of exploring Dagster, so far loving it but not without bumps 🙂 I am having trouble with properly importing pipelines from its respective modules when using Dagit and

workspace.yaml

with 2 pipelines under the same directory. Reading documentation was of no help since it covers only simple use cases. Here is the structure of the folder from where I run `dagit`:

- local_repo.py
- workspace.yaml
- dagster_baby_pipeline (folder with pipeline definition)
- dagster_try (folder with pipeline definition)

In

local_repo.py

I try to import

baby_pipeline.py

from

dagster_baby_pipeline

and

hello_cereal.py

from

dagster_try

to return from

@repository

function. From the logs I am suspecting that those module aren’t getting loaded into Dagster process hence the relative imports don’t work. If I specify

working_directory

for ONLY

dagster_baby_pipeline

folder in

workspace.yaml

and don’t import

dagster_try

then it works. It seems like if I create a repository for each pipeline in its respective modules and import them separately in

workspace.yaml

then it will achieve what I want but will require boilerplate overhead. Can you please clarify what is happening under the hood with imports in this scenario and what is the best practice on using these abstractions? Is there a way to load both

dagster_baby_pipeline

and

dagster_try

modules so they can be accessed by a single

local_repo.py

?

Owen Kephart just published a new version: 0.10.3. See https://github.com/dagster-io/dagster/releases/tag/0.10.3 for the release notes.

Hi all, another quick question. So I know it’s a configuration issue now but if I could get some tips on where to look next, that’d be great. Problem: I am able to get a list of repositories via GraphQL if I run it locally but when I run it as a docker container, it returns back an empty array. Things I’ve tried/confirmed: • Confirmed what I think is a valid working_directory and relative_path in workspace.yaml • I run both of the curl calls (from GraphQL playground) so nothing changes there. When local, I run it from terminal and when in docker container, i ssh into the container and run the same curl call • Workspace.yml is pretty simple:

load_from:
  - python_environment:
      executable_path: /usr/local/bin/python
      target:
        python_file:
          relative_path: /opt/dagster/dags/example/hello_dagster.py
          working_directory: /opt/dagster/dags

The curl call that I do is also pretty simple:

curl -v '<http://localhost:3000/graphql>' \
-H 'Accept-Encoding: gzip, deflate, br' \
-H 'Content-Type: application/json' \
-H 'Accept: application/json' \
-H 'Connection: keep-alive' \
-H 'DNT: 1' \
-H 'Origin: <http://localhost:3000>' \
--data-binary '{"query":"query RepositoriesQuery {\n  repositoriesOrError {\n    ... on RepositoryConnection {\n      nodes {\n        name\n        location {\n          name\n        }\n      }\n    }\n  }\n}"}' --compressed

Johann Miller just published a new version: 0.10.4. See https://github.com/dagster-io/dagster/releases/tag/0.10.4 for the release notes.

I wonder if I missed something in the docs... but I didn't find anything about how to coordinate the launch of multiple pipelines... e.g. I've Pipelines A and B and want to trigger Pipeline B right after Pipeline A is finished. How do you do this?

Good day all. How can we pass resources to sensors? E.g., I have a pipeline with a DB or API connection as a required resource, and I'm writing a sensor to target that pipeline. The sensor itself needs to poll that DB/API for some state change in order to decide whether to kick off the rest of the pipeline. I know the

SensorExecutionContext

has a

DagsterInstance

, but not clear how to get from there to any resource information?

Morning... Regarding

schedules

and

presets

- If i have a pipeline with one or multiple presents (say "dev" and "prod" related) and i'm using the UI, I can easily pick which preset I need. And I understand the function decorated by

@schedule

must return a

run config

but how do I get it to return a preset? (how to I get the pipeline to run a specific preset on a schedule)

hi again, does the dagster logger allow lazy logging? it doesn't like having extra args for string formatting in the way the normal python logger does e.g.

<http://log.info|log.info>("the next train is at %s", never)

hey all, i just updated to

0.10.4

(from 0.9.22post0) and am having issues installing

dagster

and

dagit

together using

docker

. for one, there doesnt seem to be an explicitly stateded requirement for

gevent

in

dagit

:

The 'gevent' distribution was not found and is required by dagit, gevent-websocket

I then seem to have to jump through an endless string of adding different dependencies or addressing version mismatches (most related to graphql). is there some reason why these dependency issues would occur in a docker env and not in my local dev env (Mac OS X)? My

requirements.txt

and

Dockerfile

are pasted below: requirements:

dagster>=0.10.4
dagit>=0.10.4
dagster-pandas>=0.10.4
dagster-slack>=0.10.4
dagster_postgres>=0.10.4
dagster_dbt>=0.10.4
pandas<=1.1.4
marketorestpython==0.5.8
pandasql==0.7.3 
python-dotenv==0.10.3
retry>=0.9.2
simple-salesforce==1.10.1
SQLAlchemy==1.3.20
PyMySQL>=1.0.0
pytz<2021.0
pyaml==20.4.0
psycopg2>=2.8.0
CensusData>=1.11.post1

Dockerfile:

FROM python:3.9.0-slim

RUN apt-get update -yqq && \
    apt-get install -yqq cron gcc git && \
    apt-get install -yqq libpq-dev && \
    apt-get install -yqq default-libmysqlclient-dev && \
    apt-get install libmariadbclient-dev && \
    apt-get clean && \
    rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/*

ENV DAGSTER_HOME=/opt/dagster/dagster_home/

COPY . /opt/dagster/dagster_home

WORKDIR /opt/dagster/dagster_home

RUN python3 setup.py install

EXPOSE 3000

ENTRYPOINT ["dagit", "-h", "0.0.0.0", "-p", "3000"]

Say if I have a daily pipeline that takes 10 hours to compute and one given task (solid A) takes 3 hours. But solid A's data only updates once a week. What mechanism does Dagster have to let the pipeline know to use solid A's generated data set for a week instead of regenerating the same data daily and thus wasting that 3 hours.

In theory you could use partial re-runs

But I would probably implement something more specific in the solid itself

Missing required config entry "solids" at the root. Hello, I am trying to learn dagster. My idea at the end is to couple mlflow with dagster. First, I want to have a "generic workflow" with one branch for different models. The best would be to make those branch dynamically created. I use multiple output (split train/test). My idea is to feed ml_model function with a dict containing a configuration. Dict is here: https://github.com/slamer59/dagster-mlflow/blob/fdbe6ba7f4e9352cfae3e749dcf861a47cfbd8b5/classify_wines.py#L92 Here ml_model https://github.com/slamer59/dagster-mlflow/blob/fdbe6ba7f4e9352cfae3e749dcf861a47cfbd8b5/classify_wines.py#L129 Why i am having this error ? How to pass dict as input ? Thanks 🙂

Accessing the demo inside the video ? Hello, a video have been played inside the demo. A nice feature better do to double compressoin we cannot see the code 🙂 Is it possible to access it somewhere ? @schrockn? Really interesting ! That is really nice demo.

I am new to Dagster and trying to determine if it is a good fit. I would like to port in our data engineering pipeline that I run in a single ScriptRun in AzureML where I generate all potential features prior to training a model and then later filter to a specific feature set for each model. I’m hoping to filter the DAG built in dagster in our deployed model to only run the parts of the pipeline we need to generate the features we used in the model. Can dagster be used in reverse to go from leaf columns in a dataframe back to a set of pipeline sets and inputs required?

Can someone please help in understanding

composite_solid

abstraction? I am getting

UserWarning: Error loading repository location local_repo.py:dagster.core.errors.DagsterUserCodeProcessError: AttributeError: 'InputMappingNode' object has no attribute 'startswith'

when passing a value to

composite_solid

that will then based on that value will determine which

solid

to execute. Code is attached in the thread to spare the channel.

Hi, I am getting the following error while using DynamicOutput

dagster.core.errors.DagsterSubprocessError: dagster.check.CheckError: Member of list mismatches type. Expected <class 'dagster.core.execution.plan.inputs.StepInput'>. Got UnresolvedStepInput(name='region', dagster_type_key='String', source=FromPendingDynamicStepOutput(step_output_handle=StepOutputHandle(step_key='get_regions', output_name='result', mapping_key=None), solid_handle=SolidHandle(name='execute_per_region', parent=None), input_name='region')) of type <class 'dagster.core.execution.plan.inputs.UnresolvedStepInput'>.

Hi @everyone, I am getting following sort of bug in dagster UI: So basically I have a solid that reads a specific file from S3 and returns the number of rows in that file which is working properly as expected however this solid only takes around 1 to 2 seconds max but in UI, it keeps on running forever, even its status gets

SUCCESS

but that bar for solid keeps on moving to infinite time. Here is screencast of the Bug: https://www.loom.com/share/b4f6b031cbea43c1893e45a552f917aa Any Help or suggestions would be appreciated🙂👍

Hey folks 👋 Is there a good pattern in Dagster to avoid running solids when the output of a previous solid is

None

or an empty list? I have an

extract

solid that pulls data from a REST API as a list of dicts, which I take as input in a

to_df

solid to format as a dataframe. The API is not guaranteed to return results for a given partition, which then causes issues downstream with my IO manager and Dagster’s type validation. I could address that in the

to_df

solid, but curious if there was a better pattern to use — some kind of conditional execution logic in the pipeline definition?

Guys, just checking out the status of MySQL support... I was selling Dagster really hard to my team months ago and the team has decided to use it... however, the MySQL support is the must have.. Now I’m in the middle of weird situation: everyone is excited about Dagster I sold to them, but we can’t do anything about it, and they are giving me pressure to re-consider Airflow again..

I know we have an open PR here: https://dagster.phacility.com/D5710 Is there any update?