Pipeline orchestration with Airflow in Apache Spark - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

What is the output of this Airflow DAG schedule interval?

Consider this Airflow DAG snippet scheduling a Spark job. What is the effective schedule interval for the DAG?

Apache Spark

from airflow import DAG
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from datetime import datetime

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

dag = DAG(
    'spark_pipeline',
    default_args=default_args,
    schedule_interval='0 12 * * MON-FRI',
    catchup=False
)

spark_task = SparkSubmitOperator(
    task_id='run_spark_job',
    application='/path/to/spark_app.py',
    conn_id='spark_default',
    dag=dag
)

A Runs every hour on weekdays

B Runs every day at midnight

C Runs every Monday at 12:00 PM

D Runs every weekday at 12:00 PM

💡 Hint

Look at the cron expression '0 12 * * MON-FRI' and recall how cron schedules work.

The cron expression '0 12 * * MON-FRI' means the job runs at minute 0, hour 12 (noon), every day of the month, every month, but only on Monday through Friday.

So the DAG runs every weekday at 12:00 PM.

❓ data_output

intermediate

What is the output DataFrame schema after this Spark task in Airflow?

Given this Spark job code submitted via Airflow, what is the schema of the resulting DataFrame?

Apache Spark

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('TestApp').getOrCreate()
data = [
    (1, 'Alice', 29),
    (2, 'Bob', 31),
    (3, 'Cathy', 25)
]
columns = ['id', 'name', 'age']
df = spark.createDataFrame(data, columns)
df_filtered = df.filter(df.age > 28)
df_filtered.printSchema()

A

root
 |-- id: integer (nullable = true)
 |-- name: string (nullable = true)
 |-- age: integer (nullable = true)

B

root
 |-- id: integer (nullable = false)
 |-- name: string (nullable = false)
 |-- age: integer (nullable = false)

C

root
 |-- id: long (nullable = true)
 |-- name: string (nullable = true)
 |-- age: long (nullable = true)

D

root
 |-- id: string (nullable = true)
 |-- name: string (nullable = true)
 |-- age: string (nullable = true)

💡 Hint

Check the data types of the input tuples and how Spark infers schema from Python data.

Spark infers integer types for numeric literals in Python lists as IntegerType by default, and strings as StringType.

Since the data is a list of tuples with integers and strings, the schema will have integer types for 'id' and 'age', and string for 'name'.

Nullable is true by default when creating DataFrame from Python data.

🔧 Debug

advanced

Why does this Airflow SparkSubmitOperator task fail with 'FileNotFoundError'?

This Airflow DAG uses SparkSubmitOperator to run a Spark job. The task fails with FileNotFoundError. What is the most likely cause?

Apache Spark

spark_task = SparkSubmitOperator(
    task_id='run_spark_job',
    application='/home/airflow/dags/spark_jobs/my_spark_app.py',
    conn_id='spark_default',
    dag=dag
)

A The application path is incorrect or not accessible from the Spark cluster nodes

B The DAG start_date is in the future

C The conn_id 'spark_default' is missing in Airflow connections

D The SparkSubmitOperator requires a PythonOperator wrapper

💡 Hint

Think about where the Spark driver runs and where the application file must be located.

FileNotFoundError usually means the Spark cluster cannot find the application file at the given path.

In Airflow, the path must be accessible from the Spark cluster nodes, not just the Airflow scheduler machine.

🚀 Application

advanced

How to ensure task dependencies in an Airflow DAG running Spark jobs?

You have two SparkSubmitOperator tasks in an Airflow DAG: 'extract_data' and 'transform_data'. You want 'transform_data' to run only after 'extract_data' completes successfully. Which code snippet correctly sets this dependency?

A transform_data >> extract_data

B extract_data >> transform_data

C extract_data.set_downstream(transform_data)

D transform_data.set_upstream(extract_data)

💡 Hint

Recall the Airflow bitshift operators for setting task order.

In Airflow, 'task1 >> task2' means task1 runs before task2.

So 'extract_data >> transform_data' ensures 'transform_data' runs after 'extract_data'.

Option B is also valid but less commonly used; however, only one correct answer allowed, so we pick the bitshift operator which is clearer.

🧠 Conceptual

expert

What is the main benefit of using Airflow for orchestrating Spark pipelines?

Choose the best explanation for why Airflow is used to orchestrate Spark data pipelines.

A Airflow replaces the need for Spark clusters by running Spark code directly within its scheduler.

B Airflow automatically optimizes Spark job performance by tuning Spark configurations.

C Airflow provides a user-friendly interface to schedule, monitor, and manage complex Spark workflows with dependencies and retries.

D Airflow stores Spark data outputs in a distributed file system automatically.

💡 Hint

Think about what Airflow does versus what Spark does.

Airflow is a workflow orchestration tool that schedules and manages tasks, including Spark jobs.

It does not optimize Spark performance or replace Spark clusters.

It also does not handle data storage directly.

Practice

1. What is the main purpose of using Airflow in Apache Spark pipelines?

easy

A. To store large datasets

B. To write Spark SQL queries

C. To visualize Spark data

D. To schedule and manage the order of Spark jobs

Solution

1. Step 1: Understand Airflow's role

   Airflow is designed to schedule and orchestrate workflows, ensuring tasks run in the correct order.

2. Step 2: Compare with other options

   Writing Spark SQL, storing data, and visualizing data are not Airflow's main functions.

3. Final Answer:

   To schedule and manage the order of Spark jobs -> Option D

4. Quick Check:

   Airflow schedules pipelines = C [OK]

Hint: Airflow controls job order and timing [OK]

Common Mistakes:

• Confusing Airflow with Spark SQL or storage tools
• Thinking Airflow processes data directly
• Assuming Airflow is for data visualization

2. Which Airflow operator is commonly used to run a Spark job using spark-submit?

easy

A. BashOperator

B. PythonOperator

C. SparkOperator

D. EmailOperator

Solution

1. Step 1: Identify operator for shell commands

   BashOperator runs shell commands, perfect for spark-submit.

2. Step 2: Exclude other operators

   PythonOperator runs Python code, SparkOperator is not a default Airflow operator, EmailOperator sends emails.

3. Final Answer:

   BashOperator -> Option A

4. Quick Check:

   Run shell commands with BashOperator [OK]

Hint: Use BashOperator for shell commands like spark-submit [OK]

Common Mistakes:

• Choosing PythonOperator for shell commands
• Assuming SparkOperator exists by default
• Confusing EmailOperator with job execution

3. Given this Airflow DAG snippet, what is the order of task execution?

task1 = BashOperator(task_id='task1', bash_command='spark-submit job1.py', dag=dag)
task2 = BashOperator(task_id='task2', bash_command='spark-submit job2.py', dag=dag)
task3 = BashOperator(task_id='task3', bash_command='spark-submit job3.py', dag=dag)
task1 >> task2 >> task3

medium

A. task3 -> task2 -> task1

B. task1 -> task2 -> task3

C. All tasks run in parallel

D. task1 -> task3 -> task2

Solution

1. Step 1: Understand the dependency operator

   The operator >> means 'runs before', so task1 runs before task2, and task2 before task3.

2. Step 2: Determine execution order

   Tasks run sequentially: task1, then task2, then task3.

3. Final Answer:

   task1 -> task2 -> task3 -> Option B

4. Quick Check:

   Dependencies define order = D [OK]

Hint: Use >> to read task order left to right [OK]

Common Mistakes:

• Reading >> as reverse order
• Assuming tasks run in parallel without dependencies
• Mixing up task names and order

4. Identify the error in this Airflow DAG code snippet:

task1 = BashOperator(task_id='task1', bash_command='spark-submit job1.py', dag=dag)
task2 = BashOperator(task_id='task1', bash_command='spark-submit job2.py', dag=dag)
task1 >> task2

medium

A. Incorrect use of >> operator

B. Missing dag parameter in task2

C. Duplicate task_id 'task1' causes conflict

D. bash_command syntax is wrong

Solution

1. Step 1: Check task_id uniqueness

   Both tasks have the same task_id 'task1', which is not allowed in Airflow.

2. Step 2: Verify other parts

   dag parameter is present, >> operator is correctly used, bash_command syntax is valid.

3. Final Answer:

   Duplicate task_id 'task1' causes conflict -> Option C

4. Quick Check:

   Unique task_id needed = A [OK]

Hint: Each task_id must be unique in Airflow DAGs [OK]

Common Mistakes:

• Reusing task_id for different tasks
• Ignoring error messages about task_id
• Thinking >> operator causes error here

5. You want to run three Spark jobs in sequence using Airflow. Job 2 depends on Job 1, and Job 3 depends on Job 2. Which Airflow code snippet correctly sets this up?

hard

A. job1 = BashOperator(task_id='job1', bash_command='spark-submit job1.py', dag=dag)
job2 = BashOperator(task_id='job2', bash_command='spark-submit job2.py', dag=dag)
job3 = BashOperator(task_id='job3', bash_command='spark-submit job3.py', dag=dag)
job1.set_downstream(job2)
job2.set_downstream(job3)

B. job1 = BashOperator(task_id='job1', bash_command='spark-submit job1.py', dag=dag)
job2 = BashOperator(task_id='job2', bash_command='spark-submit job2.py', dag=dag)
job3 = BashOperator(task_id='job3', bash_command='spark-submit job3.py', dag=dag)
job3.set_downstream(job2)
job2.set_downstream(job1)

C. job1 = BashOperator(task_id='job1', bash_command='spark-submit job1.py', dag=dag)
job2 = BashOperator(task_id='job2', bash_command='spark-submit job2.py', dag=dag)
job3 = BashOperator(task_id='job3', bash_command='spark-submit job3.py', dag=dag)
job3 >> job2 >> job1

D. job1 = BashOperator(task_id='job1', bash_command='spark-submit job1.py', dag=dag)
job2 = BashOperator(task_id='job2', bash_command='spark-submit job2.py', dag=dag)
job3 = BashOperator(task_id='job3', bash_command='spark-submit job3.py', dag=dag)
job1 >> job3 >> job2

Solution

1. Step 1: Understand dependency direction

   job1 must run before job2, and job2 before job3, so job1 is upstream of job2, job2 upstream of job3.

2. Step 2: Check each option's dependency setup

   job1 = BashOperator(task_id='job1', bash_command='spark-submit job1.py', dag=dag)
   job2 = BashOperator(task_id='job2', bash_command='spark-submit job2.py', dag=dag)
   job3 = BashOperator(task_id='job3', bash_command='spark-submit job3.py', dag=dag)
   job1.set_downstream(job2)
   job2.set_downstream(job3)

   uses set_downstream correctly: job1 -> job2 -> job3. The other snippets reverse or mix the order incorrectly.

3. Final Answer:

   job1.set_downstream(job2); job2.set_downstream(job3) -> Option A

4. Quick Check:

   Use set_downstream for correct order = A [OK]

Hint: Use set_downstream or >> left to right for order [OK]

Common Mistakes:

• Reversing dependency direction
• Mixing task order in >> chains
• Confusing set_downstream with set_upstream