Pipeline orchestration with Airflow in Apache Spark - Mini Project: Build & Apply

Pipeline orchestration with Airflow

📖 Scenario: You work as a data engineer at a company that processes daily sales data using Apache Spark. You want to automate the process of running your Spark job every day using Apache Airflow. This project will guide you through creating a simple Airflow DAG to orchestrate your Spark job.

🎯 Goal: Build an Airflow DAG that runs a Spark job daily. The DAG will have a single task that submits a Spark job using the SparkSubmitOperator. You will set up the DAG, configure the Spark job parameters, and print the DAG run status.

📋 What You'll Learn

Create a Python dictionary called default_args with Airflow DAG default parameters.

Create a variable called dag that defines the Airflow DAG with the given default_args and a daily schedule.

Create a task called spark_task using SparkSubmitOperator with specified Spark job parameters.

Print the task ID of the spark_task to confirm the DAG setup.

💡 Why This Matters

🌍 Real World

Automating data processing pipelines is essential in data engineering. Airflow helps schedule and manage workflows, while Spark handles big data processing efficiently.

💼 Career

Data engineers and data scientists use Airflow and Spark together to build reliable, automated data pipelines that run on schedule without manual intervention.

1

Set up Airflow DAG default arguments

Create a dictionary called default_args with these exact entries: 'owner': 'airflow', 'start_date': datetime(2024, 1, 1), and 'retries': 1. Import datetime from the datetime module.

Apache Spark

from datetime import datetime

# Create default_args dictionary with owner, start_date, and retries
# Your code here

Hint

Use datetime(2024, 1, 1) for the start date and set retries to 1.

2

Define the Airflow DAG

Import DAG from airflow. Create a variable called dag that defines a DAG with the ID 'spark_daily_job', uses default_args, and has a schedule interval of '@daily'.

Apache Spark

from datetime import datetime
from airflow import DAG

# Create the DAG with id 'spark_daily_job', default_args, and schedule '@daily'
# Your code here

Hint

Use DAG('spark_daily_job', default_args=default_args, schedule_interval='@daily').

3

Create the SparkSubmitOperator task

Import SparkSubmitOperator from airflow.providers.apache.spark.operators.spark_submit. Create a task called spark_task using SparkSubmitOperator with these exact parameters: task_id='run_spark_job', application='/path/to/spark_job.py', and dag=dag.

Apache Spark

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

# Create spark_task using SparkSubmitOperator with task_id 'run_spark_job', application '/path/to/spark_job.py', and dag
# Your code here

Hint

Use SparkSubmitOperator(task_id='run_spark_job', application='/path/to/spark_job.py', dag=dag).

4

Print the Spark task ID

Write a print statement to display the task_id of the spark_task.

Apache Spark

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

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

dag = DAG(
    'spark_daily_job',
    default_args=default_args,
    schedule_interval='@daily'
)

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

# Print the task_id of spark_task
# Your code here

Hint

Use print(spark_task.task_id) to show the task ID.

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