Airflow Orchestration

Orchestrating EMR pipelines with an external scheduler like Apache Airflow is a standard production design pattern. Airflow controls the workflow lifecycle: launching a transient EMR cluster, submitting Spark steps, monitoring the steps until completion, and terminating the cluster to avoid high platform fees.

1. Airflow's EMR Operators

Apache Airflow provides specific AWS operators designed to manage Amazon EMR seamlessly:

• EmrCreateJobFlowOperator: Launches a new EMR cluster based on a Python dictionary configuration.
• EmrAddStepsOperator: Appends processing steps (like a spark-submit command) to an active EMR cluster.
• EmrStepSensor: Polls the status of EMR steps until they succeed, fail, or time out.
• EmrTerminateJobFlowOperator: Explicitly shuts down the EMR cluster.

2. Production DAG Example

Here is a complete, production-grade DAG showing a transient cluster workflow:

emr_spark_pipeline_dag.py

from datetime import datetime, timedelta
from airflow import DAG
from airflow.providers.amazon.aws.operators.emr import (
    EmrCreateJobFlowOperator,
    EmrAddStepsOperator,
    EmrTerminateJobFlowOperator,
)
from airflow.providers.amazon.aws.sensors.emr import EmrStepSensor

# Default arguments for the DAG
default_args = {
    'owner': 'data_engineering',
    'depends_on_past': False,
    'start_date': datetime(2026, 1, 1),
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

# Define the cluster specification
JOB_FLOW_OVERRIDES = {
    'Name': 'Airflow-Orchestrated-Transient-Cluster',
    'ReleaseLabel': 'emr-6.10.0',
    'Applications': [{'Name': 'Spark'}, {'Name': 'Hadoop'}],
    'Instances': {
        'InstanceGroups': [
            {
                'Name': 'Primary node',
                'Market': 'ON_DEMAND',
                'InstanceRole': 'MASTER',
                'InstanceType': 'm5.xlarge',
                'InstanceCount': 1,
            },
            {
                'Name': 'Core node',
                'Market': 'ON_DEMAND',
                'InstanceRole': 'CORE',
                'InstanceType': 'm5.xlarge',
                'InstanceCount': 1,
            },
            {
                'Name': 'Task node (Spot)',
                'Market': 'SPOT',
                'InstanceRole': 'TASK',
                'InstanceType': 'm5.xlarge',
                'InstanceCount': 2,
            }
        ],
        'KeepJobFlowAliveWhenNoSteps': True, # Keep alive during task steps; Airflow terminates at the end
        'TerminationProtected': False,
    },
    'JobFlowRole': 'EMR_EC2_DefaultRole',
    'ServiceRole': 'EMR_DefaultRole',
}

# Define the Spark Step to run
SPARK_STEPS = [
    {
        'Name': 'Execute PySpark ETL',
        'ActionOnFailure': 'CONTINUE', # Let Airflow handle cluster termination rather than automatic EMR exit
        'HadoopJarStep': {
            'Jar': 'command-runner.jar',
            'Args': [
                'spark-submit',
                '--deploy-mode', 'cluster',
                's3://my-spark-jobs-bucket/scripts/emr_pyspark_etl.py',
                '--input', 's3://my-spark-jobs-bucket/raw-data/',
                '--output', 's3://my-spark-jobs-bucket/processed/'
            ]
        }
    }
]

with DAG(
    'emr_pyspark_orchestration_workflow',
    default_args=default_args,
    description='Create transient EMR cluster, run PySpark step, check status, terminate EMR',
    schedule_interval='@daily',
    catchup=False,
    max_active_runs=1,
) as dag:

    # 1. Spin up a transient EMR cluster
    create_emr_cluster = EmrCreateJobFlowOperator(
        task_id='create_emr_cluster',
        job_flow_overrides=JOB_FLOW_OVERRIDES,
        aws_conn_id='aws_default',
    )

    # 2. Add the PySpark script step to the created cluster
    add_step = EmrAddStepsOperator(
        task_id='add_spark_step',
        job_flow_id="{{ task_instance.xcom_pull(task_ids='create_emr_cluster', key='return_value') }}",
        steps=SPARK_STEPS,
        aws_conn_id='aws_default',
    )

    # 3. Monitor the Step execution status
    watch_step = EmrStepSensor(
        task_id='watch_spark_step',
        job_flow_id="{{ task_instance.xcom_pull(task_ids='create_emr_cluster', key='return_value') }}",
        step_id="{{ task_instance.xcom_pull(task_ids='add_spark_step', key='return_value')[0] }}",
        aws_conn_id='aws_default',
        poke_interval=30, # Check status every 30 seconds
        timeout=3600,     # Time out after 1 hour
    )

    # 4. Terminate the cluster (Ensures cluster is closed whether step succeeded or failed)
    terminate_emr_cluster = EmrTerminateJobFlowOperator(
        task_id='terminate_emr_cluster',
        job_flow_id="{{ task_instance.xcom_pull(task_ids='create_emr_cluster', key='return_value') }}",
        aws_conn_id='aws_default',
        trigger_rule='all_done', # Always execute to prevent orphan cluster fees
    )

    # Define DAG execution sequence
    create_emr_cluster >> add_step >> watch_step >> terminate_emr_cluster

3. Best Practices for Airflow + EMR Orchestration

1. Always use the all_done Trigger Rule for Termination: Set the termination operator's trigger_rule to all_done (or one_failed and all_success separately). This guarantees that even if your Spark step fails mid-run, Airflow will successfully terminate the cluster so you do not run up infinite EC2 execution fees.
2. Dynamic Configurations: Utilize Airflow's templating features ({{ ds }}) to dynamically inject the current execution date into your S3 inputs and outputs (e.g., --input s3://my-bucket/raw/{{ ds }}/).
3. Use Transient Clusters: Avoid running persistent 24/7 EMR clusters unless they are actively utilized for ad-hoc querying. Building a dynamic transient cluster per DAG run minimizes idle compute expenses.