Have the inputs and outputs each be a list or similar instead

If I could work out how to simply produce Asset materialisations to the storage system I'd do it that way, just so the tester has access to the intermediate steps. Then I'd be happy to just emit collections.

Not sure you mean, you produce those by yielding asset events from the solid

Yes you're right, asset events are just a way of flagging that we've stored something. I guess I'm fighting against having to explicitly persist something just for the tester. Using intermediates storage seemed like an elegant approach until I ran into this.

I guess I don't understand why returning a list instead of a variable number of outputs is related to any of that.

Hello is it possible to execute other .bat file from Dagster ?

Hello Everyone! I am trying Dagster, thinking about maybe replacing our Azkaban (Pipeline orchestrator similar to Airflow). Our schedules are usually Spark jobs. I was wondering if someone have spark job flows built on Dagster that can be shared.... I couldn't find doing some research.... Thanks a lot!

Is there a way to set "retry from failure" as the default behavior for a pipeline (ideally with some retry limit)?

Hello! Is it possible to declare something like a global input for the solids in the configuration? For instance many solids take the input

bag_cleaned_schema

which I want to pass in the configuration, and preferably only declare it at once place and refer to that in the solid inputs. Now I have this configuration:

solids:
  bag_ahn_date:
    inputs:
      bag_cleaned: "bag_cleaned.pandactueelbestaand"
      acquisition_dates: "ahn3.acquisition_dates"
      bag_cleaned_schema: "bag_cleaned"
  bag_ahn_date_index:
    inputs:
      bag_cleaned_schema: "bag_cleaned"
  bag_ahn_overlap:
    inputs:
      bag_cleaned: "bag_cleaned.pandactueelbestaand"
      buffer: "schema_name.buffer_table"
      bag_cleaned_schema: "bag_cleaned"
  bag_ahn_overlap_index:
    inputs:
      bag_cleaned_schema: "bag_cleaned"

And I would like to have a configuration like this one, where the

global

is not a solid, but a way to declare a gobal input parameter

solids:
  global:
      bag_cleaned_schema: "bag_cleaned"
  bag_ahn_date:
    inputs:
      bag_cleaned: "bag_cleaned.pandactueelbestaand"
      acquisition_dates: "ahn3.acquisition_dates"
      bag_cleaned_schema: $bag_cleaned_schema
  bag_ahn_date_index:
    inputs:
      bag_cleaned_schema: $bag_cleaned_schema
  bag_ahn_overlap:
    inputs:
      bag_cleaned: "bag_cleaned.pandactueelbestaand"
      buffer: "schema_name.buffer_table"
      bag_cleaned_schema: $bag_cleaned_schema
  bag_ahn_overlap_index:
    inputs:
      bag_cleaned_schema: $bag_cleaned_schema

Hello! We recently upgraded dagster and it looks like an update in

0.9.3

allows dagster to infer input and output descriptions from docstrings. The issue is it appears to do this at runtime and throw an error if the docstring isn't formatted properly (e.g if you use hyphens instead of colons after argument names). Is there a way to turn this off? The docstring parser module brought in by dagster throws this error at runtime:

pypi__36__docstring_parser_0_7_1_linux_x86_64/docstring_parser/google.py", line 106, in _build_meta
    before, desc = text.split(":", 1)
ValueError: not enough values to unpack (expected 2, got 1)

if you have for example:

Args:    
    arg_name - description of arg

instead of:

Args:
    arg_name: description of the arg

Hi Guys! What is the Dagster Way to simplify / use config like parameters in Playground / GraphQL? I put some context into the thread. Thanks in advance if you can help me

You can use the configured() function/decorator to bind in default values

I finally have end-to-end autoscaling for my pipelines, on both the launcher workers and the solids 🙂

Including scale to 0, so when not in use the only runtime overhead is dagit itself and 1 grpc daemon for each workspace

Next up, CI/CD for pipeline repos

what's the difference between using intermediate storage (https://docs.dagster.io/tutorial/advanced_intermediates) vs asset stores (https://docs.dagster.io/overview/asset-stores/asset-stores) in a pipeline?

(i also would love to contribute the code to Dagster, but of cause that would require more sophisticated efforts after getting correct pointers from you guys)

Here's a new one,

dagster.check.CheckError: Invariant failed. Description: Pipeline run condition_asset (20b0921f-2547-4e10-a237-05b83b27f696) in state PipelineRunStatus.STARTED, expected PipelineRunStatus.NOT_STARTED

Any way my launcher could have a race condition?

@max FYI https://github.com/dagster-io/dagster/discussions/3326

Or anyone else who wants to weigh in.

Hello! I am new to dagster. I have a question. Is there a way to run a pipeline multiple times based on a list of values that I can pass in somehow? For example, I slightly modified the hello world example from the main github page where I am executing 2 tasks that don't depend on each other.

from dagster import (

    

execute_pipeline,

      

pipeline,

      

solid

)

@solid

def get_name(_, plant_code: str):

    

print(f"Plant code: {plant_code}")

@solid

def hello(context, plant_code: str):

<http://context.log.info|context.log.info>('Hello, {name}!'.format(name=plant_code))

@pipeline

def hello_pipeline():

    

get_name()

    

hello()

if __name__ == "__main__":

    

run_config = {

        

"solids": {

            

"get_name": {"inputs": {"plant_code": {"value": "Plant1"}}},

            

"hello": {"inputs": {"plant_code": {"value": "Plant1"}}}

        

}

    

}

    

execute_pipeline(hello_pipeline, run_config=run_config)

# want to do something like this:     value = ["Plant1", "Plant2", "Plant3"]     for plant_code in value:         execute_my_pipeline(plant_code) etc... I can create a pipeline for one plant code "Plant1", but let's say I want to execute this set of tasks in this pipeline for each additional plant code. With the run_config, not sure if it is possible to loop through a list of values passed into the "value" parameter. I hope my question makes sense. Any help is much appreciated!

Hello, is it possible to trigger the dagit repo refresh logic programmatically?

Is anyone else having trouble with the

dagster_ge

library?

Hello, I passed in my machine's IP# and port# to the dagit command, but remote users in my immediate group within my network, still can't get to the dagit UI. Is there setting(s) I have to make to open up access? I looked at the options for dagster.yaml, but nothing stood out. Granted, I am not knowledgeable in network protocols. I've had other Python servers where I've had to enable CORS or enable websockets, but I am not too familiar with the dagit's flask setup.

Bob Chen just published a new version: 0.9.21. See https://github.com/dagster-io/dagster/releases/tag/0.9.21 for the release notes.

What’s the difference between ConsolidatedSqliteEventLogStorage and SqliteEventLogStorage? Is it just that the consolidated version is asset-aware? If I’m not looking to persist intermediates, should I be using SqliteEventLogStore?

Here are change notes from yesterday's release 🤗 0.9.21 Change Notes Community Contributions • Fixed helm chart to only add flower to the K8s ingress when enabled (thanks [@PenguinToast](https://github.com/PenguinToast)!) • Updated helm chart to use more lenient timeouts for liveness probes on user code deployments (thanks [@PenguinToast](https://github.com/PenguinToast)!) Bugfixes • [Helm/K8s] Due to Flower being incompatible with Celery 5.0, the Helm chart for Dagster now uses a specific image

mher/flower:0.9.5

for the Flower pod.

Hey guys, quick question, I am trying to execute two solids synchronously and having some trouble. Here is my code where

generate_job_params

is a solid (which is a simple date manipulation) and

spark_solid

is another solid. What happens when I run is that I get an error because

spark_solid

needs APPLICATION_PARAMS to be defined, and it is but with a

<dagster.core.definitions.composition.InvokedSolidOutputHandle object at 0x10920c790>

where I expected to be a formatted datetime.

@pipeline(
    mode_defs=[ModeDefinition(resource_defs={"spark": spark_resource})]
)
def pipeline():
    yd = generate_job_params()
    os.environ["APPLICATION_PARAMS"] = f"--app-name CountCSV --input-csv-path /tmp/aksmiyazaki/dummy.csv --group-column name --output-data-path /tmp/aksmiyazaki/{yd}"
    spark_solid = create_spark_solid("spark_process_simple_csv", "GroupedCount")
    spark_solid()

hi all! I was wondering whether it's possible to use localstack's S3 mock service together with dagster. I see that the endpoint URL can be configured, however localstack uses path-style addressing; this can be configured when using boto3 directly, but can't find how to do this through dagster configuration. Any hints on this? TIA