Hi! I'm trying to get a solid to start a new run of its own pipeline, asynchronously, in a recursive fashion. My code:

@solid
def my_solid(context, num: int):

    <http://context.log.info|context.log.info>(f'num = {num}')

    execute_pipeline_iterator(my_pipeline, instance=DagsterInstance.get(), environment_dict={
        'solids': {
            'my_solid': {
                'inputs': {
                    'num': num + 1
                }
            }
        }
    })

@pipeline
def my_pipeline():
    my_solid()

The result is that a new pipeline run is listed in dagit but is stuck on state

Starting...

- am I doing something wrong?

Hi guys! Thanks for your work on dagster, everything looks really cool 👍 Me and @matas are trying to deploy dagster at bestplace.ai to facilitate pipeline creation and execution for our analysts. Analysts don't like to write python code, they want to create pipelines described in yaml configuration files 😁 So we want to build system around dagster to make this possible. We deployed dagster using docker and use such containers: - dagster-master with dagit process; - several dagster-worker containers with dagster-celery executor; - rabbitmq container for communication between master and workers; - postgres and minio (s3) containers for persistent storage. Dagster-master and dagster workers use such directory structure: solids/ pipelines/ repository.py repository.yaml solids - directory with solids library, i.e. solids written in python (in future we will add jupyter solids); pipelines - directory with yaml pipelines, that use solids from solids library; repository.py contains define_repo() function that reads all yaml files and creates pipelines from them (I used code similar to this: https://github.com/dagster-io/dagster/blob/master/examples/dagster_examples/dep_dsl/pipeline.py); repository.yaml refers to define_repo() in repository.py. Everything works mostly fine but we encountered several problems and have questions that we would like to discuss with you. 1. We want users to add their own yaml pipelines to dagit. Then everyone should be able to explore and execute these pipelines. So what is best way to reload yaml pipelines to dagit? As I understand pipelines are reloaded when user clicks reload button in dagit UI. Is this the best way possible? What is happenning after click on reload button? Will define_repo() function be called as it is called at dagit start? 2. What happens to already running pipelines when user clicks reload button in dagit UI? What if user starts execution of the pipeline and then immediately clicks reload? I tried to do this and it seemed that pipeline is stuck in infinitely running state. I attached screenshots with this problem. 3. Is it possible to inform user about error in his pipeline.yaml through dagit interface? I can catch exceptions that are raised during yaml pipelines creation but how can I show them in dagit? Thanks

I'm still struggling through using a composite solid for wrapping a bash solid. Are there any non-trivial examples of this that anyone can share?

I seem to be cycling through the same set of errors as I iterate on this.

Looking through the airline demo now...

So, my current struggle is that the output that I'm yielding from a solid that is intended to be the full command string for the bash solid isn't treated as a string at pipeline execution time https://github.com/mitodl/ol-data-pipelines/blob/master/ol_data_pipelines/edx/solids.py#L565-L566

I'm seeing this when I try to launch dagit

Param "bash_command" is not a str. Got <dagster.core.definitions.composition.InvokedSolidOutputHandle object at 0x7f7a69f208e0> which is type <class 'dagster.core.definitions.composition.InvokedSolidOutputHandle'>

how is the pickling/unpickling of internal dagster objects supposed to be handled? i'm following the code in the dagster-aws

emr

module to see how it serializes and stores

DagsterEvent

s from a pipeline executed remotely (trying to create a similar Databricks integration), but it looks like events aren't rehydrated properly?

@>>> event
DagsterEvent(event_type_value='STEP_START', pipeline_name='pyspark_pipe', step_key='make_df_solid.compute', solid_handle=SolidHandle(name='make_df_solid', parent=None), step_kind_value='COMPUTE', logging_tags={'pipeline': 'pyspark_pipe', 'step_key': 'make_df_solid.compute', 'solid': 'make_df_solid', 'solid_definition': 'make_df_solid'}, event_specific_data=None, message='Started execution of step "make_df_solid.compute".')
@>>> pickle.loads(pickle.dumps(event))
_DagsterEvent(event_type_value='STEP_START', pipeline_name='pyspark_pipe', step_key='make_df_solid.compute', solid_handle=_SolidHandle(name='make_df_solid', parent=None), step_kind_value='COMPUTE', logging_tags={'pipeline': 'pyspark_pipe', 'step_key': 'make_df_solid.compute', 'solid': 'make_df_solid', 'solid_definition': 'make_df_solid'}, event_specific_data=None, message='Started execution of step "make_df_solid.compute".')

Hi, @nate Following issue, i think that using 

Permissive()

 is the best solution. 1. Where do you think is the right place for 

client_type

 in 

environment_dict

  configuration? 2. How do you suggest to handle the imports for all client_types, so that if a user want to use one type of cluster he does not need to install all of them? 3. What else do we need to handle?

Hi We are want to run a 

spark UDF function

 on a 

yarn

 cluster, but having trouble configuring the 

spark resource

. Since we are running on anaconda we are trying to use the instructions from conda to config spark. When running outside of 

dagster

 we can use the 

spark.yarn.dist.archives

 configuration pointing to a hdfs folder that is accessible to all the workers. We tried to define the sparksubmitpyFiles resource config but we are still getting the error:

 Cannot run program "path/to/python/env/bin/python3.7" error=2, No such file or directory.

Additionally we tried sparkdriverextraLibararyPath  but still getting the same error. What would be the best way to try and implement the 

PYSPARK_PYTHON

 and 

archives

 variables? is it through parse_spark_configs.py ?

i'm having a bit more fun with external step launchers today 🙂 my current issue is that i need some secrets as part of config in order to launch the step, but they're not necessary when it's running remotely. for example, my resource config looks something like...

@resource(
    {
        'databricks_host': Field(
            StringSource,
            is_required=True,
            description='Databricks host, e.g. <http://uksouth.azuredatabricks.com|uksouth.azuredatabricks.com>',
        ),
        'databricks_token': Field(
            StringSource, is_required=True, description='Databricks access token',
        ),
    }
)
def databricks_pyspark_step_launcher(context):
    ...

then i use something like this when running it:

resources:
  pyspark_step_launcher:
    config:
      databricks_host: <http://uksouth.azuredatabricks.net|uksouth.azuredatabricks.net>
      databricks_token:
        env: DATABRICKS_TOKEN

which launches the job just fine, but when it gets to executing the step remotely the

DATABRICKS_TOKEN

env var isn't present (and doesn't need to be) so the step fails. is there an idiomatic solution to this? maybe passing credentials in some other way a'la boto?

hey, random question/feedback about docs - there are a few places in the documentation where it says "update your config yaml" but it doesn't say which config yaml, and it also doesn't say which section: https://dagster.readthedocs.io/en/0.6.6/sections/learn/guides/logging/logging.html#configuring-the-built-in-loggers i assumed it meant

dagster.yaml

, but i get

Undefined field "loggers" at the root. Expected:

... so i'm not sure where to put this

i'm having an issue with solids getting stuck waiting in queue - my whole dagster cluster is stuck atm, anyone seen this happen before?

What's the recommended way to access

SystemComputeExecutionContext

within a composite solid? E.g., for logging

Question. I sometimes terminate solids that are taking a really long time so that I can debug what they are stuck on. Is there a way to resume the pipeline from the terminated step? When I try re-executing it lets me start from the "failure" (even though I terminated and there was no failure), but that just started it from the beginning of the pipeline again.

Hi, great work on dagster, I am really excited to see where this project goes. I do have a question... Is it possible to organise multiple repositories within one dagster instance? For example... I want all pipelines relating to 'car data' in one repo, and one relating to 'van data' in another repo within one dagit ui instance. Possible? Cheers.

is possible to automatize R scripts, maybe using dagster-bash?

As I understand it, Dagster does not (yet?) have support for a streaming pipeline. Nevertheless, i'm looking to deploy dagster in a production environment where I need to process inputs as they arrive. I can think of two ways of doing this. One is to have a scheduled pipeline that runs periodically to process all new inputs. The other is to have another python script running as a daemon that listens for new inputs and invokes the dagster function to execute a pipeline as inputs arrive. Does anyone have experience with either of these approaches or have a suggestion of their own? I think Dagster's great and I'd like to use it in production if possible.

Hi, I'm wondering how to stub an input to the first solid in my pipeline. The docs say this is possible here but I can't figure out how. I tried passing a dummy instance of the input type (a custom

DagsterType

) in the pipeline definition, but got a

DagsterInvalidDefinitionError

saying

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

What is the composition function in this context?

Hi, can I have solid dependencies without having to pass the result as a parameter? For example, I want solid_b to execute after solid_a if solid_a succeeds. Solid_a doesn't return anything and solid_b doesn't need anything from solid_a.

Hey, is there any update on https://github.com/dagster-io/dagster/pull/2469 ? I'd appreciate if someone could give me a clue where to look next

Hi everyone, my name is Fran and I'm new to dagster. I have experience with other Workflow frameworks like Airflow or Argo. At my company currently we have a custom framework which pretty much maps 1:1 to the dagster concepts, so currently I'm exploring how feasible it would be to use dagster as a library to build on top. I'm being a bit troubled with the documentation, I cannot find anywhere a good description of how the

repository.yaml

or

dagster.yaml

should be built. For example, in the case of

dagster.yaml

I don't want to hard-coded any credentials there, how can I do that? Or what are all the options for

repository.yaml

? I hope to learn more about dagster, it seems super cool and I'm excited to start exploring it.

Hi everyone, I'm back again ! I've updated Dagster to 0.7.13, and I've got an issues with the "imagePullSecrets" and dagster_k8s helm: In my

values.yaml

, i've got the following

imagePullSecrets:
  - name: secret-name

When I try to up my stack using Helm, the container is in status "CrashloopBackOff", and, if i get the logs of the pods, I got the following :

Error 1: Invalid scalar at path root:image_pull_secrets[0]. Value "{'name': 'secret-name'}" of type "<class 'dict'>" is not valid for expected type "String".

I've tried to dig into the issue, and try to edit the configmap-instance.yaml file, specificaly, this part

run_launcher:
      module: dagster_k8s.launcher
      class: K8sRunLauncher
      config:
        {{- with .Values.imagePullSecrets }}
        image_pull_secrets:
            {{- toYaml . | nindent 10 }}
        {{- end }}

I've tried to "hardcode" the value of

image_pull_secrets

After some tries, I've tried

run_launcher:
      module: dagster_k8s.launcher
      class: K8sRunLauncher
      config:
        image_pull_secrets:
        -  "secret-name"

And got the following issue

dagster.check.CheckError: Member of list mismatches type. Expected <class 'dict'>. Got 'ecr-auth' of type <class 'str'>

Seems like there is a missmatch betwwen the expected type in the config, and the one in the k8sLauncher Any input on this, or any solutions to solve this issue ? Thanks again for your great work 💪

set the channel topic: General discussion and announcements!

Just our of curiosity is the ability to perform a pipeline backfill from the UI on the roadmap? Right now it looks like it’s only executable from the CLI (https://docs.dagster.io/docs/learn/guides/partitioning/backfill)

hi, i was trying to deploy some code to aws but at the end i got No such command 'up'. when i try dagster-aws up, maybe i did something wrong in the config? or, now is there other command ?

I'm in a scenario where I can have a lot of inputs that may be processed in parallel and I've been going through some of the past discussions on this topic.

I'm trying to use the solid factory pattern, what is the best way to access the inputs in the solid business logic? Can the inputs be read from the

context

or do I need to use

*args

or

**kwargs

?

I have a number of solids that require some information from the same source, in this case a file. Opening the file is expensive because an index needs to be built but once extracted each part can be serialised to disk as small parquet files. What is the best way to implement this type of data source in dagster? A solid? A resource? How to make it work efficiently with multiprocessing and avoid loading the same file many times?

What's the best way of launching a pipeline through code in a nonblocking way? Both

execute_pipeline

and

execute_pipeline_iterator

seem to always block, and I tried copying how the CLI does it (https://github.com/dagster-io/dagster/blob/master/python_modules/dagster/dagster/cli/pipeline.py#L405-L413) but both my code and the cli are throwing this:

$ DAGSTER_HOME=tmp/dagster dagster pipeline launch my_pipeline -y pipelines/repository.yaml -e pipelines/my_pipeline_config.yaml 
Traceback (most recent call last):
  File "/home/danny/.pyenv/versions/project-3.7.7/bin/dagster", line 8, in <module>
    sys.exit(main())
  File "/home/danny/.pyenv/versions/3.7.7/envs/project-3.7.7/lib/python3.7/site-packages/dagster/cli/__init__.py", line 38, in main
    cli(obj={})  # pylint:disable=E1123
  File "/home/danny/.pyenv/versions/3.7.7/envs/project-3.7.7/lib/python3.7/site-packages/click/core.py", line 764, in __call__
    return self.main(*args, **kwargs)
  File "/home/danny/.pyenv/versions/3.7.7/envs/project-3.7.7/lib/python3.7/site-packages/click/core.py", line 717, in main
    rv = self.invoke(ctx)
  File "/home/danny/.pyenv/versions/3.7.7/envs/project-3.7.7/lib/python3.7/site-packages/click/core.py", line 1137, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/home/danny/.pyenv/versions/3.7.7/envs/project-3.7.7/lib/python3.7/site-packages/click/core.py", line 1137, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/home/danny/.pyenv/versions/3.7.7/envs/project-3.7.7/lib/python3.7/site-packages/click/core.py", line 956, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/home/danny/.pyenv/versions/3.7.7/envs/project-3.7.7/lib/python3.7/site-packages/click/core.py", line 555, in invoke
    return callback(*args, **kwargs)
  File "/home/danny/.pyenv/versions/3.7.7/envs/project-3.7.7/lib/python3.7/site-packages/dagster/core/telemetry.py", line 231, in wrap
    result = f(*args, **kwargs)
  File "/home/danny/.pyenv/versions/3.7.7/envs/project-3.7.7/lib/python3.7/site-packages/dagster/cli/pipeline.py", line 417, in pipeline_launch_command
    return instance.launch_run(pipeline_run.run_id)
  File "/home/danny/.pyenv/versions/3.7.7/envs/project-3.7.7/lib/python3.7/site-packages/dagster/core/instance/__init__.py", line 788, in launch_run
    return self._run_launcher.launch_run(self, run)
AttributeError: 'NoneType' object has no attribute 'launch_run'

In case it matters: I haven't configured a dagster.yaml file anywhere. If it's a

run_launcher

config that I'm missing, what should it be if I'm running my code or the cli on the same machine where dagit is running on localhost?