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

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

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

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

Hint

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

Practice

1. What is the primary purpose of the S3CreateObjectOperator in Airflow?

easy

A. To delete an S3 bucket

B. To list all objects in an S3 bucket

C. To create or upload an object (file) to an S3 bucket

D. To move objects between two S3 buckets

Solution

1. Step 1: Understand the operator's name and function

   The name S3CreateObjectOperator suggests it creates or uploads an object to S3.

2. Step 2: Match the operator to its main action

   This operator is designed to create or upload files to an S3 bucket, not delete or list buckets.

3. Final Answer:

   To create or upload an object (file) to an S3 bucket -> Option C

4. Quick Check:

   S3CreateObjectOperator = Upload file to S3 [OK]

Hint: Operator names usually describe their main action clearly [OK]

Common Mistakes:

• Confusing object creation with bucket deletion
• Thinking it lists objects instead of creating
• Assuming it moves objects between buckets

2. Which of the following is the correct way to instantiate an RedshiftSQLOperator in Airflow?

easy

A. RedshiftSQLOperator(sql='SELECT * FROM table', aws_conn_id='my_conn')

B. RedshiftSQLOperator(query='SELECT * FROM table', aws_conn_id='my_conn')

C. RedshiftSQLOperator(command='SELECT * FROM table', redshift_conn='my_conn')

D. RedshiftSQLOperator(sql='SELECT * FROM table', redshift_conn_id='my_conn')

Solution

1. Step 1: Identify correct parameter names for RedshiftSQLOperator

   The operator expects sql for the query and redshift_conn_id for the connection ID.

2. Step 2: Check each option for correct parameters

   RedshiftSQLOperator(sql='SELECT * FROM table', redshift_conn_id='my_conn') uses sql and redshift_conn_id, which is correct. Others use wrong parameter names.

3. Final Answer:

   RedshiftSQLOperator(sql='SELECT * FROM table', redshift_conn_id='my_conn') -> Option D

4. Quick Check:

   Use sql and redshift_conn_id for RedshiftSQLOperator [OK]

Hint: Remember: SQL query = sql, connection = redshift_conn_id [OK]

Common Mistakes:

• Using 'query' instead of 'sql' parameter
• Using 'aws_conn_id' instead of 'redshift_conn_id'
• Misspelling parameter names

3. Given the following Airflow task using EmrCreateJobFlowOperator:

emr_task = EmrCreateJobFlowOperator(
    task_id='start_emr_cluster',
    job_flow_overrides={
        'Name': 'TestCluster',
        'ReleaseLabel': 'emr-6.3.0',
        'Instances': {'InstanceCount': 3, 'KeepJobFlowAliveWhenNoSteps': True}
    },
    aws_conn_id='aws_default'
)

What will be the state of the EMR cluster after this task runs successfully?

medium

A. A running EMR cluster named 'TestCluster' with 3 instances, waiting for steps

B. An EMR cluster terminated immediately after creation

C. An EMR cluster with zero instances running

D. An error because 'ReleaseLabel' is missing

Solution

1. Step 1: Analyze job_flow_overrides parameters

   The cluster is named 'TestCluster', uses EMR release 6.3.0, has 3 instances, and is set to keep alive when no steps run.

2. Step 2: Understand KeepJobFlowAliveWhenNoSteps effect

   This setting keeps the cluster running after creation, waiting for steps to be added.

3. Final Answer:

   A running EMR cluster named 'TestCluster' with 3 instances, waiting for steps -> Option A

4. Quick Check:

   KeepJobFlowAliveWhenNoSteps=True means cluster stays running [OK]

Hint: KeepJobFlowAliveWhenNoSteps=True keeps cluster alive after creation [OK]

Common Mistakes:

• Assuming cluster terminates immediately
• Ignoring instance count setting
• Thinking ReleaseLabel is missing

4. You wrote this Airflow task to copy data from S3 to Redshift using RedshiftToS3Operator:

copy_task = RedshiftToS3Operator(
    task_id='copy_data',
    redshift_conn_id='redshift_default',
    s3_bucket='my-bucket',
    s3_key='data/output/',
    sql='SELECT * FROM users'
)

But the task fails with an error about missing parameters. What is the likely cause?

medium

A. The sql parameter is not supported by RedshiftToS3Operator

B. The operator name is incorrect; it should be S3ToRedshiftOperator

C. The s3_key parameter must be a full S3 URI starting with s3://

D. The redshift_conn_id must be named aws_conn_id

Solution

1. Step 1: Check operator purpose and name

   RedshiftToS3Operator copies data from Redshift to S3, but the task description says copying from S3 to Redshift.

2. Step 2: Identify correct operator for S3 to Redshift copy

   The correct operator for copying data from S3 to Redshift is S3ToRedshiftOperator.

3. Final Answer:

   The operator name is incorrect; it should be S3ToRedshiftOperator -> Option B

4. Quick Check:

   RedshiftToS3Operator copies Redshift to S3, not reverse [OK]

Hint: Check operator name matches data flow direction [OK]

Common Mistakes:

• Mixing up RedshiftToS3Operator and S3ToRedshiftOperator
• Assuming sql parameter works for all operators
• Incorrect S3 URI format

5. You want to automate a workflow where you first upload a CSV file to S3, then run a Redshift COPY command to load the data, and finally start an EMR cluster for processing. Which sequence of Airflow operators correctly implements this?

hard

A. S3CreateObjectOperator -> RedshiftSQLOperator -> EmrCreateJobFlowOperator

B. RedshiftSQLOperator -> S3CreateObjectOperator -> EmrCreateJobFlowOperator

C. EmrCreateJobFlowOperator -> S3CreateObjectOperator -> RedshiftSQLOperator

D. S3CreateObjectOperator -> EmrCreateJobFlowOperator -> RedshiftSQLOperator

Solution

1. Step 1: Identify the logical order of tasks

   First, upload the CSV to S3 using S3CreateObjectOperator.

2. Step 2: Load data into Redshift using COPY command via RedshiftSQLOperator

   After the file is in S3, run the COPY command in Redshift to load data.

3. Step 3: Start EMR cluster with EmrCreateJobFlowOperator for processing

   Finally, start the EMR cluster to process the loaded data.

4. Final Answer:

   S3CreateObjectOperator -> RedshiftSQLOperator -> EmrCreateJobFlowOperator -> Option A

5. Quick Check:

   Upload -> Load -> Process = A [OK]

Hint: Think data flow: upload, load, then process [OK]

Common Mistakes:

• Running Redshift load before uploading to S3
• Starting EMR before data is loaded
• Mixing operator order