AWS operators (S3, Redshift, EMR) in Apache Airflow - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ service_behavior

intermediate

What happens when an Airflow S3CreateBucketOperator runs with an existing bucket name?

You use S3CreateBucketOperator in Airflow to create a bucket named my-data-bucket. The bucket already exists in your AWS account.

What will be the result of running this operator?

Apache Airflow

from airflow.providers.amazon.aws.operators.s3 import S3CreateBucketOperator

create_bucket = S3CreateBucketOperator(
    task_id='create_bucket',
    bucket_name='my-data-bucket',
    aws_conn_id='aws_default'
)

create_bucket.execute(context={})

A The operator completes successfully without error because the bucket already exists.

B The operator creates a new bucket with a suffix added to the name to avoid conflict.

C The operator deletes the existing bucket and creates a new one with the same name.

D The operator raises an error indicating the bucket name is already taken.

💡 Hint

Think about AWS S3 bucket naming rules and uniqueness across all AWS accounts.

S3 bucket names must be globally unique. Trying to create a bucket with a name that already exists (even in another account) causes an error. The operator does not silently succeed or rename the bucket.

❓ Configuration

intermediate

Which Airflow Redshift operator configuration correctly runs a SQL query on Redshift?

You want to run a SQL query on an Amazon Redshift cluster using Airflow's RedshiftSQLOperator. Which configuration will successfully execute the query?

Apache Airflow

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

redshift_query = RedshiftSQLOperator(
    task_id='run_query',
    sql='SELECT COUNT(*) FROM users;',
    redshift_conn_id='redshift_default'
)

redshift_query.execute(context={})

A sql='SELECT COUNT(*) FROM users;', redshift_conn_id='redshift_default'

B sql=['SELECT COUNT(*) FROM users;'], redshift_conn_id='redshift_default'

C sql='SELECT COUNT(*) FROM users;', aws_conn_id='aws_default'

D sql='SELECT COUNT(*) FROM users;', redshift_conn_id='aws_default'

💡 Hint

Check the parameter names and types expected by RedshiftSQLOperator.

The sql parameter must be a string or list of strings with SQL commands. The connection ID must be redshift_conn_id to connect to Redshift. Using aws_conn_id is incorrect here.

❓ Architecture

advanced

Which Airflow EMR operator setup correctly launches an EMR cluster with a bootstrap action?

You want to launch an EMR cluster using Airflow's EmrCreateJobFlowOperator and run a bootstrap action script stored in S3. Which configuration is correct?

Apache Airflow

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

JOB_FLOW_OVERRIDES = {
    'Name': 'TestCluster',
    '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
    },
    'BootstrapActions': [
        {
            'Name': 'Install libraries',
            'ScriptBootstrapAction': {
                'Path': 's3://my-bucket/bootstrap.sh'
            }
        }
    ]
}

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

create_emr_cluster.execute(context={})

A JOB_FLOW_OVERRIDES includes 'BootstrapActions' as a list of strings with S3 paths only.

B JOB_FLOW_OVERRIDES includes 'BootstrapActions' with 'ScriptBootstrapAction' containing 'Args' but no 'Path'.

C JOB_FLOW_OVERRIDES includes 'BootstrapActions' with 'ScriptBootstrapAction' containing 'Path' to S3 script.

D JOB_FLOW_OVERRIDES does not include 'BootstrapActions' but uses 'Steps' to run the bootstrap script.

💡 Hint

Bootstrap actions require a specific dictionary structure with 'Path' to the script.

The correct way to specify bootstrap actions is a list of dictionaries with 'Name' and 'ScriptBootstrapAction' keys, where 'ScriptBootstrapAction' must include the 'Path' to the S3 script. Missing 'Path' or wrong format causes failure.

❓ security

advanced

Which Airflow S3 operator configuration follows best security practices for AWS credentials?

You want to use S3ListOperator in Airflow to list objects in a bucket. Which configuration best follows AWS security best practices?

Apache Airflow

from airflow.providers.amazon.aws.operators.s3 import S3ListOperator

list_s3 = S3ListOperator(
    task_id='list_s3',
    bucket='my-secure-bucket',
    aws_conn_id='aws_default'
)

list_s3.execute(context={})

A Use aws_conn_id with full admin permissions for convenience.

B Use aws_conn_id that points to an IAM role with least privilege for S3 access.

C Hardcode AWS access key and secret key in the operator parameters.

D Use no aws_conn_id and rely on default environment variables with admin rights.

💡 Hint

Think about the principle of least privilege and secure credential management.

Best practice is to use an IAM role or user with only the permissions needed for the task. Hardcoding keys or using admin rights increases security risks.

✅ Best Practice

expert

What is the best approach to handle EMR cluster termination in Airflow to avoid extra costs?

You have an Airflow DAG that creates an EMR cluster, runs steps, and then should terminate the cluster. Which approach ensures the cluster is always terminated even if a step fails?

A Use <code>EmrTerminateJobFlowOperator</code> in a separate task with <code>trigger_rule='all_done'</code> to always run after steps.

B Terminate the cluster manually after the DAG finishes successfully.

C Set <code>KeepJobFlowAliveWhenNoSteps</code> to True and terminate cluster later.

D Do not terminate the cluster to reuse it for future jobs and save startup time.

💡 Hint

Consider Airflow task dependencies and trigger rules for cleanup tasks.

Using EmrTerminateJobFlowOperator with trigger_rule='all_done' ensures the cluster terminates regardless of previous task success or failure, preventing unexpected costs.