async pipeline execution

[ahoneycutt]

Hello, I'm attempting to launch a pipeline asynchronously from a flask (user-facing) application. My goal is to launch the pipeline, and return the user to a status screen while the pipeline works. I have tried execute_pipeline_iterator but this only prepares the pipeline for execution as far as I can tell.

I decided to take a different approach, something like the following:

        instance = DagsterInstance.get()
        run = instance.create_run_for_pipeline(
            my_pipeline,
            run_config=run_config
        )
        instance.submit_run(run.run_id, my_pipeline)

The problem I am seeing with this approach is that my_pipeline is not seen as an ExternalPipelineOrigin and I receive the following error:

Object None is not a ExternalPipelineOrigin. Got None with type <class ‘NoneType’>. Desc: External pipeline origin must be set for submitted runs

Is there an example of how to achieve an async execution of a pipeline?

Thanks much.

[schrockn]

cc: @gibsondan

[gibsondan]

Hi @ahoneycutt - we do have functionality that does this but it's a bit buried / not as intuitive as it should be at the moment.

What you want is basically the 'dagster pipeline launch' command - there's a method in dagster.cli.pipeline called 'execute_launch_command' that is pretty much the exact code that that CLI command runs, and I believe is what you want - that currently takes in a kwargs map that's exactly the arguments you would pass into dagster pipeline launch (see the docs here: https://docs.dagster.io/_apidocs/cli or just run "dagster pipeline launch --help" in the command line to see what they are).

For now what I would maybe recommend is just calling that execute_launch_command function directly, and once we have a 'launch_pipeline' method exposed publicly (which will probably take in similar arguments, just expressed more cleanly than just a kwargs map), then maybe you could switch to that?

For example, you might write something like:

 execute_launch_command({"python_module": <module in which the pipeline is defined>, "attribute": <name of the function that defines your repositoy>", "pipeline": <name of your pipeline>}}

Let us know if that works for now!

[ahoneycutt]

Hey, thanks @gibsondan for the quick reply! I will try this out this morning and post my findings here.

[ahoneycutt]

@gibsondan I got things working. Thanks so much for your time on this. This does work for our use-case and I am happy to move to a newer API as you guys roll it out. Here's the code I used to get things working:

instance = DagsterInstance.get()
run_config = json.dumps({
    'solids': {
        'my_solid': {
            'inputs': {
                'list_of_dicts': [{'value': val} for val in my_input],
            }
        }
    }
})
result = execute_launch_command(instance, {
    'repository': 'my_repository',
    'location': 'my_module.my_pipelines.my_pipeline_module',
    'pipeline': 'my_pipeline',
    'config_json': run_config,
})
print(result.is_failure)