that's very clever!

I'm getting an error running a solid on airflow that I don't get when running using dagit. This specific solid fails the input type check where the type is a list of custom dagster types.

Hi, I've been checking out pipeline frameworks. How is Dagster different to Kedro (https://github.com/quantumblacklabs/kedro/)?

Have you seen https://www.digdag.io/ ? A decent amount of overlap with dagster

Max Gasner just published a new version: 0.5.9.

@Amanda Crawford ^ this should include the fix for the issue you were facing earlier

The amount of DAG tools is crazy now

Someone should write a tool that helps you select DAG tools using DAGs 😄

🙂

QQ: Is it possible to have resources that depend on other resources? Like with

required_resource_keys

for solids

👋 Wanted to get your all's take on file cleanup/dealing with tempfiles in dagster. A toy example I am playing with is downloading a csv (let's say the iris dataset from UCI) converting the csv into a multiline json file which I want to eventually load into a table in a sqlite db. I have everything working, but I am struggling with composing my pipeline in a way that deletes files once I am done with them. For example, I would like to delete my csv file once I transform everything to a json blob to not waste space. Here is one way I composed the DAG (I am excluding the load to sqlite solid for simplicity)

@pipeline
def iris_ingestion_pipeline():
    convert_csv_to_json(
        download_csv_from_url_to_file()
    )

Currently, I am deleting the csv file in my

convert_csv_to_json

solid just to get it working. This is not ideal though because I am coupling transformation and IO logic. One option I got working was having

convert_csv_to_json

return the file to be deleted which would then get passed into a

delete_file

solid like so:

@pipeline
def iris_ingestion_pipeline():
	delete_file(
	    convert_csv_to_json(
	        download_csv_from_url_to_file()
	    )
	)

This is also not ideal because the semantics of the convert_csv_to_json become brittle if you plan on reusing that solid in contexts where you don't want to delete the file I then tried to go the tempfile route, but the issue here is that I have no way of persisting a tempfile_fp across solids unless I hack the execution context. It is likely I am missing something obvious here with inputs but would love to get A: Your opinions for how to do this? Just so I have a better intuition around how solids are to be composed B: Opinions on support for tempfiles and tmpdir's?

Has there ever been a follow-up to https://github.com/dagster-io/dagster/pull/699? What's your current view on the issue (fanning in and out of subgraphs / handling Sequence[T])?

hey everyone! Quick question: how can I reuse output of the solid, should I just put it twice in the pipeline? Will it be calculated twice or not?

I also catch

ClientError: An error occurred (403) when calling the HeadObject operation: Forbidden

when I try to download a file from private bucket, I have creds in environment as usual. I can download it manually via

import boto3
s3 = boto3.resource("s3")
s3.Bucket(bucket_name).download_file

though

I don't know if this is the appropriate place to ask but is there a simple guide on how I can actually run a unit test using pytest on a solid? I have tried using the code examples in the GitHub repo but always end up with

solid_name = 'generate_date_suffix'

    def assert_in_composition(solid_name):
        if len(_composition_stack) < 1:
            raise DagsterInvariantViolationError(
                'Attempted to call solid "{solid_name}" outside of a composition function. '
                'Calling solids is only valid in a function decorated with '
>               '@pipeline or @composite_solid.'.format(solid_name=solid_name)
            )
E           dagster.core.errors.DagsterInvariantViolationError: Attempted to call solid "generate_date_suffix" outside of a composition function. Calling solids is only valid in a function decorated with @pipeline or @composite_solid.

it's called using pytest

def test_generate_date_suffix():
    execute_solid(generate_date_suffix("test"))
    assert True

execute_solid(generate_date_suffix)

should do what you want -- you'll want to pass the input in the

input_values

param to that function

is

input_values

required if the input is blank?

nope, then

execute_solid(generate_date_suffix)

should do

i see

ok, let me try a few more variations to see what happens

thanks for you help!

This is the stack trace I’m getting when I try to run the execution of the ingest pipeline in the airline demo:

An exception was thrown during execution that is likely a framework error, rather than an error in user code.
Original error message: dagster.core.errors.DagsterResourceFunctionError: Error executing resource_fn on ResourceDefinition db_info

Stack Trace:
  File "/Users/johnstory/.local/share/virtualenvs/dagster_airline_demo-WDH392CN/lib/python3.7/site-packages/dagster_graphql/implementation/pipeline_execution_manager.py", line 299, in execute_pipeline_through_queue
    DagsterInstance.from_ref(instance_ref),

  File "/Users/johnstory/.local/share/virtualenvs/dagster_airline_demo-WDH392CN/lib/python3.7/site-packages/dagster/core/execution/api.py", line 133, in execute_run_iterator
    pipeline, pipeline_run.environment_dict, run_config, instance

  File "/usr/local/Cellar/python/3.7.3/Frameworks/Python.framework/Versions/3.7/lib/python3.7/contextlib.py", line 112, in __enter__
    return next(self.gen)

  File "/Users/johnstory/.local/share/virtualenvs/dagster_airline_demo-WDH392CN/lib/python3.7/site-packages/dagster/core/execution/context_creation_pipeline.py", line 204, in scoped_pipeline_context
    raise dagster_error

  File "/Users/johnstory/.local/share/virtualenvs/dagster_airline_demo-WDH392CN/lib/python3.7/site-packages/dagster/core/execution/context_creation_pipeline.py", line 179, in scoped_pipeline_context
    log_manager,

  File "/usr/local/Cellar/python/3.7.3/Frameworks/Python.framework/Versions/3.7/lib/python3.7/contextlib.py", line 112, in __enter__
    return next(self.gen)

  File "/Users/johnstory/.local/share/virtualenvs/dagster_airline_demo-WDH392CN/lib/python3.7/site-packages/dagster/core/execution/context_creation_pipeline.py", line 63, in create_resource_builder
    yield resources_stack.create()

  File "/Users/johnstory/.local/share/virtualenvs/dagster_airline_demo-WDH392CN/lib/python3.7/site-packages/dagster/core/execution/context_creation_pipeline.py", line 346, in create
    user_fn, DagsterResourceFunctionError, _create_msg_fn(resource_name)

  File "/Users/johnstory/.local/share/virtualenvs/dagster_airline_demo-WDH392CN/lib/python3.7/site-packages/contextlib2.py", line 442, in enter_context
    result = _cm_type.__enter__(cm)

  File "/usr/local/Cellar/python/3.7.3/Frameworks/Python.framework/Versions/3.7/lib/python3.7/contextlib.py", line 112, in __enter__
    return next(self.gen)

  File "/Users/johnstory/.local/share/virtualenvs/dagster_airline_demo-WDH392CN/lib/python3.7/site-packages/dagster/core/execution/context_creation_pipeline.py", line 462, in user_code_context_manager
    check.invariant(stopped, 'Must yield one item. Yielded more than one item')

  File "/usr/local/Cellar/python/3.7.3/Frameworks/Python.framework/Versions/3.7/lib/python3.7/contextlib.py", line 130, in __exit__
    self.gen.throw(type, value, traceback)

  File "/Users/johnstory/.local/share/virtualenvs/dagster_airline_demo-WDH392CN/lib/python3.7/site-packages/dagster/core/errors.py", line 112, in user_code_error_boundary
    error_cls(msg_fn(), user_exception=e, original_exc_info=sys.exc_info(), **kwargs), e

  File "/Users/johnstory/.local/share/virtualenvs/dagster_airline_demo-WDH392CN/lib/python3.7/site-packages/future/utils/__init__.py", line 398, in raise_from
    exec(execstr, myglobals, mylocals)

  File "<string>", line 1, in <module>

Anyone have a clue? cc @nate

Is it a good idea to write something to the database in each solid? I have feature extraction pipeline, a depends from b and c. I want to calculate b and write it to db, calculate c and write it to db, query all changed items in db and calculate a. Or should I calculate b and c, then query missing data to complement it (ie b was calculated for 10 items but c for 22, still have to query this 12 missing)?

Another Q: Was trying to install Dagster on another machine, followed the guide on the site, got the following message after

pip install dagster dagit

ERROR: Command errored out with exit status 1:
     command: /usr/bin/python -c 'import sys, setuptools, tokenize; sys.argv[0] = '"'"'/private/var/folders/m_/wf6sq5_d3xx7x1gd5b2xpfwm0000gn/T/pip-install-191_6L/dagster/setup.py'"'"'; __file__='"'"'/private/var/folders/m_/wf6sq5_d3xx7x1gd5b2xpfwm0000gn/T/pip-install-191_6L/dagster/setup.py'"'"';f=getattr(tokenize, '"'"'open'"'"', open)(__file__);code=f.read().replace('"'"'\r\n'"'"', '"'"'\n'"'"');f.close();exec(compile(code, __file__, '"'"'exec'"'"'))' egg_info --egg-base pip-egg-info
         cwd: /private/var/folders/m_/wf6sq5_d3xx7x1gd5b2xpfwm0000gn/T/pip-install-191_6L/dagster/
    Complete output (3 lines):
    /System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/distutils/dist.py:267: UserWarning: Unknown distribution option: 'long_description_content_type'
      warnings.warn(msg)
    error in dagster setup command: 'install_requires' must be a string or list of strings containing valid project/version requirement specifiers; Expected ',' or end-of-list in pathlib2>=2.3.4; python_version<"3" at ; python_version<"3"
    ----------------------------------------
ERROR: Command errored out with exit status 1: python setup.py egg_info Check the logs for full command output.

What am I doing wrong here?

Max Gasner just published a new version: 0.6.0.

👋 Hey <!channel>! Dagster 0.6.0 -- Impossible Princess 👸 We just pushed up a major new release of Dagster code-named Impossible Princess. We’ve been heads down working on this for the last few months, incorporating feedback from folks in this channel. The feedback was pretty consistent: Dagit is a really high quality frontend, the programming model has a lot of promise, and the focus on local development and testing is great. However, it was also very clear that you wanted to use Dagit as a monitoring and debugging tool for production pipelines, and you wanted an out-of-box, cloud-friendly scheduling and execution. That’s exactly what we built for 0.6.0. There’s lots of new stuff but the big-ticket items are: - A host-able version of Dagit, usable for production workflows. Before Dagit was a local development tool; now the same tool can also be used for monitoring and operating production workloads. You can view live-updating events, logs, and summary information from your production runs. - A native scheduler. You can define schedules using a Python API, deploy them server-side, and then have them execute Dagster jobs. This is a lightweight wrapper over cron, an extremely reliable and proven scheduler. - A fully supported Dagster-native, single-node, multi-process execution engine. - Devops tools for easily deploying Dagster on aws. We added a command line tool to the dagster-aws module for spinning up a Dagit node and all the supporting infrastructure—security group, RDS PostgreSQL instance, etc.—without having to touch the AWS console, and for deploying your code to that instance. It is still the same pluggable platform and system that you know. You can still deploy it to any infrastructure (not just AWS), execute Dagster pipelines on other orchestrators such as Airflow and Dask, and so forth. But we also wanted to provide an incredibly easy spin up process for those looking to experiment or use it for greenfield projects. Please try it out and give us feedback! We’ll be posting to the broader world next week with more extensive tutorials and screencasts, but in the meantime please ask us questions let us know what you think!

Nate Kupp just published a new version: 0.6.0.post0.

Congrats with new release! Have a question: can I pass the name of the alias as a param into solid? I have lots of sql queries, don't want to add them all into config and them just read sql file with the same name as the alias (ie sql_to_dict.alias("changed_addresses") will read changed_addresses.sql)

What is the correct way of handling inputs of composite solids? I want to use the inputs multiple times inside the composite solid. The example shown works fine if i comment out the “added = add(one, two)” line. How can I use the inputs more than one time correctly? I will add a picture to show how the pipeline will be build from the sample code in dagit. The examples in the docs and in the github repo do not seem to encounter this specific problem.