AWS operators (S3, Redshift, EMR) in Apache Airflow - Commands & Configuration

Introduction

AWS operators in Airflow help automate tasks like moving files to S3, running queries on Redshift, and managing EMR clusters. They solve the problem of manually handling cloud data workflows by letting you schedule and control these tasks easily.

When you want to upload or download files automatically to Amazon S3 as part of your data pipeline.

When you need to run SQL queries or load data into Amazon Redshift without manual intervention.

When you want to start, monitor, and terminate Amazon EMR clusters for big data processing jobs.

When you want to chain cloud data tasks in a workflow that runs on a schedule.

When you want to track success or failure of AWS tasks inside Airflow for better monitoring.

Config File - aws_operators_dag.py

aws_operators_dag.py

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 airflow.providers.amazon.aws.operators.emr_terminate_job_flow import EmrTerminateJobFlowOperator
from airflow.utils.dates import days_ago

JOB_FLOW_OVERRIDES = {
    'Name': 'example-emr-cluster',
    'ReleaseLabel': 'emr-6.9.0',
    'Instances': {
        'InstanceGroups': [
            {
                'Name': 'Master nodes',
                'Market': 'ON_DEMAND',
                'InstanceRole': 'MASTER',
                'InstanceType': 'm5.xlarge',
                'InstanceCount': 1,
            },
            {
                'Name': 'Core nodes',
                'Market': 'ON_DEMAND',
                'InstanceRole': 'CORE',
                'InstanceType': 'm5.xlarge',
                'InstanceCount': 2,
            },
        ],
        'KeepJobFlowAliveWhenNoSteps': True,
        'TerminationProtected': False,
    },
    'Applications': [{'Name': 'Hadoop'}, {'Name': 'Spark'}],
}

with DAG(
    dag_id='aws_operators_example',
    start_date=days_ago(1),
    schedule_interval='@once',
    catchup=False,
) as dag:

    create_s3_object = S3CreateObjectOperator(
        task_id='create_s3_object',
        s3_bucket='my-example-bucket',
        s3_key='example_folder/example_file.txt',
        data='Hello from Airflow AWS S3 Operator!',
    )

    run_redshift_query = RedshiftSQLOperator(
        task_id='run_redshift_query',
        sql='SELECT current_date;',
        redshift_conn_id='redshift_default',
    )

    create_emr_cluster = EmrCreateJobFlowOperator(
        task_id='create_emr_cluster',
        job_flow_overrides=JOB_FLOW_OVERRIDES,
    )

    terminate_emr_cluster = EmrTerminateJobFlowOperator(
        task_id='terminate_emr_cluster',
        job_flow_id='{{ task_instance.xcom_pull(task_ids="create_emr_cluster", key="return_value") }}',
    )

    create_s3_object >> run_redshift_query >> create_emr_cluster >> terminate_emr_cluster

This Airflow DAG file defines a workflow using AWS operators:

• S3CreateObjectOperator: uploads a text file to an S3 bucket.
• RedshiftSQLOperator: runs a simple SQL query on Redshift.
• EmrCreateJobFlowOperator: creates an EMR cluster with specified settings.
• EmrTerminateJobFlowOperator: terminates the EMR cluster after use.

The DAG runs these tasks in order, automating a common cloud data pipeline.

Commands

List all available DAGs to confirm the new AWS operators DAG is recognized by Airflow.

Terminal

airflow dags list

Expected Output

aws_operators_example

example_dag

Trigger the AWS operators DAG manually to start the workflow immediately.

Terminal

airflow dags trigger aws_operators_example

Expected Output

Created <DagRun aws_operators_example @ 2024-06-01T12:00:00+00:00: manual__2024-06-01T12:00:00+00:00, externally triggered: True>

List all tasks in the aws_operators_example DAG to see the steps involved.

Terminal

airflow tasks list aws_operators_example

Expected Output

create_s3_object

run_redshift_query

create_emr_cluster

terminate_emr_cluster

Test the create_s3_object task for the given date to verify it uploads the file to S3 correctly.

Terminal

airflow tasks test aws_operators_example create_s3_object 2024-06-01

Expected Output

[2024-06-01 12:00:00,000] {s3.py:123} INFO - Uploading data to s3://my-example-bucket/example_folder/example_file.txt

[2024-06-01 12:00:01,000] {s3.py:130} INFO - Upload successful

Key Concept

If you remember nothing else from this pattern, remember: AWS operators in Airflow let you automate cloud tasks like file uploads, queries, and cluster management as part of scheduled workflows.

Common Mistakes

Not setting up AWS connection IDs in Airflow before using AWS operators.

The operators need valid AWS credentials configured in Airflow to connect and perform actions; without them, tasks fail with authentication errors.

Configure AWS connections in Airflow UI or CLI with proper access keys and region before running AWS operator tasks.

Using incorrect or missing S3 bucket names or Redshift connection IDs in operator parameters.

The operators cannot find the target resources and will error out or silently fail.

Double-check bucket names, Redshift connection IDs, and other parameters for accuracy before running the DAG.

Not handling EMR cluster job flow IDs properly when terminating clusters.

If the terminate task does not get the correct cluster ID from the create task, the cluster will not be terminated, causing extra costs.

Use XComs to pass the cluster ID from the create task to the terminate task as shown in the example.

Summary

Define AWS operators in an Airflow DAG to automate S3 uploads, Redshift queries, and EMR cluster management.

Use airflow CLI commands to list, trigger, and test the DAG and its tasks.

Ensure AWS connections and parameters are correctly set to avoid authentication and resource errors.

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