AWS operators (S3, Redshift, EMR) in Apache Airflow - Step-by-Step Execution

Process Flow - AWS operators (S3, Redshift, EMR)

Start Airflow DAG

↓

Trigger S3 Operator

↓

Upload/Download Data

↓

Trigger Redshift Operator

↓

Run SQL Queries

↓

Trigger EMR Operator

↓

Run Big Data Jobs

↓

Complete DAG Execution

The Airflow DAG triggers AWS operators in sequence: first S3 for data transfer, then Redshift for SQL tasks, and finally EMR for big data processing.

Execution Sample

Apache Airflow

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 import EmrCreateJobFlowOperator

# Simplified DAG tasks

This code snippet shows how Airflow uses AWS operators to interact with S3, Redshift, and EMR services.

Process Table

Step	Operator Triggered	Action	AWS Service Response	Next Step
1	S3CreateObjectOperator	Upload file to S3 bucket	File uploaded successfully	Trigger RedshiftSQLOperator
2	RedshiftSQLOperator	Execute SQL query on Redshift cluster	Query executed successfully	Trigger EmrCreateJobFlowOperator
3	EmrCreateJobFlowOperator	Start EMR cluster and run job flow	EMR job started	DAG completes
4	DAG Completion	All AWS tasks done	Success	End

💡 All AWS operators executed successfully, DAG run completes.

Status Tracker

Variable	Start	After Step 1	After Step 2	After Step 3	Final
s3_file_status	Not uploaded	Uploaded	Uploaded	Uploaded	Uploaded
redshift_query_status	Not run	Not run	Executed	Executed	Executed
emr_job_status	Not started	Not started	Not started	Started	Completed

Key Moments - 2 Insights

Why does the Redshift operator run only after the S3 operator completes?

Because the Redshift operator depends on data uploaded to S3; as shown in execution_table step 1 and 2, S3 upload must finish before Redshift query runs.

What happens if the EMR job fails to start?

The DAG will stop or retry depending on configuration; the variable_tracker shows emr_job_status changes only after successful start (step 3).

Visual Quiz - 3 Questions

Look at the execution_table, what is the AWS service response after the S3 operator runs?

A File uploaded successfully

B Query executed successfully

C EMR job started

D DAG completed

💡 Hint

Check execution_table row 1, column 'AWS Service Response'

At which step does the Redshift query execute?

A Step 1

B Step 3

C Step 2

D Step 4

💡 Hint

Look at execution_table row 2, 'Operator Triggered' column

If the S3 upload fails, what happens to the Redshift operator?

A It runs anyway

B It waits until S3 upload succeeds

C It runs before S3 upload

D It skips and DAG completes

💡 Hint

Refer to key_moments about operator dependencies and execution_table step order

Concept Snapshot

Airflow AWS Operators:
- S3CreateObjectOperator: upload/download files
- RedshiftSQLOperator: run SQL queries
- EmrCreateJobFlowOperator: start EMR jobs
Operators run in sequence with dependencies
Success means each AWS service confirms action

Full Transcript

This visual execution shows how an Airflow DAG uses AWS operators to manage cloud services. First, the S3 operator uploads a file to a bucket. Once successful, the Redshift operator runs a SQL query on the Redshift cluster. After that, the EMR operator starts a cluster and runs a job flow. Variables track the status of each step, ensuring the next operator runs only after the previous completes. Key moments clarify why order matters and what happens on failure. The quiz tests understanding of the sequence and dependencies.

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