Apache Airflow Orchestration

Apache Airflow Orchestration

A comprehensive skill for mastering Apache Airflow workflow orchestration. This skill covers DAG development, operators, sensors, task dependencies, dynamic workflows, XCom communication, scheduling patterns, and production deployment strategies.

When to Use This Skill

Use this skill when:

• Building and managing complex data pipelines with task dependencies
• Orchestrating ETL/ELT workflows across multiple systems
• Scheduling and monitoring batch processing jobs
• Coordinating multi-step data transformations
• Managing workflows with conditional execution and branching
• Implementing event-driven or asset-based workflows
• Deploying production-grade workflow automation
• Creating dynamic workflows that generate tasks programmatically
• Coordinating distributed task execution across clusters
• Building data engineering platforms with workflow orchestration

Core Concepts

What is Apache Airflow?

Apache Airflow is an open-source platform for programmatically authoring, scheduling, and monitoring workflows. It allows you to define workflows as Directed Acyclic Graphs (DAGs) using Python code, making complex workflow orchestration maintainable and version-controlled.

Key Principles:

• Dynamic: Workflows are defined in Python, enabling dynamic generation
• Extensible: Rich ecosystem of operators, sensors, and hooks
• Scalable: Can scale from single machine to large clusters
• Observable: Comprehensive UI for monitoring and troubleshooting

DAGs (Directed Acyclic Graphs)

A DAG is a collection of tasks organized to reflect their relationships and dependencies.

DAG Properties:

• dag_id: Unique identifier for the DAG
• start_date: When the DAG should start being scheduled
• schedule: How often to run (cron, timedelta, or asset-based)
• catchup: Whether to run missed intervals on DAG activation
• tags: Labels for organization and filtering
• default_args: Default parameters for all tasks in the DAG

DAG Definition Example:

from datetime import datetime
from airflow.sdk import DAG

with DAG(
    dag_id="example_dag",
    start_date=datetime(2022, 1, 1),
    schedule="0 0 * * *",  # Daily at midnight
    catchup=False,
    tags=["example", "tutorial"],
) as dag:
    # Tasks defined here
    pass

Tasks and Operators

Tasks are the basic units of execution in Airflow. Operators are templates for creating tasks.

Common Operator Types:

1. BashOperator: Execute bash commands
2. PythonOperator: Execute Python functions
3. EmailOperator: Send emails
4. EmptyOperator: Placeholder/dummy tasks
5. Custom Operators: User-defined operators for specific needs

Operator vs. Task:

• Operator: Template/class definition
• Task: Instantiation of an operator with specific parameters

Task Dependencies

Task dependencies define the execution order and workflow structure.

Dependency Operators:

• >>: Sets downstream dependency (task1 >> task2)
• <<: Sets upstream dependency (task2 << task1)
• chain(): Sequential dependencies for multiple tasks
• cross_downstream(): Many-to-many relationships

Dependency Examples:

# Simple linear flow
task1 >> task2 >> task3

# Fan-out pattern
task1 >> [task2, task3, task4]

# Fan-in pattern
[task1, task2, task3] >> task4

# Complex dependencies
first_task >> [second_task, third_task]
third_task << fourth_task

Executors

Executors determine how and where tasks run.

Executor Types:

• SequentialExecutor: Single-threaded, local (default, not for production)
• LocalExecutor: Multi-threaded, single machine
• CeleryExecutor: Distributed execution using Celery
• KubernetesExecutor: Each task runs in a separate Kubernetes pod
• DaskExecutor: Distributed execution using Dask

Scheduler

The Airflow scheduler:

• Monitors all DAGs and their tasks
• Triggers task instances based on dependencies and schedules
• Submits tasks to executors for execution
• Handles retries and task state management

Starting the Scheduler:

airflow scheduler

DAG Development Patterns

Basic DAG Structure

Every DAG follows this structure:

from datetime import datetime
from airflow.sdk import DAG
from airflow.providers.standard.operators.bash import BashOperator

with DAG(
    dag_id="basic_dag",
    start_date=datetime(2022, 1, 1),
    schedule="0 0 * * *",
    catchup=False,
) as dag:

    task1 = BashOperator(
        task_id="task1",
        bash_command="echo 'Task 1 executed'"
    )

    task2 = BashOperator(
        task_id="task2",
        bash_command="echo 'Task 2 executed'"
    )

    task1 >> task2

Task Dependencies and Chains

Linear Chain:

from airflow.sdk import chain

# These are equivalent:
task1 >> task2 >> task3 >> task4
chain(task1, task2, task3, task4)

Dynamic Chain:

from airflow.sdk import chain
from airflow.operators.empty import EmptyOperator

# Dynamically generate and chain tasks
chain(*[EmptyOperator(task_id=f"task_{i}") for i in range(1, 6)])

Pairwise Chain:

from airflow.sdk import chain

# Creates paired dependencies:
# op1 >> op2 >> op4 >> op6
# op1 >> op3 >> op5 >> op6
chain(op1, [op2, op3], [op4, op5], op6)

Cross Downstream:

from airflow.sdk import cross_downstream

# Both op1 and op2 feed into both op3 and op4
cross_downstream([op1, op2], [op3, op4])

Branching and Conditional Execution

BranchPythonOperator:

from airflow.operators.python import BranchPythonOperator

def choose_branch(**context):
    if context['data_interval_start'].day == 1:
        return 'monthly_task'
    return 'daily_task'

branch = BranchPythonOperator(
    task_id='branch_task',
    python_callable=choose_branch
)

daily_task = BashOperator(task_id='daily_task', bash_command='echo daily')
monthly_task = BashOperator(task_id='monthly_task', bash_command='echo monthly')

branch >> [daily_task, monthly_task]

Custom Branch Operator:

from airflow.operators.branch import BaseBranchOperator

class MyBranchOperator(BaseBranchOperator):
    def choose_branch(self, context):
        """
        Run extra branch on first day of month
        """
        if context['data_interval_start'].day == 1:
            return ['daily_task_id', 'monthly_task_id']
        elif context['data_interval_start'].day == 2:
            return 'daily_task_id'
        else:
            return None  # Skip all downstream tasks

TaskGroups for Organization

TaskGroups help organize related tasks hierarchically:

from airflow.sdk import task_group
from airflow.operators.empty import EmptyOperator

@task_group()
def data_processing_group():
    extract = EmptyOperator(task_id="extract")
    transform = EmptyOperator(task_id="transform")
    load = EmptyOperator(task_id="load")

    extract >> transform >> load

@task_group()
def validation_group():
    validate_schema = EmptyOperator(task_id="validate_schema")
    validate_data = EmptyOperator(task_id="validate_data")

    validate_schema >> validate_data

start = EmptyOperator(task_id="start")
end = EmptyOperator(task_id="end")

start >> data_processing_group() >> validation_group() >> end

Edge Labeling

Add labels to dependency edges for clarity:

from airflow.sdk import Label

# Inline labeling
my_task >> Label("When empty") >> other_task

# Method-based labeling
my_task.set_downstream(other_task, Label("When empty"))

LatestOnlyOperator

Skip tasks if not the latest DAG run:

from airflow.operators.latest_only import LatestOnlyOperator
from airflow.operators.empty import EmptyOperator
import pendulum

with DAG(
    dag_id='latest_only_example',
    start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
    catchup=True,
    schedule="@daily",
) as dag:
    latest_only = LatestOnlyOperator(task_id='latest_only')