@prha will let you take over

meeting 🙂

Thanks!

typical pattern is to name it "on_table_create" or something like that

and you can omit the argument

I found the docs at: https://docs.dagster.io/latest/api/apidocs/types/#dagster.Nothing this should be what I need I think? Thanks so much!

Hello, I'm trying to run a specific step of a pipeline using

execute_pipeline

with

step_keys_to_execute

defined in the

RunConfig

however, I'm running into this error:

dagster.core.errors.DagsterStepOutputNotFoundError: When executing transform.compute discovered required outputs missing from previous step: [StepOutputHandle(step_key='load.copy_django_logs_to_redshift.compute', output_name='result')]

I thought maybe I could fix it by passing in

rexecution_config

to RunConfig, but when I do I get this error:

TypeError: __new__() got an unexpected keyword argument 'rexecution_config'

Could someone give me pointers on how to run a specific pipeline step that depends on previous steps? Thanks so much really appreciate all the help.

Is there a way to generate solids in a loop, with aliases, but pass them all the same config, without having to redefine the config more than once in the yaml config? Let me know if my question makes sense or not. Thanks!

Like somehow define the config for the generic solid, not for the aliased names?

Hello 👋 I've deployed a dagster environment on k8s ( Daggit / Rabbitmq / Dagster-celery ) 🎉 I'm able to launch Pipeline using Celery, and the worker do their work, and execute the solid they were ask to execute. I'm trying to configure k8s-launcher. It's working great so far, but, I was wondering if there is any way to start solids in parallel. So far, when I start my pipeline, even if the solid can be executed in parallel, the job only run one solid at a time. Is there anyway to configure that ? Thanks for your great work ! dagsir

Hey, another related k8s question. When I start my deployments, dagster-celery worker take about 200 MiB of RAM. When a job get executed on the worker, the memory get used (this is normal as a process is running on the worker). But, when the worker is done with the process, the memory never goes back to its original value. I'm trying to investigate what happen after the end of the process. It's pretty annoying, because, each time a new "job" has to run on the worker, the RAM increase a little, and, at some point, kubernete kill the worker because there is not enough Memory Have you ever got something similar to it ?

hey! i notice that in the tutorial for materialization, files are just saved to the current working directory which is

/opt/dagster/app

if you also follow the docker tutorial. i also noticed that in an older version of one of the tutorials if i'm not mistaken, the example was to create an

/opt/dagster/data

directory. what is the best practice for storing data / intermediate files on the filesystem? assuming s3 or something like that is the best practice, but if i have to use a local dir

also, what is the best way to use environment variables for dagster.yaml inputs - since i don't want to check postgres credentials into git. would you recommend using sed to replace strings during docker builds?

(https://docs.dagster.io/latest/deploying/aws)

Hiya I'm trying to follow along with the airline demo using dagster 0.7.5 but I can't seem to find the execute tab, what am I doing wrong??

Hello Guys, I am having this error, when trying to parameterize a solid on dagit:

Undefined field "inputs" at path root:solids:augment_5 Expected: "{ outputs?: [{ result?: { json: { path: Path } pickle: { path: Path } } }] }"

Details posted in the Thread.

Hi everyone! I’m new to this slack, so please forward me to proper chan if this isn’t the appropriate one to ask this question. I’m trying to link two pipelines via scheduling. For example, imagine, my two pipelines have two different cron cycles, pipeline A runs once a month and pipeline B runs every day. But pipeline B depends on last run of pipeline A. So, on the day A runs, I would like to be able to run pipeline B only if A has finished and succeeded, and postpone any subsequent run of B until A is fixed (if it happens on the weekend for instance). Is this something that is currently possible with current scheduler?

Hello - believe I've read all the documentation and searched all of the threads resembling this issue. Getting an error I can't fix ...in the toy problem 😕 Any advice on

Error 1: Missing required field "solids" at document config root. Available Fields: "['execution', 'loggers', 'resources', 'solids', 'storage']".

Still on serial_pipeline.py, Windows 10, Anaconda, pip installed

@pipeline

def serial_pipeline():

    

sort_by_calories(load_cereals())

def test_serial_pipeline():

    

res = execute_pipeline(serial_pipeline, environment_dict=ed)

    

assert res.success

if __name__ == '__main__':

    

result = execute_pipeline(serial_pipeline)

    

assert result.success

hi… i want to build the pipeline dynamically with the collection of solids… is it possible using yaml config?

Hi, I'm looking for a way to parameterise a pipeline, so I can namespace all the outputs of various jobs in the DAG. For instance, ideally I'd like to: 1. Grab a username from the initiating user, if they are kicking it off in the Dagster UI for instance. 2. Use that to programmatically generate a database name, which would include the username. 3. Set that name in the pipeline context somewhere programmatically. 4. Use that database name in different solids in the pipeline, where one solid might delete/create the database, and another solid runs a job loading the database. Is there anyway to achieve this? Apologies, if I have missed something obvious in the docs.

Has anyone had problems not being able to reconnect the dagit UI to the dagit server after restarting the dagit server? I'm using latest chromium snap on Ubuntu 19.10. dagit always connects fine the first time. but if i have to restart the dagit server, no amount of cookie/cache clearing and refreshing will make it connect. i can open a new incognito window and it connects just fine right away. Sometimes after some random time like 5 minutes has passed the main chromium window can connect again.

We are having problem running

dagster

with

pyspark

but succeed to run

dagit

with the exact same pipeline. The below code succeeds with dagit:

(conda_env) myuser@myserver:~/projects/formatter> dagit -f dagster_pyspark/compsite_solids_types.py -n pyspark_pipeline -p 3001
Loading repository...
Serving on <http://127.0.0.1:3001> in process 73783
2020-04-01 20:45:35 - dagster - DEBUG - pyspark_pipeline - c56443eb-fd20-4d61-863a-7af6a1bfc4a4 - ENGINE_EVENT - Starting initialization of resources [spark].
 event_specific_data = {"error": null, "marker_end": null, "marker_start": "resources", "metadata_entries": []}
WARNING: User-defined SPARK_HOME (/opt/cloudera/parcels/SPARK2-2.3.0.cloudera3-1.cdh5.13.3.p0.458809/lib/spark2) overrides detected (/opt/cloudera/parcels/SPARK2/lib/spark2/).
WARNING: Running spark-class from user-defined location.
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
2020-04-01 20:45:45 - dagster - DEBUG - pyspark_pipeline - c56443eb-fd20-4d61-863a-7af6a1bfc4a4 - ENGINE_EVENT - Finished initialization of resources [spark].
 event_specific_data = {"error": null, "marker_end": "resources", "marker_start": null, "metadata_entries": [["spark", "Initialized in 10.05s", ["dagster_pyspark.resources", "SystemPySparkResource"]]]}
2020-04-01 20:45:45 - dagster - DEBUG - pyspark_pipeline - c56443eb-fd20-4d61-863a-7af6a1bfc4a4 - PIPELINE_START - Started execution of pipeline "pyspark_pipeline".

But when running the same pipeline with

dagster

the

pyspark

initialising fails . (conda_env) myuser@myserver:~/projects/formatter> dagster pipelione execute -f dagster_pyspark/compsite_solids_types.py -n pyspark_pipeline -e dagster_pyspark/test_pyspark.yaml

> dagster - ERROR - pyspark_pipeline - 1d12e4e7-15e0-49f8-a394-27dcf29ef9cb - PIPELINE_INIT_FAILURE - Pipeline failure during initialization of pipeline "pyspark_pipeline". This may be due to a failure in initializing a resource or logger.
 event_specific_data = {"error": ["py4j.protocol.Py4JError: An error occurred while calling <http://None.org|None.org>.apache.spark.api.python.PythonAccumulatorV2. Trace:\npy4j.Py4JException: Constructor org.apache.spark.api.python.PythonAccumulatorV2([class java.lang.String, class java.lang.Integer, class java.lang.String]) does not exist\n\tat py4j.reflection.ReflectionEngine.getConstructor(ReflectionEngine.java:179)\n\tat py4j.reflection.ReflectionEngine.getConstructor(ReflectionEngine.java:196)\n\tat py4j.Gateway.invoke(Gateway.java:237)\n\tat py4j.commands.ConstructorCommand.invokeConstructor(ConstructorCommand.java:80)\n\tat py4j.commands.ConstructorCommand.execute(ConstructorCommand.java:69)\n\tat py4j.GatewayConnection.run(GatewayConnection.java:238)\n\tat java.lang.Thread.run(Thread.java:748)\n\n\n", ["  File \"/opt/anaconda3/envs/dask_pyarrow/lib/python3.7/site-packages/dagster/core/errors.py\", line 161, in user_code_error_boundary\n    yield\n", "  File \"/opt/anaconda3/envs/dask_pyarrow/lib/python3.7/site-packages/dagster/core/execution/resources_init.py\", line 138, in single_resource_event_generator\n    resource = next(gen)\n", "  File \"/opt/anaconda3/envs/dask_pyarrow/lib/python3.7/site-packages/dagster_pyspark/resources.py\", line 66, in pyspark_resource\n    pyspark = SystemPySparkResource(init_context.resource_config['spark_conf'])\n", "  File \"/opt/anaconda3/envs/dask_pyarrow/lib/python3.7/site-packages/dagster_pyspark/resources.py\", line 30, in __init__\n    self._spark_session = spark_session_from_config(spark_conf)\n", "  File \"/opt/anaconda3/envs/dask_pyarrow/lib/python3.7/site-packages/dagster_pyspark/resources.py\", line 25, in spark_session_from_config\n    return builder.getOrCreate()\n", "  File \"/opt/anaconda3/envs/dask_pyarrow/lib/python3.7/site-packages/pyspark/sql/session.py\", line 173, in getOrCreate\n    sc = SparkContext.getOrCreate(sparkConf)\n", "  File \"/opt/anaconda3/envs/dask_pyarrow/lib/python3.7/site-packages/pyspark/context.py\", line 367, in getOrCreate\n    SparkContext(conf=conf or SparkConf())\n", "  File \"/opt/anaconda3/envs/dask_pyarrow/lib/python3.7/site-packages/pyspark/context.py\", line 136, in __init__\n    conf, jsc, profiler_cls)\n", "  File \"/opt/anaconda3/envs/dask_pyarrow/lib/python3.7/site-packages/pyspark/context.py\", line 207, in _do_init\n    self._javaAccumulator = self._jvm.PythonAccumulatorV2(host, port, auth_token)\n", "  File \"/opt/anaconda3/envs/dask_pyarrow/lib/python3.7/site-packages/py4j/java_gateway.py\", line 1525, in __call__\n    answer, self._gateway_client, None, self._fqn)\n", "  File \"/opt/anaconda3/envs/dask_pyarrow/lib/python3.7/site-packages/py4j/protocol.py\", line 332, in get_return_value\n    format(target_id, \".\", name, value))\n"], "Py4JError", null]}

We tried to play with

SPARK_HOME

and tried to use findspark in the

dagster_pyspark/resources.py

file but had no success . Any suggestions?

any good blog posts about rabbitmq in docker vs sqs for dagster?

i'm getting

Undefined field "s3" at path root:storage Expected

while trying to set up s3 for intermediate storage: running solid with this config:

storage:
  s3:
    config:
      s3_bucket: <my bucket name>

following the aws deployment docs https://dagster.readthedocs.io/en/stable/sections/deploying/aws.html#using-s3-for-intermediates-storage maybe i missed something?

waitttt ignore, i found it

system_storage_defs=s3_plus_default_storage_defs,

hi… i want to build the pipeline dynamically with the collection of solids… is it possible using yaml config?

Hi, I'm running into an issue in which I've created a dependency structure between a few composite_solids by passing Nothing between them (since I don't have real values to pass between them), however, I expected the dependency structure to be enforced for their child solids as well, but this doesn't seem to be the case. For instance, I have two composite_solids. I expect all the child solids within my first composite_solid to finish before any of the child solids in my second composite_solid begin. However, the dependency structure is only being enforced for the child solid who's return value is returned by the composite. The other child solids in my second composite_solid start at the same time as the child solids in my first composite_solid. Please let me know if what I've described makes sense sorry if that is a bit verbose/confusing. Thanks so much for any help!

Hi, is there a way to change/configure the Python venv that a whole pipeline and individual solids are run under?