Dagster engine

The Dagster adaptor materialises workflows as Dagster jobs so that you can operate them through Dagster’s tooling (UI, sensors, schedules) while preserving AiiDA provenance.

Installation

Install the Dagster extra:

pip install node_graph_engine[dagster]

Setup an AiiDA profile

verdi presto

Example project

1. Scaffold a Dagster project

dagster project scaffold --name my_project
cd my_project

2. Replace the generated my_project/__init__.py with the following content to define a workflow and expose it to Dagster:

   from dagster import DagsterInstance, Definitions
   from node_graph_engine.engines.dagster import DagsterEngine
   from aiida import load_profile
   from node_graph import task
   
   
   load_profile()
   
   
   @task()
   def add(x, y):
       return x + y
   
   
   @task()
   def multiply(x, y):
       return x * y
   
   
   @task.graph()
   def add_then_multiply(x, y, z):
       the_sum = add(x=x, y=y).result
       return multiply(x=the_sum, y=z).result
   
   
   ng = add_then_multiply.build(x=1, y=2, z=3)
   
   engine = DagsterEngine("add_multiply", instance=DagsterInstance.get())
   job = engine.build_job(ng)
   defs = Definitions(jobs=[job])
   

3. Start the Dagster development server from the project directory.

dagster dev -m my_project

4. Go to the Jobs tab in the Dagster UI, and find the add_multiply job. Click on it, then click on the Launchpad tab to run the job.

Here is the preview of the Dagster UI showing the job execution:

Use AiiDA commands to inspect the processes and their provenance:

verdi process list -a

Which will show something like:

2214  11s ago    NodeGraph<add_then_multiply>         ⏹ Finished [0]
2215  8s ago     add                                  ⏹ Finished [0]
2217  6s ago     multiply                             ⏹ Finished [0]

Then generate a provenance graph for a workflow:

verdi node graph generate 2214 -f png

Here is the resulting graph: