Pipeline orchestration with Airflow in Apache Spark - Step-by-Step Execution

Concept Flow - Pipeline orchestration with Airflow

Define DAG and tasks

↓

Set task dependencies

↓

Airflow Scheduler triggers DAG

↓

Tasks execute in order

↓

Monitor task status

↓

Handle failures or retries

↓

DAG run completes

Airflow runs a pipeline by defining tasks and their order, then scheduling and executing them step-by-step.

Execution Sample

Apache Spark

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

dag = DAG('example_dag', start_date=datetime(2024,1,1))

t1 = BashOperator(task_id='task1', bash_command='echo Task 1', dag=dag)
t2 = BashOperator(task_id='task2', bash_command='echo Task 2', dag=dag)
t1 >> t2

Defines two tasks in Airflow where task1 runs before task2.

Execution Table

Step	Action	Task	Status	Notes
1	DAG defined with tasks	example_dag	Defined	Tasks task1 and task2 created
2	Set dependency	task1 >> task2	Set	task2 waits for task1 completion
3	Scheduler triggers DAG run	example_dag	Running	Start executing tasks
4	Execute task1	task1	Success	Runs 'echo Task 1'
5	Execute task2	task2	Success	Runs 'echo Task 2' after task1
6	DAG run completes	example_dag	Success	All tasks finished successfully

💡 All tasks executed in order, DAG run finished successfully

Variable Tracker

Variable	Start	After Step 1	After Step 3	After Step 4	After Step 5	Final
DAG State	Not defined	Defined	Running	Running	Running	Success
task1 Status	Not created	Created	Waiting	Success	Success	Success
task2 Status	Not created	Created	Waiting	Waiting	Success	Success

Key Moments - 3 Insights

Why does task2 wait before running?

Because task2 depends on task1 (see execution_table step 2), Airflow enforces this order to ensure task1 finishes first.

What happens if task1 fails?

Airflow will mark task1 as failed and by default will not run task2, stopping the DAG run early (not shown here but important to know).

How does Airflow know when to start the DAG?

The scheduler triggers the DAG run based on the start_date and schedule (see execution_table step 3).

Visual Quiz - 3 Questions

Look at the execution table, what is the status of task1 after step 4?

A Failed

B Waiting

C Success

D Running

💡 Hint

Check the 'task1 Status' in variable_tracker after Step 4

At which step does the DAG start running tasks?

A Step 3

B Step 2

C Step 1

D Step 6

💡 Hint

Look at the 'DAG State' in variable_tracker and execution_table step descriptions

If task1 failed, what would happen to task2?

A task2 runs immediately

B task2 does not run

C task2 waits indefinitely

D task2 runs before task1

💡 Hint

Refer to key_moments about task failure handling

Concept Snapshot

Airflow pipelines are defined as DAGs with tasks.
Tasks run in order based on dependencies.
Scheduler triggers DAG runs automatically.
Tasks report success or failure.
Failures stop downstream tasks by default.

Full Transcript

Pipeline orchestration with Airflow means defining a Directed Acyclic Graph (DAG) of tasks that run in a specific order. We create tasks, set their dependencies, and Airflow's scheduler triggers the DAG to run. Each task executes its command, and the next task waits for the previous to finish. We track task statuses like success or failure. If a task fails, downstream tasks usually do not run. This process helps automate and monitor complex data workflows.

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