Using Dagster Pipes, part 1: Define a Dagster asset that invokes a subprocess#

In this part of the tutorial, you'll create a Dagster asset that, in its execution function, opens a Dagster pipes session and invokes a subprocess that executes some external code.

Step 1: Define the Dagster asset#

Before getting started, make sure you have fulfilled all the prerequisites for the tutorial. You should have a standalone Python script named external_code.py which looks like the following:

import pandas as pd


def main():
    orders_df = pd.DataFrame({"order_id": [1, 2], "item_id": [432, 878]})
    total_orders = len(orders_df)
    print(f"processing total {total_orders} orders")


if __name__ == "__main__":
    main()

Step 1.1: Define the asset#

First, create a new file named dagster_code.py in the same directory as the external_code.py file you created earlier in the Prerequisites step.

Next, you’ll define the asset. Copy and paste the following into the file:

import shutil

from dagster import (
    AssetExecutionContext,
    MaterializeResult,
    PipesSubprocessClient,
    asset,
    file_relative_path,
)


@asset
def subprocess_asset(
    context: AssetExecutionContext, pipes_subprocess_client: PipesSubprocessClient
) -> MaterializeResult:
    cmd = [shutil.which("python"), file_relative_path(__file__, "external_code.py")]

Here’s what we did in this example:

• Created an asset named subprocess_asset
• Provided AssetExecutionContext as the context argument to the asset. This object provides system information such as resources, config, and logging. We’ll come back to this a bit later in this section.
• Specified a resource for the asset to use, PipesSubprocessClient. We’ll also come back to this in a little bit.
• Declared a command list cmd to run the external script. In the list:
  • First, found the path to the Python executable on the system using shutil.which("python").
  • Then, provided the file path to the file that we want to execute. In this case, it’s the external_code.py file that you created earlier.

Step 1.2: Invoke the external code from the asset#

Then, invoke a subprocess that executes the external code from the asset using the pipes_subprocess_client resource:

import shutil

from dagster import (
    AssetExecutionContext,
    MaterializeResult,
    PipesSubprocessClient,
    asset,
    file_relative_path,
)


@asset
def subprocess_asset(
    context: AssetExecutionContext, pipes_subprocess_client: PipesSubprocessClient
) -> MaterializeResult:
    cmd = [shutil.which("python"), file_relative_path(__file__, "external_code.py")]
    return pipes_subprocess_client.run(
        command=cmd, context=context
    ).get_materialize_result()

Let’s take a look at what this code does:

• The PipesSubprocessClient resource used by the asset exposes a run method.
• When the asset is executed, this method will synchronously execute the subprocess in in a pipes session, and it will return a PipesClientCompletedInvocation object.
• This object contains a get_materialize_result method, which you can use to access the MaterializeResult event reported by the subprocess. We'll talk about how to report events from the subprocess in the next section.
• Lastly, return the result of the subprocess.

Step 2: Define a Definitions object#

To make the asset and subprocess resource loadable and accessible by Dagster's tools, such as the CLI, UI, and Dagster+, you’ll create a Definitions object that contains them.

Copy and paste the following to the bottom of dagster_code.py:

from dagster import Definitions

defs = Definitions(
    assets=[subprocess_asset],
    resources={"pipes_subprocess_client": PipesSubprocessClient()},
)

At this point, dagster_code.py should look like the following:

import shutil

from dagster import (
    AssetExecutionContext,
    Definitions,
    MaterializeResult,
    PipesSubprocessClient,
    asset,
    file_relative_path,
)


@asset
def subprocess_asset(
    context: AssetExecutionContext, pipes_subprocess_client: PipesSubprocessClient
) -> MaterializeResult:
    cmd = [shutil.which("python"), file_relative_path(__file__, "external_code.py")]
    return pipes_subprocess_client.run(
        command=cmd, context=context
    ).get_materialize_result()


defs = Definitions(
    assets=[subprocess_asset],
    resources={"pipes_subprocess_client": PipesSubprocessClient()},
)

Step 3: Run the subprocess from the Dagster UI#

In this step, you’ll execute the subprocess asset you created in earlier steps from the Dagster UI.

1. In a new command line session, run the following to start the UI:

   dagster dev -f dagster_code.py
   

2. Navigate to http://localhost:3000, where you should see the UI:

   

3. Click Materialize located in the top right to run your code:

   

4. Navigate to the Run details page, where you should see the logs for the run:

   

5. In external_code.py, we have a print statement that outputs to stdout. Dagster will display these in the UI's raw compute log view. To see the stdout log, toggle the log section to stdout:

   

What's next?#

At this point, you've created a Dagster asset that invokes an external Python script, launched the code in a subprocess, and viewed the result in Dagster UI. Next, you'll learn how to modify your external code to work with Dagster Pipes to send information back to Dagster.