AWS operators (S3, Redshift, EMR) in Apache Airflow - Mini Project: Build & Apply

AWS Operators with Airflow: S3, Redshift, and EMR

📖 Scenario: You are working as a data engineer. Your team uses Apache Airflow to automate cloud tasks. You need to create a simple workflow that uploads a file to AWS S3, loads data into Redshift, and runs a job on EMR.

🎯 Goal: Build an Airflow DAG that uses AWS operators to upload a file to S3, copy data into Redshift, and start an EMR job flow.

📋 What You'll Learn

Create an Airflow DAG named aws_data_pipeline

Use S3CreateObjectOperator to upload a file to an S3 bucket

Use RedshiftSQLOperator to run a SQL COPY command in Redshift

Use EmrCreateJobFlowOperator to start an EMR cluster

Set task dependencies so the S3 upload runs before Redshift load, which runs before EMR job

💡 Why This Matters

🌍 Real World

Automating data workflows in the cloud is common in data engineering. Airflow helps schedule and manage these tasks reliably.

💼 Career

Knowing how to use AWS operators in Airflow is valuable for cloud data engineers and DevOps professionals managing ETL pipelines.

1

Create the Airflow DAG and S3 upload task

Create an Airflow DAG named aws_data_pipeline with default arguments. Inside it, create a task called upload_to_s3 using S3CreateObjectOperator to upload the string 'Hello, Airflow!' as a file named greeting.txt to the bucket my-test-bucket.

Apache Airflow

from airflow import DAG
from airflow.providers.amazon.aws.operators.s3 import S3CreateObjectOperator
from datetime import datetime

# Your code here

Need a hint?

Remember to import S3CreateObjectOperator and set bucket_name, object_key, and data exactly as specified.

2

Add Redshift load task configuration

Add a task called load_to_redshift using RedshiftSQLOperator inside the same DAG. Configure it to run the SQL command COPY my_table FROM 's3://my-test-bucket/greeting.txt' IAM_ROLE 'arn:aws:iam::123456789012:role/MyRedshiftRole' FORMAT AS TEXT; on the Redshift cluster. Use redshift_default as the connection ID.

Apache Airflow

from airflow.providers.amazon.aws.operators.redshift_sql import RedshiftSQLOperator

from airflow import DAG
from airflow.providers.amazon.aws.operators.s3 import S3CreateObjectOperator
from datetime import datetime

default_args = {
    'start_date': datetime(2024, 1, 1)
}

with DAG('aws_data_pipeline', default_args=default_args, schedule_interval=None, catchup=False) as dag:
    upload_to_s3 = S3CreateObjectOperator(
        task_id='upload_to_s3',
        aws_conn_id='aws_default',
        bucket_name='my-test-bucket',
        object_key='greeting.txt',
        data='Hello, Airflow!'
    )

    # Add load_to_redshift task below
    # Your code here

Need a hint?

Use triple quotes for the SQL string and set redshift_conn_id exactly as shown.

3

Add EMR job flow creation task

Add a task called start_emr_cluster using EmrCreateJobFlowOperator inside the DAG. Configure it with job_flow_overrides to create an EMR cluster with the name AirflowEMRCluster, release label emr-6.3.0, and one master node of type m5.xlarge. Use aws_default as the AWS connection ID.

Apache Airflow

from airflow.providers.amazon.aws.operators.emr_create_job_flow import EmrCreateJobFlowOperator

from airflow import DAG
from airflow.providers.amazon.aws.operators.s3 import S3CreateObjectOperator
from airflow.providers.amazon.aws.operators.redshift_sql import RedshiftSQLOperator
from datetime import datetime

default_args = {
    'start_date': datetime(2024, 1, 1)
}

with DAG('aws_data_pipeline', default_args=default_args, schedule_interval=None, catchup=False) as dag:
    upload_to_s3 = S3CreateObjectOperator(
        task_id='upload_to_s3',
        aws_conn_id='aws_default',
        bucket_name='my-test-bucket',
        object_key='greeting.txt',
        data='Hello, Airflow!'
    )

    load_to_redshift = RedshiftSQLOperator(
        task_id='load_to_redshift',
        redshift_conn_id='redshift_default',
        sql="""COPY my_table FROM 's3://my-test-bucket/greeting.txt' IAM_ROLE 'arn:aws:iam::123456789012:role/MyRedshiftRole' FORMAT AS TEXT;"""
    )

    # Add start_emr_cluster task below
    # Your code here

Need a hint?

Define JOB_FLOW_OVERRIDES as a dictionary with the specified keys and values before the DAG context.

4

Set task dependencies to define workflow order

Set the task dependencies so that upload_to_s3 runs before load_to_redshift, and load_to_redshift runs before start_emr_cluster.

Apache Airflow

from airflow import DAG
from airflow.providers.amazon.aws.operators.s3 import S3CreateObjectOperator
from airflow.providers.amazon.aws.operators.redshift_sql import RedshiftSQLOperator
from airflow.providers.amazon.aws.operators.emr_create_job_flow import EmrCreateJobFlowOperator
from datetime import datetime

default_args = {
    'start_date': datetime(2024, 1, 1)
}

JOB_FLOW_OVERRIDES = {
    'Name': 'AirflowEMRCluster',
    'ReleaseLabel': 'emr-6.3.0',
    'Instances': {
        'InstanceGroups': [
            {
                'Name': 'Master nodes',
                'Market': 'ON_DEMAND',
                'InstanceRole': 'MASTER',
                'InstanceType': 'm5.xlarge',
                'InstanceCount': 1
            }
        ],
        'KeepJobFlowAliveWhenNoSteps': True,
        'TerminationProtected': False
    },
    'JobFlowRole': 'EMR_EC2_DefaultRole',
    'ServiceRole': 'EMR_DefaultRole'
}

with DAG('aws_data_pipeline', default_args=default_args, schedule_interval=None, catchup=False) as dag:
    upload_to_s3 = S3CreateObjectOperator(
        task_id='upload_to_s3',
        aws_conn_id='aws_default',
        bucket_name='my-test-bucket',
        object_key='greeting.txt',
        data='Hello, Airflow!'
    )

    load_to_redshift = RedshiftSQLOperator(
        task_id='load_to_redshift',
        redshift_conn_id='redshift_default',
        sql="""COPY my_table FROM 's3://my-test-bucket/greeting.txt' IAM_ROLE 'arn:aws:iam::123456789012:role/MyRedshiftRole' FORMAT AS TEXT;"""
    )

    start_emr_cluster = EmrCreateJobFlowOperator(
        task_id='start_emr_cluster',
        aws_conn_id='aws_default',
        job_flow_overrides=JOB_FLOW_OVERRIDES
    )

    # Set task dependencies below
    # Your code here

Need a hint?

Use the >> operator to set the order of tasks.