|
| 1 | +--- |
| 2 | +title: "Integrating AWS EMR on EKS with Dagster Pipes | Dagster Docs" |
| 3 | +description: "Learn to integrate Dagster Pipes with AWS EMR Containers to launch external code from Dagster assets." |
| 4 | +--- |
| 5 | + |
| 6 | +# AWS EMR on EKS & Dagster Pipes |
| 7 | + |
| 8 | +This tutorial gives a short overview on how to use [Dagster Pipes](/concepts/dagster-pipes) with [AWS EMR on EKS](https://aws.amazon.com/emr/features/eks/) (the corresponding AWS API is called `emr-containers`). |
| 9 | + |
| 10 | +The [dagster-aws](/\_apidocs/libraries/dagster-aws) integration library provides the <PyObject object="PipesEMRContainersClient" module="dagster_aws.pipes" /> resource, which can be used to launch EMR 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. |
| 11 | + |
| 12 | +--- |
| 13 | + |
| 14 | +## Prerequisites |
| 15 | + |
| 16 | +- **In the Dagster environment**, you'll need to: |
| 17 | + |
| 18 | + - Install the following packages: |
| 19 | + |
| 20 | + ```shell |
| 21 | + pip install dagster dagster-webserver dagster-aws |
| 22 | + ``` |
| 23 | + |
| 24 | + Refer to the [Dagster installation guide](/getting-started/install) for more info. |
| 25 | + |
| 26 | + - **AWS authentication credentials configured.** If you don't have this set up already, refer to the [boto3 quickstart](https://boto3.amazonaws.com/v1/documentation/api/latest/guide/quickstart.html). |
| 27 | + |
| 28 | +- **In AWS**: |
| 29 | + |
| 30 | + - An existing AWS account |
| 31 | + - A [EMR Virtual Cluster](https://docs.aws.amazon.com/emr/latest/EMR-on-EKS-DevelopmentGuide/virtual-cluster.html) set up |
| 32 | + |
| 33 | +--- |
| 34 | + |
| 35 | +## Step 1: Install the dagster-pipes module in your EMR environment |
| 36 | + |
| 37 | +There are [a few options](https://aws.github.io/aws-emr-containers-best-practices/submit-applications/docs/spark/pyspark/#python-code-with-python-dependencies) for deploying Python code & dependencies for PySpark jobs. In this tutorial, we are going to build a custom Docker image for this purpose. |
| 38 | + |
| 39 | +Install `dagster-pipes`, `dagster-aws` and `boto3` Python packages in your image: |
| 40 | + |
| 41 | +```Dockerfile,file=/guides/dagster/dagster_pipes/emr-containers/Dockerfile |
| 42 | +# start from EMR image |
| 43 | +FROM public.ecr.aws/emr-containers/spark/emr-7.2.0:latest |
| 44 | +
|
| 45 | +USER root |
| 46 | +
|
| 47 | +RUN python -m pip install dagster-pipes |
| 48 | +
|
| 49 | +# copy the job script |
| 50 | +COPY . . |
| 51 | +
|
| 52 | +USER hadoop |
| 53 | +``` |
| 54 | + |
| 55 | +<Note> |
| 56 | + It's also recommended to upgrade the default Python version included in the |
| 57 | + base EMR image (as it has been done in the `Dockerfile` above) |
| 58 | +</Note> |
| 59 | + |
| 60 | +--- |
| 61 | + |
| 62 | +We copy the EMR job script (`script.py`) to the image in the last step. |
| 63 | + |
| 64 | +## Step 2: Invoke dagster-pipes in the EMR job script |
| 65 | + |
| 66 | +Call `open_dagster_pipes` in the EMR script to create a context that can be used to send messages to Dagster: |
| 67 | + |
| 68 | +```python file=/guides/dagster/dagster_pipes/emr-containers/script.py |
| 69 | +import sys |
| 70 | + |
| 71 | +import boto3 |
| 72 | +from dagster_pipes import PipesS3MessageWriter, open_dagster_pipes |
| 73 | +from pyspark.sql import SparkSession |
| 74 | + |
| 75 | + |
| 76 | +def main(): |
| 77 | + s3_client = boto3.client("s3") |
| 78 | + with open_dagster_pipes( |
| 79 | + message_writer=PipesS3MessageWriter(client=s3_client), |
| 80 | + ) as pipes: |
| 81 | + pipes.log.info("Hello from AWS EMR Containers!") |
| 82 | + |
| 83 | + spark = SparkSession.builder.appName("HelloWorld").getOrCreate() |
| 84 | + |
| 85 | + df = spark.createDataFrame( |
| 86 | + [(1, "Alice", 34), (2, "Bob", 45), (3, "Charlie", 56)], |
| 87 | + ["id", "name", "age"], |
| 88 | + ) |
| 89 | + |
| 90 | + # calculate a really important statistic |
| 91 | + avg_age = float(df.agg({"age": "avg"}).collect()[0][0]) |
| 92 | + |
| 93 | + # attach it to the asset materialization in Dagster |
| 94 | + pipes.report_asset_materialization( |
| 95 | + metadata={"average_age": {"raw_value": avg_age, "type": "float"}}, |
| 96 | + data_version="alpha", |
| 97 | + ) |
| 98 | + |
| 99 | + print("Hello from stdout!") |
| 100 | + print("Hello from stderr!", file=sys.stderr) |
| 101 | + |
| 102 | + |
| 103 | +if __name__ == "__main__": |
| 104 | + main() |
| 105 | +``` |
| 106 | + |
| 107 | +<Note> |
| 108 | + It's best to use the `PipesS3MessageWriter` with EMR on EKS, because this |
| 109 | + message writer has the ability to capture the Spark driver logs and send them |
| 110 | + to Dagster. |
| 111 | +</Note> |
| 112 | + |
| 113 | +--- |
| 114 | + |
| 115 | +## Step 3: Create an asset using the PipesEMRcontainersClient to launch the job |
| 116 | + |
| 117 | +In the Dagster asset/op code, use the `PipesEMRcontainersClient` resource to launch the job: |
| 118 | + |
| 119 | +```python file=/guides/dagster/dagster_pipes/emr-containers/dagster_code.py startafter=start_asset_marker endbefore=end_asset_marker |
| 120 | +from dagster_aws.pipes import PipesEMRContainersClient |
| 121 | + |
| 122 | +import dagster as dg |
| 123 | + |
| 124 | + |
| 125 | +@dg.asset |
| 126 | +def emr_containers_asset( |
| 127 | + context: dg.AssetExecutionContext, |
| 128 | + pipes_emr_containers_client: PipesEMRContainersClient, |
| 129 | +): |
| 130 | + image = ( |
| 131 | + ... |
| 132 | + ) # it's likely the image can be taken from context.run_tags["dagster/image"] |
| 133 | + |
| 134 | + return pipes_emr_containers_client.run( |
| 135 | + context=context, |
| 136 | + start_job_run_params={ |
| 137 | + "releaseLabel": "emr-7.5.0-latest", |
| 138 | + "virtualClusterId": ..., |
| 139 | + "clientToken": context.run_id, # idempotency identifier for the job run |
| 140 | + "executionRoleArn": ..., |
| 141 | + "jobDriver": { |
| 142 | + "sparkSubmitJobDriver": { |
| 143 | + "entryPoint": "local:///app/script.py", |
| 144 | + "sparkSubmitParameters": f"--conf spark.kubernetes.container.image={image}", |
| 145 | + } |
| 146 | + }, |
| 147 | + }, |
| 148 | + ).get_materialize_result() |
| 149 | +``` |
| 150 | + |
| 151 | +<Note> |
| 152 | + Setting `include_stdio_in_messages` to `True` in the `PipesS3MessageReader` |
| 153 | + will allow the driver logs to be forwarded to the Dagster process. |
| 154 | +</Note> |
| 155 | + |
| 156 | +Materializing this asset will launch the AWS on EKS job and wait for it to complete. 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. |
| 157 | + |
| 158 | +--- |
| 159 | + |
| 160 | +## Step 4: Create Dagster definitions |
| 161 | + |
| 162 | +Next, add the `PipesEMRContainersClient` resource to your project's <PyObject object="Definitions" /> object: |
| 163 | + |
| 164 | +```python file=/guides/dagster/dagster_pipes/emr-containers/dagster_code.py startafter=start_definitions_marker endbefore=end_definitions_marker |
| 165 | +import boto3 |
| 166 | +from dagster_aws.pipes import PipesS3ContextInjector, PipesS3MessageReader |
| 167 | + |
| 168 | +from dagster import Definitions |
| 169 | + |
| 170 | +defs = Definitions( |
| 171 | + assets=[emr_containers_asset], |
| 172 | + resources={ |
| 173 | + "pipes_emr_containers_client": PipesEMRContainersClient( |
| 174 | + message_reader=PipesS3MessageReader( |
| 175 | + client=boto3.client("s3"), |
| 176 | + bucket=..., |
| 177 | + include_stdio_in_messages=True, |
| 178 | + ), |
| 179 | + ) |
| 180 | + }, |
| 181 | +) |
| 182 | +``` |
| 183 | + |
| 184 | +Dagster will now be able to launch the AWS EMR Containers job from the `emr_containers_asset` asset, and receive logs and events from the job. If `include_stdio_in_messages` is set to `True`, the logs will be forwarded to the Dagster process. |
| 185 | + |
| 186 | +--- |
| 187 | + |
| 188 | +## Related |
| 189 | + |
| 190 | +<ArticleList> |
| 191 | + <ArticleListItem |
| 192 | + title="Dagster Pipes" |
| 193 | + href="/concepts/dagster-pipes" |
| 194 | + ></ArticleListItem> |
| 195 | + <ArticleListItem |
| 196 | + title="AWS EMR Containers API reference" |
| 197 | + href="/_apidocs/libraries/dagster-aws#dagster_aws.pipes.PipesEMRContainersClient" |
| 198 | + ></ArticleListItem> |
| 199 | +</ArticleList> |
0 commit comments