AWS EMR Serverless & Dagster Pipes#

This tutorial gives a short overview on how to use Dagster Pipes with AWS EMR Serverless.

The dagster-aws integration library provides the PipesEMRServerlessClient resource, which can be used to launch AWS EMR Serverless jobs from Dagster assets and ops. Dagster can receive regular events such as logs, asset checks, or asset materializations from jobs launched with this client. Using it requires minimal code changes to your EMR jobs.

Prerequisites#

In the Dagster environment, you'll need to:
- Install the following packages:
```
pip install dagster dagster-webserver dagster-aws
```
  Refer to the Dagster installation guide for more info.
- AWS authentication credentials configured. If you don't have this set up already, refer to the boto3 quickstart.

In AWS:

An existing AWS account
An AWS EMR Serverless job. AWS CloudWatch logging has to be enabled in order to receive logs from the job:


{
  "monitoringConfiguration": {
    "cloudWatchLoggingConfiguration": { "enabled": true }
  }
}

Step 1: Install the dagster-pipes module in your EMR Serverless environment#

There are a few options available for shipping Python packages to a PySpark job. For example, install it in your Docker image:

Install the dagster-pipes module in the image used for your EMR job. For example, you can install the dependency with pip in your image Dockerfile:


# start from EMR image
FROM public.ecr.aws/emr-serverless/spark/emr-7.2.0:latest

USER root

RUN python -m pip install dagster-pipes

# copy the job script
COPY . .

USER hadoop

Step 2: Add dagster-pipes to the EMR Serverless job script#

Call open_dagster_pipes in the EMR Serverless script to create a context that can be used to send messages to Dagster:


from dagster_pipes import open_dagster_pipes
from pyspark.sql import SparkSession


def main():
    with open_dagster_pipes() as pipes:
        pipes.log.info("Hello from AWS EMR Serverless!")

        spark = SparkSession.builder.appName("HelloWorld").getOrCreate()

        df = spark.createDataFrame(
            [(1, "Alice", 34), (2, "Bob", 45), (3, "Charlie", 56)],
            ["id", "name", "age"],
        )

        # calculate a really important statistic
        avg_age = float(df.agg({"age": "avg"}).collect()[0][0])

        # attach it to the asset materialization in Dagster
        pipes.report_asset_materialization(
            metadata={"average_age": {"raw_value": avg_age, "type": "float"}},
            data_version="alpha",
        )

        spark.stop()


if __name__ == "__main__":
    main()

Step 3: Create an asset using the PipesEMRServerlessClient to launch the job#

In the Dagster asset/op code, use the PipesEMRServerlessClient resource to launch the job:


import os

import boto3
from dagster_aws.pipes import PipesEMRServerlessClient

from dagster import AssetExecutionContext, asset


@asset
def emr_serverless_asset(
    context: AssetExecutionContext,
    pipes_emr_serverless_client: PipesEMRServerlessClient,
):
    return pipes_emr_serverless_client.run(
        context=context,
        start_job_run_params={
            "applicationId": "<app-id>",
            "executionRoleArn": "<emr-role>",
            "clientToken": context.run_id,  # idempotency identifier for the job run
            "configurationOverrides": {
                "monitoringConfiguration": {
                    "cloudWatchLoggingConfiguration": {"enabled": True}
                }
            },
        },
    ).get_results()

This will launch the AWS EMR Serverless job and wait for it completion. If the job fails, the Dagster process will raise an exception. If the Dagster process is interrupted while the job is still running, the job will be terminated.

Step 4: Create Dagster definitions#

Next, add the PipesEMRServerlessClient resource to your project's Definitions object:


from dagster import Definitions  # noqa


defs = Definitions(
    assets=[emr_serverless_asset],
    resources={"pipes_emr_serverless_client": PipesEMRServerlessClient()},
)