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')
    task1 = EmptyOperator(task_id='task1')
    task2 = EmptyOperator(task_id='task2')
    task3 = EmptyOperator(task_id='task3')
    task4 = EmptyOperator(task_id='task4', trigger_rule='all_done')

    latest_only >> task1 >> task3
    latest_only >> task4
    task2 >> task3
    task2 >> task4

Operators Deep Dive

BashOperator

Execute bash commands:

from airflow.providers.standard.operators.bash import BashOperator

bash_task = BashOperator(
    task_id="bash_example",
    bash_command="echo 'Hello from Bash'; date",
    env={'MY_VAR': 'value'},  # Environment variables
    append_env=True,  # Append to existing env vars
    output_encoding='utf-8'
)

Complex Bash Command:

bash_complex = BashOperator(
    task_id="complex_bash",
    bash_command="""
        cd /path/to/dir
        python process_data.py --input {{ ds }} --output {{ tomorrow_ds }}
        if [ $? -eq 0 ]; then
            echo "Success"
        else
            echo "Failed" && exit 1
        fi
    """,
)

PythonOperator

Execute Python functions:

from airflow.providers.standard.operators.python import PythonOperator

def my_python_function(name, **context):
    print(f"Hello {name}!")
    print(f"Execution date: {context['ds']}")
    return "Success"

python_task = PythonOperator(
    task_id="python_example",
    python_callable=my_python_function,
    op_kwargs={'name': 'Airflow'},
    provide_context=True
)

Traditional ETL with PythonOperator:

import json
import pendulum
from airflow.sdk import DAG
from airflow.providers.standard.operators.python import PythonOperator

def extract():
    data_string = '{"1001": 301.27, "1002": 433.21, "1003": 502.22}'
    return json.loads(data_string)

def transform(ti):
    # Pull from XCom
    order_data_dict = ti.xcom_pull(task_ids="extract")
    total_order_value = sum(order_data_dict.values())
    return {"total_order_value": total_order_value}

def load(ti):
    # Pull from XCom
    total = ti.xcom_pull(task_ids="transform")["total_order_value"]
    print(f"Total order value is: {total:.2f}")

with DAG(
    dag_id="legacy_etl_pipeline",
    schedule=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
) as dag:
    extract_task = PythonOperator(task_id="extract", python_callable=extract)
    transform_task = PythonOperator(task_id="transform", python_callable=transform)
    load_task = PythonOperator(task_id="load", python_callable=load)

    extract_task >> transform_task >> load_task

EmailOperator

Send email notifications:

from airflow.providers.smtp.operators.smtp import EmailOperator

email_task = EmailOperator(
    task_id='send_email',
    to='recipient@example.com',
    subject='Airflow Notification',
    html_content='<h3>Task completed successfully!</h3>',
    cc=['cc@example.com'],
    bcc=['bcc@example.com']
)

EmptyOperator

Placeholder for workflow structure:

from airflow.operators.empty import EmptyOperator

start = EmptyOperator(task_id='start')
end = EmptyOperator(task_id='end')

# Useful for organizing complex DAGs
start >> [task1, task2, task3] >> end

Custom Operators

Create reusable custom operators:

from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults

class MyCustomOperator(BaseOperator):
    @apply_defaults
    def __init__(self, my_param, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.my_param = my_param

    def execute(self, context):
        self.log.info(f"Executing with param: {self.my_param}")
        # Custom logic here
        return "Result"

# Usage
custom_task = MyCustomOperator(
    task_id="custom",
    my_param="value"
)

Sensors Deep Dive

Sensors are a special type of operator that wait for a certain condition to be met before proceeding.

ExternalTaskSensor

Wait for tasks in other DAGs:

from airflow.providers.standard.sensors.external_task import ExternalTaskSensor
import pendulum

with DAG(
    dag_id="example_external_task_sensor",
    start_date=pendulum.datetime(2021, 10, 20, tz="UTC"),
    catchup=False,
    schedule=None,
) as dag:

    wait_for_task = ExternalTaskSensor(
        task_id="wait_for_task",
        external_dag_id="upstream_dag",
        external_task_id="upstream_task",
        allowed_states=["success"],
        failed_states=["failed"],
        execution_delta=None,  # Same execution_date
        timeout=600,  # 10 minutes
        poke_interval=60,  # Check every 60 seconds
    )

Deferrable ExternalTaskSensor:

# More efficient - releases worker slot while waiting
wait_for_task_async = ExternalTaskSensor(
    task_id="wait_for_task_async",
    external_dag_id="upstream_dag",
    external_task_id="upstream_task",
    allowed_states=["success"],
    failed_states=["failed"],
    deferrable=True,  # Use async mode
)

FileSensor

Wait for files to appear:

from airflow.sensors.filesystem import FileSensor

wait_for_file = FileSensor(
    task_id="wait_for_file",
    filepath="/path/to/file.csv",
    poke_interval=30,
    timeout=600,
    mode='poke'  # or 'reschedule' for long waits
)

TimeDeltaSensor

Wait for a specific time period:

from datetime import timedelta
from airflow.sensors.time_delta import TimeDeltaSensor

wait_one_hour = TimeDeltaSensor(
    task_id="wait_one_hour",
    delta=timedelta(hours=1)
)

BigQuery Table Sensor

Wait for BigQuery table to exist:

from airflow.providers.google.cloud.sensors.bigquery import BigQueryTableExistenceSensor
import pendulum

with DAG(
    dag_id="bigquery_sensor_example",
    start_date=pendulum.datetime(2023, 10, 26, tz="UTC"),
) as dag:

    wait_for_table = BigQueryTableExistenceSensor(
        task_id="wait_for_table",
        project_id="your-project-id",
        dataset_id="your_dataset",
        table_id="your_table",
        bigquery_conn_id="google_cloud_default",
        location="US",
        poke_interval=60,
        timeout=3600,
    )

Custom Sensors

Create custom sensors for specific conditions:

from airflow.sensors.base import BaseSensorOperator
from airflow.utils.decorators import apply_defaults

class MyCustomSensor(BaseSensorOperator):
    @apply_defaults
    def __init__(self, my_condition, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.my_condition = my_condition

    def poke(self, context):
        # Return True when condition is met
        self.log.info(f"Checking condition: {self.my_condition}")
        # Custom logic to check condition
        return check_condition(self.my_condition)

Deferrable Sensors

Deferrable sensors release worker slots while waiting:

from datetime import timedelta
from airflow.sdk import BaseSensorOperator, StartTriggerArgs

class WaitHoursSensor(BaseSensorOperator):
    start_trigger_args = StartTriggerArgs(
        trigger_cls="airflow.providers.standard.triggers.temporal.TimeDeltaTrigger",
        trigger_kwargs={"moment": timedelta(hours=1)},
        next_method="execute_complete",
        next_kwargs=None,
        timeout=None,
    )
    start_from_trigger = True

    def __init__(self, *args, trigger_kwargs=None, start_from_trigger=True, **kwargs):
        super().__init__(*args, **kwargs)
        if trigger_kwargs:
            self.start_trigger_args.trigger_kwargs = trigger_kwargs
        self.start_from_trigger = start_from_trigger

    def execute_complete(self, context, event=None):
        return  # Task complete

XComs (Cross-Communication)

XComs enable task-to-task communication by storing and retrieving data.

Basic XCom Usage

Pushing to XCom:

def push_function(**context):
    value = "Important data"
    context['ti'].xcom_push(key='my_key', value=value)
    # Or simply return (uses 'return_value' key)
    return value

push_task = PythonOperator(
    task_id='push',
    python_callable=push_function,
    provide_context=True
)

Pulling from XCom:

def pull_function(**context):
    # Pull by task_id (uses 'return_value' key)
    value = context['ti'].xcom_pull(task_ids='push')

    # Pull with specific key
    value = context['ti'].xcom_pull(task_ids='push', key='my_key')

    print(f"Pulled value: {value}")

pull_task = PythonOperator(
    task_id='pull',
    python_callable=pull_function,
    provide_context=True
)

XCom with TaskFlow API

TaskFlow API automatically manages XComs:

from airflow.decorators import task

@task
def extract():
    return {"data": [1, 2, 3, 4, 5]}

@task
def transform(data_dict):
    # Automatically receives XCom from extract
    total = sum(data_dict['data'])
    return {"total": total}

@task
def load(summary):
    print(f"Total: {summary['total']}")

# Automatic XCom handling
data = extract()
summary = transform(data)
load(summary)

XCom Best Practices

Size Limitations:

• XComs are stored in the metadata database
• Keep XCom data small (< 1MB recommended)
• For large data, store in external systems and pass references

Example with External Storage:

@task
def process_large_data():
    # Process data
    large_result = compute_large_dataset()

    # Store in S3/GCS
    file_path = save_to_s3(large_result, "s3://bucket/result.parquet")

    # Return only the path
    return {"result_path": file_path}

@task
def consume_large_data(metadata):
    # Load from S3/GCS
    data = load_from_s3(metadata['result_path'])
    process(data)

XCom with Operators

Reading XCom in Templates:

from airflow.providers.standard.operators.bash import BashOperator

process_file = BashOperator(
    task_id="process",
    bash_command="python process.py {{ ti.xcom_pull(task_ids='extract') }}",
)

XCom with EmailOperator:

from airflow.sdk import task
from airflow.providers.smtp.operators.smtp import EmailOperator

@task
def get_ip():
    return "192.168.1.1"

@task(multiple_outputs=True)
def compose_email(external_ip):
    return {
        'subject': f'Server connected from {external_ip}',
        'body': f'Your server is connected from {external_ip}<br>'
    }

email_info = compose_email(get_ip())

EmailOperator(
    task_id='send_email',
    to='example@example.com',
    subject=email_info['subject'],
    html_content=email_info['body']
)

Dynamic Workflows

Create tasks dynamically based on runtime conditions or external data.

Dynamic Task Generation with Loops

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

with DAG("dynamic_loop_example", ...) as dag:
    start = EmptyOperator(task_id="start")
    end = EmptyOperator(task_id="end")

    # Dynamically create tasks
    options = ["branch_a", "branch_b", "branch_c", "branch_d"]
    for option in options:
        task = EmptyOperator(task_id=option)
        start >> task >> end

Dynamic Task Mapping

Map over task outputs to create dynamic parallel tasks:

from airflow.decorators import task

@task
def extract():
    # Returns list of items to process
    return [1, 2, 3, 4, 5]

@task
def transform(item):
    # Processes single item
    return item * 2

@task
def load(items):
    # Receives all transformed items
    print(f"Loaded {len(items)} items: {items}")

# Dynamic mapping
data = extract()
transformed = transform.expand(item=data)  # Creates 5 parallel tasks
load(transformed)

Mapping with Classic Operators:

from airflow.operators.bash import BashOperator

class ExtractOperator(BaseOperator):
    def execute(self, context):
        return ["file1.csv", "file2.csv", "file3.csv"]

class TransformOperator(BaseOperator):
    def __init__(self, input, **kwargs):
        super().__init__(**kwargs)
        self.input = input

    def execute(self, context):
        # Process single file
        return f"processed_{self.input}"

extract = ExtractOperator(task_id="extract")
transform = TransformOperator.partial(task_id="transform").expand(input=extract.output)

Task Group Mapping

Map over entire task groups:

from airflow.decorators import task, task_group

@task
def add_one(value):
    return value + 1

@task
def double(value):
    return value * 2

@task_group
def process_group(value):
    incremented = add_one(value)
    return double(incremented)

@task
def aggregate(results):
    print(f"Results: {results}")

# Map task group over values
results = process_group.expand(value=[1, 2, 3, 4, 5])
aggregate(results)

Partial Parameters with Mapping

Mix static and dynamic parameters:

@task
def process(base_path, filename):
    full_path = f"{base_path}/{filename}"
    return f"Processed {full_path}"

# Static parameter 'base_path', dynamic 'filename'
results = process.partial(base_path="/data").expand(
    filename=["file1.csv", "file2.csv", "file3.csv"]
)

TaskFlow API

The modern way to write Airflow DAGs with automatic XCom handling and cleaner syntax.

Basic TaskFlow Example

from airflow.decorators import dag, task
import pendulum

@dag(
    dag_id="taskflow_example",
    start_date=pendulum.datetime(2023, 10, 26, tz="UTC"),
    schedule=None,
    catchup=False,
)
def my_taskflow_dag():

    @task
    def extract():
        data_string = '{"1001": 301.27, "1002": 433.21, "1003": 502.22}'
        import json
        return json.loads(data_string)

    @task
    def transform(order_data_dict):
        total = sum(order_data_dict.values())
        return {"total_order_value": total}

    @task
    def load(summary):
        print(f"Total order value: {summary['total_order_value']:.2f}")

    # Function calls create task dependencies automatically
    order_data = extract()
    summary = transform(order_data)
    load(summary)

# Instantiate the DAG
my_taskflow_dag()

Multiple Outputs

Return multiple values from tasks:

@task(multiple_outputs=True)
def extract_data():
    return {
        'orders': [1, 2, 3],
        'customers': ['A', 'B', 'C'],
        'revenue': 1000.50
    }

@task
def process_orders(orders):
    print(f"Processing {len(orders)} orders")

@task
def process_customers(customers):
    print(f"Processing {len(customers)} customers")

# Access individual outputs
data = extract_data()
process_orders(data['orders'])
process_customers(data['customers'])

Mixing TaskFlow with Traditional Operators

from airflow.decorators import dag, task
from airflow.providers.standard.operators.bash import BashOperator
import pendulum

@dag(
    start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
    schedule=None,
)
def mixed_dag():

    @task
    def get_date():
        from datetime import datetime
        return datetime.now().strftime("%Y%m%d")

    # Traditional operator
    bash_task = BashOperator(
        task_id="print_date",
        bash_command="echo Processing data for {{ ti.xcom_pull(task_ids='get_date') }}"
    )

    @task
    def process_results():
        print("Processing complete")

    # Mix task types
    date = get_date()
    date >> bash_task >> process_results()

mixed_dag()

Virtual Environment for Tasks

Isolate task dependencies:

from airflow.decorators import dag, task
import pendulum

@dag(
    dag_id="virtualenv_example",
    start_date=pendulum.datetime(2023, 10, 26, tz="UTC"),
)
def virtualenv_dag():

    @task.virtualenv(
        requirements=["pandas==1.5.0", "numpy==1.23.0"],
        system_site_packages=False
    )
    def analyze_data():
        import pandas as pd
        import numpy as np

        df = pd.DataFrame({'col1': [1, 2, 3], 'col2': [4, 5, 6]})
        result = np.mean(df['col1'])
        return float(result)

    @task
    def report_results(mean_value):
        print(f"Mean value: {mean_value}")

    result = analyze_data()
    report_results(result)

virtualenv_dag()

Asset-Based Scheduling

Schedule DAGs based on data assets (formerly datasets) rather than time.

Producer-Consumer Pattern

from airflow.sdk import DAG, Asset
from airflow.operators.bash import BashOperator
from datetime import datetime

# Define asset
customer_data = Asset("s3://my-bucket/customers.parquet")

# Producer DAG
with DAG(
    dag_id="producer_dag",
    start_date=datetime(2023, 1, 1),
    schedule="@daily",
) as producer:

    BashOperator(
        task_id="generate_data",
        bash_command="python generate_customers.py",
        outlets=[customer_data]  # Marks asset as updated
    )

# Consumer DAG - triggered when asset updates
with DAG(
    dag_id="consumer_dag",
    schedule=[customer_data],  # Triggered by asset
    start_date=datetime(2023, 1, 1),
    catchup=False,
) as consumer:

    BashOperator(
        task_id="process_data",
        bash_command="python process_customers.py"
    )

Multiple Asset Dependencies

AND Logic (all assets must update):

from airflow.datasets import Dataset

asset_1 = Dataset("s3://bucket/file1.csv")
asset_2 = Dataset("s3://bucket/file2.csv")

with DAG(
    dag_id="wait_for_both",
    schedule=[asset_1 & asset_2],  # Both must update
    start_date=datetime(2023, 1, 1),
):
    pass

OR Logic (any asset update triggers):

asset_1 = Dataset("s3://bucket/file1.csv")
asset_2 = Dataset("s3://bucket/file2.csv")

with DAG(
    dag_id="triggered_by_either",
    schedule=[asset_1 | asset_2],  # Either can trigger
    start_date=datetime(2023, 1, 1),
):
    pass

Complex Logic:

asset_1 = Dataset("s3://bucket/file1.csv")
asset_2 = Dataset("s3://bucket/file2.csv")
asset_3 = Dataset("s3://bucket/file3.csv")

with DAG(
    dag_id="complex_condition",
    schedule=(asset_1 | (asset_2 & asset_3)),  # asset_1 OR (asset_2 AND asset_3)
    start_date=datetime(2023, 1, 1),
):
    pass

Asset Aliases

Use aliases for flexible asset references:

from airflow.datasets import Dataset, AssetAlias
from airflow.decorators import task

# Producer with alias
with DAG(dag_id="alias_producer", start_date=datetime(2023, 1, 1)):
    @task(outlets=[AssetAlias("my-alias")])
    def produce_data(*, outlet_events):
        # Dynamically add actual asset
        outlet_events[AssetAlias("my-alias")].add(
            Dataset("s3://bucket/my-file.csv")
        )

# Consumer depending on alias
with DAG(
    dag_id="alias_consumer",
    schedule=AssetAlias("my-alias"),
    start_date=datetime(2023, 1, 1),
):
    pass

Accessing Asset Event Information

@task
def process_asset_data(*, triggering_asset_events):
    for event in triggering_asset_events:
        print(f"Asset: {event.asset.uri}")
        print(f"Timestamp: {event.timestamp}")
        print(f"Extra: {event.extra}")

Scheduling Patterns

Cron Expressions

# Every day at midnight
schedule="0 0 * * *"

# Every Monday at 9 AM
schedule="0 9 * * 1"

# Every 15 minutes
schedule="*/15 * * * *"

# First day of month at noon
schedule="0 12 1 * *"

# Weekdays at 6 PM
schedule="0 18 * * 1-5"

Timedelta Scheduling

from datetime import timedelta

with DAG(
    dag_id="timedelta_schedule",
    start_date=datetime(2023, 1, 1),
    schedule=timedelta(hours=6),  # Every 6 hours
):
    pass

Preset Schedules

# Common presets
schedule="@once"      # Run once
schedule="@hourly"    # Every hour
schedule="@daily"     # Daily at midnight
schedule="@weekly"    # Every Sunday at midnight
schedule="@monthly"   # First day of month at midnight
schedule="@yearly"    # January 1st at midnight
schedule=None         # Manual trigger only

Catchup and Backfilling

Catchup:

with DAG(
    dag_id="catchup_example",
    start_date=datetime(2023, 1, 1),
    schedule="@daily",
    catchup=True,  # Run all missed intervals
):
    pass

with DAG(
    dag_id="no_catchup",
    start_date=datetime(2023, 1, 1),
    schedule="@daily",
    catchup=False,  # Only run latest interval
):
    pass

Manual Backfilling:

# Backfill specific date range
airflow dags backfill \
    --start-date 2023-01-01 \
    --end-date 2023-01-31 \
    my_dag_id

# Backfill with marking success (no execution)
airflow dags backfill \
    --start-date 2023-01-01 \
    --end-date 2023-01-31 \
    --mark-success \
    my_dag_id

Production Patterns

Error Handling and Retries

Task-Level Retries:

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

task_with_retry = BashOperator(
    task_id="retry_task",
    bash_command="python might_fail.py",
    retries=3,
    retry_delay=timedelta(minutes=5),
    retry_exponential_backoff=True,
    max_retry_delay=timedelta(minutes=30),
)

DAG-Level Default Args:

from datetime import datetime, timedelta

default_args = {
    'owner': 'data-team',
    'depends_on_past': False,
    'email': ['alerts@company.com'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    dag_id="production_dag",
    default_args=default_args,
    start_date=datetime(2023, 1, 1),
    schedule="@daily",
):
    pass

Task Concurrency Control

Per-Task Concurrency:

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

# Limit concurrent instances of this task
limited_task = BashOperator(
    task_id="limited_task",
    bash_command="echo 'Processing'",
    max_active_tis_per_dag=3  # Max 3 instances running
)

DAG-Level Concurrency:

with DAG(
    dag_id="concurrent_dag",
    start_date=datetime(2023, 1, 1),
    schedule="@daily",
    max_active_runs=5,  # Max 5 DAG runs simultaneously
    concurrency=10,     # Max 10 task instances across all runs
):
    pass

Idempotency

Make tasks idempotent for safe retries:

@task
def idempotent_load(**context):
    execution_date = context['ds']

    # Delete existing data for this date first
    delete_query = f"""
        DELETE FROM target_table
        WHERE date = '{execution_date}'
    """
    execute_sql(delete_query)

    # Insert new data
    insert_query = f"""
        INSERT INTO target_table
        SELECT * FROM source
        WHERE date = '{execution_date}'
    """
    execute_sql(insert_query)

SLAs and Alerts

from datetime import timedelta

def sla_miss_callback(dag, task_list, blocking_task_list, slas, blocking_tis):
    print(f"SLA missed for {task_list}")
    # Send alert to monitoring system

with DAG(
    dag_id="sla_dag",
    start_date=datetime(2023, 1, 1),
    schedule="@daily",
    default_args={
        'sla': timedelta(hours=2),  # Task should complete in 2 hours
    },
    sla_miss_callback=sla_miss_callback,
):
    pass

Task Callbacks

def on_failure_callback(context):
    print(f"Task {context['task_instance'].task_id} failed")
    # Send to Slack, PagerDuty, etc.

def on_success_callback(context):
    print(f"Task {context['task_instance'].task_id} succeeded")

def on_retry_callback(context):
    print(f"Task {context['task_instance'].task_id} retrying")

task_with_callbacks = BashOperator(
    task_id="monitored_task",
    bash_command="python my_script.py",
    on_failure_callback=on_failure_callback,
    on_success_callback=on_success_callback,
    on_retry_callback=on_retry_callback,
)

Docker Deployment

Docker Compose for Local Development:

version: '3'
services:
  postgres:
    image: postgres:13
    environment:
      POSTGRES_USER: airflow
      POSTGRES_PASSWORD: airflow
      POSTGRES_DB: airflow

  webserver:
    image: apache/airflow:2.7.0
    depends_on:
      - postgres
    environment:
      AIRFLOW__CORE__EXECUTOR: LocalExecutor
      AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
    ports:
      - "8080:8080"
    command: webserver

  scheduler:
    image: apache/airflow:2.7.0
    depends_on:
      - postgres
    environment:
      AIRFLOW__CORE__EXECUTOR: LocalExecutor
      AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
    command: scheduler

Kubernetes Executor

KubernetesExecutor Configuration:

# In airflow.cfg
[kubernetes]
namespace = airflow
worker_container_repository = my-registry/airflow
worker_container_tag = 2.7.0
delete_worker_pods = True
delete_worker_pods_on_failure = False

[core]
executor = KubernetesExecutor

Pod Override for Specific Task:

from kubernetes.client import models as k8s

task_with_gpu = BashOperator(
    task_id="gpu_task",
    bash_command="python train_model.py",
    executor_config={
        "pod_override": k8s.V1Pod(
            spec=k8s.V1PodSpec(
                containers=[
                    k8s.V1Container(
                        name="base",
                        resources=k8s.V1ResourceRequirements(
                            limits={"nvidia.com/gpu": "1"}
                        )
                    )
                ]
            )
        )
    }
)

Monitoring and Logging

Structured Logging:

from airflow.decorators import task
import logging

@task
def monitored_task():
    logger = logging.getLogger(__name__)

    logger.info("Starting data processing", extra={
        'process_id': 'abc123',
        'record_count': 1000
    })

    try:
        process_data()
        logger.info("Processing complete")
    except Exception as e:
        logger.error(f"Processing failed: {str(e)}", extra={
            'error_type': type(e).__name__
        })
        raise

StatsD Metrics:

from airflow.stats import Stats

@task
def task_with_metrics():
    Stats.incr('my_dag.task_started')

    start_time = time.time()
    process_data()
    duration = time.time() - start_time

    Stats.timing('my_dag.task_duration', duration)
    Stats.incr('my_dag.task_completed')

Best Practices

DAG Design

1. Keep DAGs Simple: Break complex workflows into multiple DAGs 2. Use Descriptive Names: dag_id and task_id should be self-explanatory 3. Idempotent Tasks: Tasks should produce same result when re-run 4. Small XComs: Keep XCom data under 1MB 5. External Storage: Use S3/GCS for large data, pass references 6. Proper Dependencies: Model true dependencies, avoid unnecessary ones 7. Error Handling: Use retries, callbacks, and proper error logging 8. Resource Management: Set appropriate task concurrency limits

Code Organization

dags/
├── common/
│   ├── __init__.py
│   ├── operators.py      # Custom operators
│   ├── sensors.py        # Custom sensors
│   └── utils.py          # Utility functions
├── etl/
│   ├── customer_pipeline.py
│   ├── order_pipeline.py
│   └── product_pipeline.py
├── ml/
│   ├── training_dag.py
│   └── inference_dag.py
└── maintenance/
    ├── cleanup_dag.py
    └── backup_dag.py

Testing DAGs

Unit Testing:

import pytest
from airflow.models import DagBag

def test_dag_loaded():
    dagbag = DagBag(dag_folder='dags/', include_examples=False)
    assert len(dagbag.import_errors) == 0

def test_task_count():
    dagbag = DagBag(dag_folder='dags/')
    dag = dagbag.get_dag('my_dag')
    assert len(dag.tasks) == 5

def test_task_dependencies():
    dagbag = DagBag(dag_folder='dags/')
    dag = dagbag.get_dag('my_dag')

    extract = dag.get_task('extract')
    transform = dag.get_task('transform')

    assert transform in extract.downstream_list

Integration Testing:

from airflow.models import DagBag
from airflow.utils.state import State

def test_dag_runs():
    dagbag = DagBag(dag_folder='dags/')
    dag = dagbag.get_dag('my_dag')

    # Test DAG run
    dag_run = dag.create_dagrun(
        state=State.RUNNING,
        execution_date=datetime(2023, 1, 1),
        run_type='manual'
    )

    # Run specific task
    task_instance = dag_run.get_task_instance('extract')
    task_instance.run()

    assert task_instance.state == State.SUCCESS

Performance Optimization

1. Use Deferrable Operators: For sensors and long-running waits 2. Dynamic Task Mapping: For parallel processing 3. Appropriate Executor: Choose based on scale (Local, Celery, Kubernetes) 4. Connection Pooling: Reuse database connections 5. Task Parallelism: Set max_active_runs and concurrency appropriately 6. Lazy Loading: Don't execute heavy logic at DAG parse time 7. External Storage: Keep metadata database light

Security

1. Secrets Management: Use Airflow Secrets Backend (not hardcoded) 2. Connection Encryption: Use encrypted connections for databases 3. RBAC: Enable role-based access control 4. Audit Logging: Enable audit logs for compliance 5. Network Isolation: Restrict worker network access 6. Credential Rotation: Regularly rotate credentials

Configuration Management

# Use Variables for configuration
from airflow.models import Variable

config = Variable.get("my_config", deserialize_json=True)
api_key = Variable.get("api_key")

# Use Connections for external services
from airflow.hooks.base import BaseHook

conn = BaseHook.get_connection('my_postgres')
db_url = f"postgresql://{conn.login}:{conn.password}@{conn.host}:{conn.port}/{conn.schema}"

Common Patterns and Examples

See EXAMPLES.md for 18+ detailed real-world examples including:

• ETL pipelines
• Machine learning workflows
• Data quality checks
• Multi-cloud orchestration
• Event-driven architectures
• Complex branching logic
• Dynamic task generation
• Asset-based scheduling
• Sensor patterns
• Error handling strategies

Troubleshooting

DAG Not Appearing in UI

1. Check for Python syntax errors in DAG file 2. Verify DAG file is in correct directory 3. Check dag_id is unique 4. Ensure schedule is not None if you expect it to run 5. Check scheduler logs for import errors

Tasks Not Running

1. Check task dependencies are correct 2. Verify upstream tasks succeeded 3. Check task concurrency limits 4. Ensure executor has available slots 5. Review task logs for errors

Performance Issues

1. Reduce DAG complexity (break into multiple DAGs) 2. Optimize SQL queries in tasks 3. Use appropriate executor for scale 4. Enable task parallelism 5. Check for slow sensors (use deferrable mode) 6. Monitor metadata database performance

Common Errors

Import Errors:

# Bad - imports at DAG level slow parsing
from heavy_library import process

with DAG(...):
    pass

# Good - imports inside tasks
with DAG(...):
    @task
    def my_task():
        from heavy_library import process
        process()

Circular Dependencies:

# This will fail
task1 >> task2 >> task3 >> task1  # Circular!

# Must be acyclic
task1 >> task2 >> task3

Large XComs:

# Bad - storing large data in XCom
@task
def process():
    large_df = pd.read_csv('big_file.csv')
    return large_df  # Too large!

# Good - store reference
@task
def process():
    large_df = pd.read_csv('big_file.csv')
    path = save_to_s3(large_df)
    return path  # Just the path

Resources

• Official Documentation: https://airflow.apache.org/docs/
• Airflow GitHub: https://github.com/apache/airflow
• Astronomer Guides: https://docs.astronomer.io/learn
• Community Slack: https://apache-airflow.slack.com
• Stack Overflow: Tag apache-airflow
• Awesome Airflow: https://github.com/jghoman/awesome-apache-airflow

---

Skill Version: 1.0.0 Last Updated: January 2025 Apache Airflow Version: 2.7+ Skill Category: Data Engineering, Workflow Orchestration, Pipeline Management

Apache Airflow Orchestration Examples

Comprehensive collection of real-world Apache Airflow patterns and examples demonstrating DAGs, operators, sensors, XComs, dynamic workflows, and production deployment strategies.

Table of Contents

1. ETL Pipeline Examples 2. Dynamic Task Generation 3. Sensor Patterns 4. Asset-Based Scheduling 5. Branching and Conditional Logic 6. XCom Communication Patterns 7. TaskFlow API Examples 8. Production Deployment Patterns 9. Error Handling and Resilience 10. Advanced Orchestration

---

ETL Pipeline Examples

Example 1: Traditional ETL with PythonOperator

Classic ETL pattern using PythonOperator with manual XCom management.

import json
import pendulum
from airflow.sdk import DAG
from airflow.providers.standard.operators.python import PythonOperator

def extract():
    """Extract data from source"""
    # Simulate extracting data from an API or database
    data_string = '{"1001": 301.27, "1002": 433.21, "1003": 502.22}'
    order_data = json.loads(data_string)
    return order_data

def transform(ti):
    """Transform the extracted data"""
    # Pull data from XCom
    order_data_dict = ti.xcom_pull(task_ids="extract")

    # Transform: calculate total
    total_order_value = sum(order_data_dict.values())

    # Return transformed data
    return {"total_order_value": total_order_value}

def load(ti):
    """Load transformed data to destination"""
    # Pull transformed data
    total = ti.xcom_pull(task_ids="transform")["total_order_value"]

    # Simulate loading to database or data warehouse
    print(f"Total order value is: {total:.2f}")
    print("Data loaded successfully to warehouse")

with DAG(
    dag_id="traditional_etl_pipeline",
    schedule=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    tags=["etl", "example"],
) as dag:
    extract_task = PythonOperator(task_id="extract", python_callable=extract)
    transform_task = PythonOperator(task_id="transform", python_callable=transform)
    load_task = PythonOperator(task_id="load", python_callable=load)

    extract_task >> transform_task >> load_task

Key Concepts:

• Manual XCom push/pull using ti.xcom_pull()
• Explicit task dependency definition
• Return values automatically pushed to XCom

---

Example 2: Modern ETL with TaskFlow API

Same ETL pattern using TaskFlow API with automatic XCom handling.

from airflow.decorators import dag, task
import pendulum
import json

@dag(
    dag_id="taskflow_etl_pipeline",
    schedule=None,
    start_date=pendulum.datetime(2023, 10, 26, tz="UTC"),
    catchup=False,
    tags=["etl", "taskflow", "example"],
)
def modern_etl():
    """Modern ETL pipeline using TaskFlow API"""

    @task
    def extract():
        """Extract data from source"""
        data_string = '{"1001": 301.27, "1002": 433.21, "1003": 502.22}'
        return json.loads(data_string)

    @task
    def transform(order_data_dict):
        """Transform extracted data"""
        total_order_value = sum(order_data_dict.values())
        return {"total_order_value": total_order_value}

    @task
    def load(summary):
        """Load transformed data"""
        print(f"Total order value: {summary['total_order_value']:.2f}")
        print("Data loaded successfully")

    # Build pipeline - XComs handled automatically
    order_data = extract()
    summary = transform(order_data)
    load(summary)

modern_etl()

Key Concepts:

• Automatic XCom management
• Cleaner, more Pythonic syntax
• Function calls create dependencies automatically

---

Example 3: Multi-Source ETL with Fan-In Pattern

Extract from multiple sources, transform each, then combine and load.

from airflow.decorators import dag, task
import pendulum

@dag(
    dag_id="multi_source_etl",
    start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
    schedule="@daily",
    catchup=False,
    tags=["etl", "multi-source"],
)
def multi_source_pipeline():
    """ETL from multiple sources with fan-in aggregation"""

    @task
    def extract_postgres():
        """Extract from PostgreSQL"""
        # Simulate database extraction
        return {"source": "postgres", "records": 1000, "sum": 50000}

    @task
    def extract_s3():
        """Extract from S3"""
        # Simulate S3 extraction
        return {"source": "s3", "records": 500, "sum": 25000}

    @task
    def extract_api():
        """Extract from external API"""
        # Simulate API extraction
        return {"source": "api", "records": 750, "sum": 37500}

    @task
    def transform_data(data):
        """Transform individual source data"""
        return {
            "source": data["source"],
            "avg_value": data["sum"] / data["records"],
            "record_count": data["records"]
        }

    @task
    def combine_and_load(transformed_data_list):
        """Combine all transformed data and load"""
        total_records = sum(d["record_count"] for d in transformed_data_list)
        sources = [d["source"] for d in transformed_data_list]

        print(f"Combined {total_records} records from {len(sources)} sources")
        for data in transformed_data_list:
            print(f"  {data['source']}: {data['record_count']} records, "
                  f"avg value: {data['avg_value']:.2f}")

        return {"total_records": total_records, "sources": sources}

    # Extract from all sources
    postgres_data = extract_postgres()
    s3_data = extract_s3()
    api_data = extract_api()

    # Transform each source
    transformed_postgres = transform_data(postgres_data)
    transformed_s3 = transform_data(s3_data)
    transformed_api = transform_data(api_data)

    # Combine and load
    combine_and_load([transformed_postgres, transformed_s3, transformed_api])

multi_source_pipeline()

Key Concepts:

• Multiple independent extract tasks
• Parallel transformation
• Fan-in aggregation pattern
• List comprehension in final task

---

Dynamic Task Generation

Example 4: Dynamic Tasks from Configuration

Generate tasks dynamically based on configuration data.

from airflow.sdk import DAG
from airflow.operators.empty import EmptyOperator
from airflow.operators.bash import BashOperator
from datetime import datetime

# Configuration - could be loaded from file or database
PROCESSING_CONFIG = [
    {"name": "customer_data", "path": "/data/customers", "format": "parquet"},
    {"name": "order_data", "path": "/data/orders", "format": "csv"},
    {"name": "product_data", "path": "/data/products", "format": "json"},
    {"name": "inventory_data", "path": "/data/inventory", "format": "parquet"},
]

with DAG(
    dag_id="dynamic_tasks_from_config",
    start_date=datetime(2023, 1, 1),
    schedule="@daily",
    catchup=False,
    tags=["dynamic", "config-driven"],
) as dag:

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

    # Dynamically create tasks for each configuration
    for config in PROCESSING_CONFIG:
        task = BashOperator(
            task_id=f"process_{config['name']}",
            bash_command=f"python process_data.py --path {config['path']} --format {config['format']}"
        )

        start >> task >> end

Key Concepts:

• Configuration-driven task generation
• Loop-based dynamic creation
• Fan-out from start, fan-in to end

---

Example 5: Dynamic Task Mapping

Use dynamic task mapping to create parallel tasks based on runtime data.

from airflow.decorators import dag, task
import pendulum

@dag(
    dag_id="dynamic_task_mapping",
    start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
    schedule="@daily",
    catchup=False,
    tags=["dynamic", "mapping"],
)
def dynamic_mapping_example():
    """Dynamic task mapping for parallel processing"""

    @task
    def get_file_list():
        """Get list of files to process"""
        # This could query a database, API, or file system
        return [
            "sales_2023_01.csv",
            "sales_2023_02.csv",
            "sales_2023_03.csv",
            "sales_2023_04.csv",
            "sales_2023_05.csv",
        ]

    @task
    def process_file(filename):
        """Process a single file"""
        print(f"Processing {filename}")
        # Simulate processing
        record_count = len(filename) * 100  # Fake processing
        return {"file": filename, "records": record_count}

    @task
    def aggregate_results(results):
        """Aggregate all processing results"""
        total_files = len(results)
        total_records = sum(r["records"] for r in results)

        print(f"Processed {total_files} files with {total_records} total records")
        for result in results:
            print(f"  {result['file']}: {result['records']} records")

        return {"total_files": total_files, "total_records": total_records}

    # Dynamic mapping creates parallel tasks
    files = get_file_list()
    results = process_file.expand(filename=files)
    aggregate_results(results)

dynamic_mapping_example()

Key Concepts:

• .expand() for dynamic task mapping
• Automatic parallelization
• Results automatically collected as list
• Number of tasks determined at runtime

---

Example 6: Mapped Task Groups

Map entire task groups for complex parallel processing.

from airflow.decorators import dag, task, task_group
import pendulum

@dag(
    dag_id="mapped_task_groups",
    start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
    schedule="@daily",
    catchup=False,
    tags=["dynamic", "taskgroup"],
)
def mapped_taskgroup_example():
    """Map entire task groups for complex workflows"""

    @task
    def get_datasets():
        """Get list of datasets to process"""
        return ["dataset_A", "dataset_B", "dataset_C"]

    @task_group
    def process_dataset(dataset_name):
        """Task group for processing a single dataset"""

        @task
        def extract(name):
            print(f"Extracting {name}")
            return {"dataset": name, "raw_records": 1000}

        @task
        def validate(data):
            print(f"Validating {data['dataset']}")
            data["valid_records"] = int(data["raw_records"] * 0.95)
            return data

        @task
        def transform(data):
            print(f"Transforming {data['dataset']}")
            data["transformed_records"] = data["valid_records"]
            return data

        @task
        def load(data):
            print(f"Loading {data['dataset']}: {data['transformed_records']} records")
            return data

        # Build task group workflow
        extracted = extract(dataset_name)
        validated = validate(extracted)
        transformed = transform(validated)
        return load(transformed)

    @task
    def create_summary(results):
        """Create summary of all processed datasets"""
        print("Processing Summary:")
        for result in results:
            print(f"  {result['dataset']}: {result['transformed_records']} records")

    # Map task group over datasets
    datasets = get_datasets()
    results = process_dataset.expand(dataset_name=datasets)
    create_summary(results)

mapped_taskgroup_example()

Key Concepts:

• TaskGroup mapping with .expand()
• Complex multi-step processing per item
• Automatic aggregation of task group results
• Clean organization of parallel workflows

---

Sensor Patterns

Example 7: External Task Sensor

Wait for a task in another DAG to complete.

from airflow.sdk import DAG
from airflow.providers.standard.sensors.external_task import ExternalTaskSensor
from airflow.operators.empty import EmptyOperator
from airflow.operators.bash import BashOperator
import pendulum

# Upstream DAG
with DAG(
    dag_id="upstream_data_processor",
    start_date=pendulum.datetime(2021, 10, 20, tz="UTC"),
    schedule="@daily",
    catchup=False,
    tags=["upstream"],
) as upstream_dag:

    process_data = BashOperator(
        task_id="process_daily_data",
        bash_command="echo 'Processing data for {{ ds }}'"
    )

    mark_complete = EmptyOperator(task_id="mark_complete")

    process_data >> mark_complete

# Downstream DAG - waits for upstream
with DAG(
    dag_id="downstream_report_generator",
    start_date=pendulum.datetime(2021, 10, 20, tz="UTC"),
    schedule="@daily",
    catchup=False,
    tags=["downstream", "sensor"],
) as downstream_dag:

    start = EmptyOperator(task_id="start")

    # Wait for upstream DAG task to complete
    wait_for_upstream = ExternalTaskSensor(
        task_id="wait_for_data_processing",
        external_dag_id="upstream_data_processor",
        external_task_id="mark_complete",
        allowed_states=["success"],
        failed_states=["failed", "skipped"],
        execution_delta=None,  # Same execution_date
        timeout=3600,  # 1 hour timeout
        poke_interval=60,  # Check every 60 seconds
    )

    generate_report = BashOperator(
        task_id="generate_report",
        bash_command="echo 'Generating report from processed data'"
    )

    end = EmptyOperator(task_id="end")

    start >> wait_for_upstream >> generate_report >> end

Key Concepts:

• Cross-DAG dependencies with ExternalTaskSensor
• Configurable timeout and poke interval
• State-based waiting (success, failed)
• Execution date alignment

---

Example 8: Deferrable Sensor for Efficiency

Use deferrable sensors to release worker slots while waiting.

from airflow.sdk import DAG
from airflow.providers.standard.sensors.external_task import ExternalTaskSensor
from airflow.operators.bash import BashOperator
import pendulum

with DAG(
    dag_id="deferrable_sensor_example",
    start_date=pendulum.datetime(2021, 10, 20, tz="UTC"),
    schedule="@daily",
    catchup=False,
    tags=["sensor", "deferrable"],
) as dag:

    # Traditional sensor - holds worker slot
    traditional_sensor = ExternalTaskSensor(
        task_id="traditional_wait",
        external_dag_id="upstream_dag",
        external_task_id="upstream_task",
        allowed_states=["success"],
        timeout=3600,
        poke_interval=60,
        deferrable=False,  # Blocks worker
    )

    # Deferrable sensor - releases worker slot
    deferrable_sensor = ExternalTaskSensor(
        task_id="deferrable_wait",
        external_dag_id="upstream_dag",
        external_task_id="upstream_task",
        allowed_states=["success"],
        deferrable=True,  # Releases worker, uses triggerer
    )

    process = BashOperator(
        task_id="process_data",
        bash_command="echo 'Processing after wait'"
    )

    # Use deferrable for better resource utilization
    deferrable_sensor >> process

Key Concepts:

• Deferrable sensors release worker slots
• More efficient for long waits
• Requires triggerer component
• Same functionality, better resource usage

---

Example 9: Custom Sensor with Deferrable Support

Create a custom sensor with deferrable capability.

from datetime import timedelta
from airflow.sdk import BaseSensorOperator, Context, StartTriggerArgs, DAG
from airflow.operators.bash import BashOperator
import pendulum

class WaitHoursSensor(BaseSensorOperator):
    """Custom sensor that waits for specified hours"""

    start_trigger_args = StartTriggerArgs(
        trigger_cls="airflow.providers.standard.triggers.temporal.TimeDeltaTrigger",
        trigger_kwargs={"moment": timedelta(hours=1)},
        next_method="execute_complete",
        next_kwargs=None,
        timeout=None,
    )
    start_from_trigger = True

    def __init__(
        self,
        *args,
        trigger_kwargs=None,
        start_from_trigger=True,
        **kwargs
    ):
        super().__init__(*args, **kwargs)
        if trigger_kwargs:
            self.start_trigger_args.trigger_kwargs = trigger_kwargs
        self.start_from_trigger = start_from_trigger

    def execute_complete(self, context: Context, event=None):
        """Called when trigger completes"""
        self.log.info("Wait period completed")
        return

with DAG(
    dag_id="custom_deferrable_sensor",
    start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
    schedule="@daily",
    catchup=False,
) as dag:

    wait = WaitHoursSensor(
        task_id="wait_2_hours",
        trigger_kwargs={"moment": timedelta(hours=2)}
    )

    process = BashOperator(
        task_id="process_after_wait",
        bash_command="echo 'Processing after 2 hour wait'"
    )

    wait >> process

Key Concepts:

• Custom deferrable sensor implementation
• StartTriggerArgs for trigger configuration
• execute_complete callback
• Configurable wait time

---

Asset-Based Scheduling

Example 10: Producer-Consumer Asset Pattern

Event-driven workflow triggered by data asset updates.

from airflow.sdk import DAG, Asset
from airflow.operators.bash import BashOperator
from airflow.decorators import task
from datetime import datetime

# Define assets
customer_data_asset = Asset("s3://my-bucket/customers.parquet")
order_data_asset = Asset("s3://my-bucket/orders.parquet")

# PRODUCER DAG - Updates assets
with DAG(
    dag_id="producer_customer_orders",
    start_date=datetime(2023, 1, 1),
    schedule="@daily",
    catchup=False,
    tags=["producer", "asset"],
) as producer:

    extract_customers = BashOperator(
        task_id="extract_customers",
        bash_command="python extract_customers.py",
        outlets=[customer_data_asset]  # Marks asset as updated
    )

    extract_orders = BashOperator(
        task_id="extract_orders",
        bash_command="python extract_orders.py",
        outlets=[order_data_asset]  # Marks asset as updated
    )

# CONSUMER DAG - Triggered when BOTH assets update
with DAG(
    dag_id="consumer_customer_analytics",
    schedule=[customer_data_asset & order_data_asset],  # AND condition
    start_date=datetime(2023, 1, 1),
    catchup=False,
    tags=["consumer", "asset"],
) as consumer:

    @task
    def process_customer_orders(*, triggering_asset_events):
        """Process when both assets are ready"""
        print("Processing customer analytics with fresh data")

        for event in triggering_asset_events:
            print(f"Asset updated: {event.asset.uri}")
            print(f"Update time: {event.timestamp}")

        # Process data from both sources
        print("Combining customer and order data for analytics")

    process_customer_orders()

Key Concepts:

• Asset-based event-driven scheduling
• Producer marks assets as updated with outlets
• Consumer triggered by asset updates
• AND logic with & operator
• Access to triggering asset event metadata

---

Example 11: Complex Asset Scheduling Logic

Advanced asset scheduling with OR and AND combinations.

from airflow.datasets import Dataset
from airflow.models.dag import DAG
from airflow.decorators import task
from datetime import datetime

# Define multiple data assets
daily_sales = Dataset("s3://data/daily_sales.parquet")
weekly_inventory = Dataset("s3://data/weekly_inventory.parquet")
monthly_forecast = Dataset("s3://data/monthly_forecast.parquet")

# Producer 1: Daily sales data
with DAG(
    dag_id="producer_daily_sales",
    start_date=datetime(2023, 1, 1),
    schedule="@daily",
    catchup=False,
) as producer1:

    @task(outlets=[daily_sales])
    def generate_daily_sales():
        print("Generating daily sales report")

    generate_daily_sales()

# Producer 2: Weekly inventory
with DAG(
    dag_id="producer_weekly_inventory",
    start_date=datetime(2023, 1, 1),
    schedule="@weekly",
    catchup=False,
) as producer2:

    @task(outlets=[weekly_inventory])
    def generate_weekly_inventory():
        print("Generating weekly inventory snapshot")

    generate_weekly_inventory()

# Producer 3: Monthly forecast
with DAG(
    dag_id="producer_monthly_forecast",
    start_date=datetime(2023, 1, 1),
    schedule="@monthly",
    catchup=False,
) as producer3:

    @task(outlets=[monthly_forecast])
    def generate_monthly_forecast():
        print("Generating monthly forecast")

    generate_monthly_forecast()

# Consumer with complex logic:
# Trigger when: daily_sales OR (weekly_inventory AND monthly_forecast)
with DAG(
    dag_id="consumer_complex_analytics",
    schedule=(daily_sales | (weekly_inventory & monthly_forecast)),
    start_date=datetime(2023, 1, 1),
    catchup=False,
    tags=["asset", "complex-logic"],
) as complex_consumer:

    @task
    def analyze_data(*, triggering_asset_events):
        """Run analytics based on which assets triggered"""
        triggered_assets = [event.asset.uri for event in triggering_asset_events]

        print(f"Triggered by {len(triggered_assets)} asset(s):")
        for asset in triggered_assets:
            print(f"  - {asset}")

        if "s3://data/daily_sales.parquet" in triggered_assets:
            print("Running daily sales analysis")

        if all(a in triggered_assets for a in [
            "s3://data/weekly_inventory.parquet",
            "s3://data/monthly_forecast.parquet"
        ]):
            print("Running comprehensive inventory + forecast analysis")

    analyze_data()

Key Concepts:

• Complex asset logic with OR (|) and AND (&)
• Multiple producers with different schedules
• Conditional processing based on triggering assets
• Event metadata access

---

Example 12: Asset Aliases for Flexibility

Use asset aliases for dynamic asset resolution.

from airflow.datasets import Dataset, AssetAlias
from airflow.models.dag import DAG
from airflow.decorators import task
from datetime import datetime

# Producer with dynamic asset determination
with DAG(
    dag_id="producer_with_alias",
    start_date=datetime(2023, 1, 1),
    schedule="@daily",
    catchup=False,
) as alias_producer:

    @task(outlets=[AssetAlias("daily-data-output")])
    def produce_data(*, outlet_events, **context):
        """Dynamically determine which asset to update"""
        execution_date = context['ds']

        # Determine actual asset based on date
        if execution_date.endswith('01'):  # First of month
            actual_asset = Dataset("s3://bucket/monthly_data.parquet")
        else:
            actual_asset = Dataset("s3://bucket/daily_data.parquet")

        # Register the actual asset with the alias
        outlet_events[AssetAlias("daily-data-output")].add(actual_asset)

        print(f"Updated asset: {actual_asset.uri}")

    produce_data()

# Consumer depends on alias - gets actual asset at runtime
with DAG(
    dag_id="consumer_from_alias",
    schedule=AssetAlias("daily-data-output"),
    start_date=datetime(2023, 1, 1),
    catchup=False,
) as alias_consumer:

    @task
    def process_data(*, triggering_asset_events):
        """Process whichever asset was actually updated"""
        for event in triggering_asset_events:
            print(f"Processing: {event.asset.uri}")

    process_data()

Key Concepts:

• AssetAlias for flexible asset references
• Dynamic asset resolution at runtime
• Alias resolution to actual datasets
• Decoupling producer from consumer

---

Branching and Conditional Logic

Example 13: Branch Operator for Conditional Execution

Execute different paths based on runtime conditions.

from airflow.sdk import DAG
from airflow.operators.python import BranchPythonOperator
from airflow.operators.bash import BashOperator
from airflow.operators.empty import EmptyOperator
import pendulum

def choose_branch(**context):
    """Decide which branch to take based on execution date"""
    execution_date = context['data_interval_start']

    # Run monthly task on first day of month
    if execution_date.day == 1:
        return 'monthly_processing'
    # Run weekly task on Mondays
    elif execution_date.weekday() == 0:
        return ['daily_processing', 'weekly_processing']
    # Run daily task on other days
    else:
        return 'daily_processing'

with DAG(
    dag_id="conditional_branching",
    start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
    schedule="@daily",
    catchup=False,
    tags=["branching", "conditional"],
) as dag:

    start = EmptyOperator(task_id='start')

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

    daily_processing = BashOperator(
        task_id='daily_processing',
        bash_command='echo "Running daily processing for {{ ds }}"'
    )

    weekly_processing = BashOperator(
        task_id='weekly_processing',
        bash_command='echo "Running weekly processing for week of {{ ds }}"'
    )

    monthly_processing = BashOperator(
        task_id='monthly_processing',
        bash_command='echo "Running monthly processing for month of {{ ds }}"'
    )

    end = EmptyOperator(
        task_id='end',
        trigger_rule='none_failed_min_one_success'  # Run if any branch succeeded
    )

    # Set dependencies
    start >> branch >> [daily_processing, weekly_processing, monthly_processing] >> end

Key Concepts:

• BranchPythonOperator for conditional execution
• Return task_id(s) to execute
• Trigger rules for convergence points
• Date-based branching logic

---

Example 14: Custom Branch Operator

Implement custom branching logic with BaseBranchOperator.

from airflow.operators.branch import BaseBranchOperator
from airflow.sdk import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.empty import EmptyOperator
import pendulum

class DataVolumeBranchOperator(BaseBranchOperator):
    """Branch based on data volume thresholds"""

    def __init__(self, volume_threshold=1000, **kwargs):
        super().__init__(**kwargs)
        self.volume_threshold = volume_threshold

    def choose_branch(self, context):
        """Determine branch based on data volume"""
        # Simulate checking data volume
        # In reality, query database or check file size
        import random
        data_volume = random.randint(100, 2000)

        self.log.info(f"Data volume: {data_volume}")

        if data_volume > self.volume_threshold:
            self.log.info("High volume detected - using distributed processing")
            return 'distributed_processing'
        else:
            self.log.info("Normal volume - using standard processing")
            return 'standard_processing'

with DAG(
    dag_id="custom_branch_operator",
    start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
    schedule="@daily",
    catchup=False,
    tags=["branching", "custom-operator"],
) as dag:

    start = EmptyOperator(task_id='start')

    check_volume = DataVolumeBranchOperator(
        task_id='check_data_volume',
        volume_threshold=1000
    )

    standard_processing = BashOperator(
        task_id='standard_processing',
        bash_command='echo "Running standard single-node processing"'
    )

    distributed_processing = BashOperator(
        task_id='distributed_processing',
        bash_command='echo "Running distributed Spark processing"'
    )

    end = EmptyOperator(
        task_id='end',
        trigger_rule='none_failed_min_one_success'
    )

    start >> check_volume >> [standard_processing, distributed_processing] >> end

Key Concepts:

• Custom operator inheriting from BaseBranchOperator
• choose_branch method for logic
• Configurable parameters
• Data-driven branching

---

XCom Communication Patterns

Example 15: Advanced XCom Usage with Multiple Outputs

Complex XCom patterns with multiple outputs and transformations.

from airflow.decorators import dag, task
import pendulum

@dag(
    dag_id="advanced_xcom_patterns",
    start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
    schedule="@daily",
    catchup=False,
    tags=["xcom", "advanced"],
)
def xcom_advanced():
    """Demonstrate advanced XCom patterns"""

    @task(multiple_outputs=True)
    def extract_multiple_sources():
        """Extract data from multiple sources and return as dict"""
        return {
            'database_records': 1500,
            'api_records': 800,
            'file_records': 1200,
            'total_records': 3500
        }

    @task
    def validate_database(record_count):
        """Validate database records"""
        print(f"Validating {record_count} database records")
        valid_count = int(record_count * 0.98)  # 98% valid
        return {
            'source': 'database',
            'total': record_count,
            'valid': valid_count,
            'invalid': record_count - valid_count
        }

    @task
    def validate_api(record_count):
        """Validate API records"""
        print(f"Validating {record_count} API records")
        valid_count = int(record_count * 0.95)  # 95% valid
        return {
            'source': 'api',
            'total': record_count,
            'valid': valid_count,
            'invalid': record_count - valid_count
        }

    @task
    def validate_files(record_count):
        """Validate file records"""
        print(f"Validating {record_count} file records")
        valid_count = int(record_count * 0.97)  # 97% valid
        return {
            'source': 'files',
            'total': record_count,
            'valid': valid_count,
            'invalid': record_count - valid_count
        }

    @task
    def create_summary(db_result, api_result, file_result):
        """Create summary from all validation results"""
        total_valid = db_result['valid'] + api_result['valid'] + file_result['valid']
        total_invalid = db_result['invalid'] + api_result['invalid'] + file_result['invalid']

        summary = {
            'total_records': total_valid + total_invalid,
            'total_valid': total_valid,
            'total_invalid': total_invalid,
            'validation_rate': (total_valid / (total_valid + total_invalid)) * 100,
            'by_source': {
                'database': db_result,
                'api': api_result,
                'files': file_result
            }
        }

        print(f"\nValidation Summary:")
        print(f"Total Records: {summary['total_records']}")
        print(f"Valid: {summary['total_valid']} ({summary['validation_rate']:.2f}%)")
        print(f"Invalid: {summary['total_invalid']}")

        for source_name, source_data in summary['by_source'].items():
            print(f"\n{source_name.upper()}:")
            print(f"  Total: {source_data['total']}")
            print(f"  Valid: {source_data['valid']}")
            print(f"  Invalid: {source_data['invalid']}")

        return summary

    # Build workflow
    sources = extract_multiple_sources()

    # Validate each source in parallel
    db_validated = validate_database(sources['database_records'])
    api_validated = validate_api(sources['api_records'])
    file_validated = validate_files(sources['file_records'])

    # Create final summary
    create_summary(db_validated, api_validated, file_validated)

xcom_advanced()

Key Concepts:

• multiple_outputs=True for dictionary returns
• Access individual keys from task output
• Parallel validation tasks
• Complex aggregation from multiple XComs

---

Example 16: XCom with External Storage

Handle large data by storing in external systems and passing references via XCom.

from airflow.decorators import dag, task
import pendulum
import json

@dag(
    dag_id="xcom_external_storage",
    start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
    schedule="@daily",
    catchup=False,
    tags=["xcom", "s3", "large-data"],
)
def xcom_with_s3():
    """Handle large datasets by storing in S3 and passing references"""

    @task
    def extract_large_dataset():
        """Extract large dataset and store in S3"""
        # Simulate extracting large data
        large_data = {
            'records': list(range(1000000)),  # 1M records
            'metadata': {'source': 'database', 'timestamp': '2023-01-01'}
        }

        # In real implementation, use boto3 to upload to S3
        s3_path = "s3://my-bucket/data/extract_{{ ds }}.parquet"

        print(f"Extracted {len(large_data['records'])} records")
        print(f"Stored in: {s3_path}")

        # Return only the reference, not the data!
        return {
            's3_path': s3_path,
            'record_count': len(large_data['records']),
            'file_size_mb': 150.5
        }

    @task
    def transform_large_dataset(extract_metadata):
        """Transform large dataset from S3"""
        s3_path = extract_metadata['s3_path']

        print(f"Loading data from: {s3_path}")
        # In real implementation: data = load_from_s3(s3_path)

        print(f"Transforming {extract_metadata['record_count']} records")

        # Store transformed data in S3
        transformed_path = s3_path.replace('extract', 'transform')

        print(f"Stored transformed data in: {transformed_path}")

        return {
            's3_path': transformed_path,
            'record_count': extract_metadata['record_count'],
            'file_size_mb': 120.3
        }

    @task
    def load_to_warehouse(transform_metadata):
        """Load transformed data to data warehouse"""
        s3_path = transform_metadata['s3_path']

        print(f"Loading from S3: {s3_path}")
        print(f"Loading {transform_metadata['record_count']} records to warehouse")

        # Simulate loading
        print("Data successfully loaded to warehouse")

        return {
            'status': 'success',
            'records_loaded': transform_metadata['record_count'],
            'source': s3_path
        }

    # Build pipeline
    extracted = extract_large_dataset()
    transformed = transform_large_dataset(extracted)
    load_to_warehouse(transformed)

xcom_with_s3()

Key Concepts:

• Store large data externally (S3, GCS, etc.)
• Pass only metadata/references via XCom
• Keep XCom size small (< 1MB)
• File paths and metadata in XCom

---

TaskFlow API Examples

Example 17: TaskFlow with Virtual Environments

Isolate task dependencies using virtual environments.

from airflow.decorators import dag, task
import pendulum

@dag(
    dag_id="taskflow_virtualenv",
    start_date=pendulum.datetime(2023, 10, 26, tz="UTC"),
    schedule="@daily",
    catchup=False,
    tags=["taskflow", "virtualenv"],
)
def virtualenv_example():
    """Use virtual environments for dependency isolation"""

    @task.virtualenv(
        requirements=["pandas==2.0.0", "numpy==1.24.0"],
        system_site_packages=False
    )
    def analyze_with_pandas():
        """Analyze data using specific pandas version"""
        import pandas as pd
        import numpy as np

        # Create sample data
        data = {
            'product': ['A', 'B', 'C', 'D', 'E'],
            'sales': [100, 200, 150, 300, 250],
            'profit': [20, 40, 30, 60, 50]
        }

        df = pd.DataFrame(data)

        # Analysis
        total_sales = df['sales'].sum()
        total_profit = df['profit'].sum()
        profit_margin = (total_profit / total_sales) * 100

        print(f"Total Sales: ${total_sales}")
        print(f"Total Profit: ${total_profit}")
        print(f"Profit Margin: {profit_margin:.2f}%")

        return {
            'total_sales': float(total_sales),
            'total_profit': float(total_profit),
            'profit_margin': float(profit_margin)
        }

    @task.virtualenv(
        requirements=["scikit-learn==1.3.0"],
        system_site_packages=False
    )
    def ml_prediction(sales_data):
        """Run ML prediction with specific sklearn version"""
        from sklearn.linear_model import LinearRegression
        import numpy as np

        # Simple prediction example
        X = np.array([[1], [2], [3], [4], [5]])
        y = np.array([100, 200, 150, 300, 250])

        model = LinearRegression()
        model.fit(X, y)

        # Predict next period
        next_period = model.predict([[6]])

        print(f"Predicted sales for next period: ${next_period[0]:.2f}")

        return {
            'predicted_sales': float(next_period[0]),
            'model_score': float(model.score(X, y))
        }

    @task
    def create_report(analysis, prediction):
        """Create final report"""
        print("\n=== SALES REPORT ===")
        print(f"Current Period Analysis:")
        print(f"  Total Sales: ${analysis['total_sales']}")
        print(f"  Total Profit: ${analysis['total_profit']}")
        print(f"  Profit Margin: {analysis['profit_margin']:.2f}%")
        print(f"\nNext Period Prediction:")
        print(f"  Predicted Sales: ${prediction['predicted_sales']:.2f}")
        print(f"  Model Confidence: {prediction['model_score']:.2%}")

    # Build workflow
    analysis_result = analyze_with_pandas()
    prediction_result = ml_prediction(analysis_result)
    create_report(analysis_result, prediction_result)

virtualenv_example()

Key Concepts:

• task.virtualenv decorator
• Isolated dependencies per task
• Different library versions per task
• Automatic environment creation

---

Example 18: TaskFlow with Traditional Operators

Mix TaskFlow tasks with traditional operators seamlessly.

from airflow.decorators import dag, task
from airflow.providers.standard.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.operators.email import EmailOperator
import pendulum

@dag(
    dag_id="mixed_taskflow_traditional",
    start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
    schedule="@daily",
    catchup=False,
    tags=["taskflow", "mixed"],
)
def mixed_operators():
    """Mix TaskFlow and traditional operators"""

    # Traditional BashOperator
    check_data = BashOperator(
        task_id="check_data_availability",
        bash_command="ls /data/input/*.csv | wc -l"
    )

    @task
    def get_file_list():
        """Get list of files to process"""
        # In reality, list files from directory or S3
        return ["file1.csv", "file2.csv", "file3.csv"]

    @task
    def validate_files(files):
        """Validate file integrity"""
        print(f"Validating {len(files)} files")
        valid_files = [f for f in files if 'file' in f]  # Simple validation
        return {
            'total_files': len(files),
            'valid_files': valid_files,
            'validation_rate': len(valid_files) / len(files)
        }

    # Traditional PythonOperator
    def process_function(**context):
        # Pull XCom from TaskFlow task
        validation_result = context['ti'].xcom_pull(task_ids='validate_files')
        print(f"Processing {validation_result['total_files']} files")

    process_data = PythonOperator(
        task_id="process_data",
        python_callable=process_function
    )

    @task(multiple_outputs=True)
    def generate_email_content(validation_result):
        """Generate email content from validation"""
        return {
            'subject': f"Data Processing Complete - {validation_result['total_files']} files",
            'body': f"""
                <h2>Processing Summary</h2>
                <p>Total Files: {validation_result['total_files']}</p>
                <p>Valid Files: {len(validation_result['valid_files'])}</p>
                <p>Validation Rate: {validation_result['validation_rate']:.1%}</p>
            """
        }

    # Traditional EmailOperator using TaskFlow output
    email_content = generate_email_content(validate_files(get_file_list()))

    send_email = EmailOperator(
        task_id="send_notification",
        to="team@example.com",
        subject="{{ ti.xcom_pull(task_ids='generate_email_content', key='subject') }}",
        html_content="{{ ti.xcom_pull(task_ids='generate_email_content', key='body') }}"
    )

    # Dependencies
    check_data >> get_file_list()
    process_data << validate_files(get_file_list())
    email_content >> send_email

mixed_operators()

Key Concepts:

• Mixing TaskFlow and traditional operators
• XCom between different operator types
• Template variables with XCom
• Flexible workflow composition

---

Production Deployment Patterns

Example 19: Production DAG with Full Error Handling

Comprehensive production-ready DAG with retries, callbacks, and monitoring.

from airflow.sdk import DAG
from airflow.decorators import task
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta
import pendulum

def on_failure_callback(context):
    """Called when task fails"""
    task_instance = context['task_instance']
    dag_id = context['dag'].dag_id

    print(f"ALERT: Task {task_instance.task_id} in DAG {dag_id} failed!")
    print(f"Execution date: {context['ds']}")
    print(f"Log URL: {task_instance.log_url}")

    # In production: send to Slack, PagerDuty, etc.
    # send_slack_alert(f"Task {task_instance.task_id} failed!")

def on_success_callback(context):
    """Called when task succeeds"""
    task_instance = context['task_instance']
    print(f"SUCCESS: Task {task_instance.task_id} completed successfully")

def on_retry_callback(context):
    """Called when task retries"""
    task_instance = context['task_instance']
    print(f"RETRY: Task {task_instance.task_id} is retrying (attempt {context['ti'].try_number})")

def sla_miss_callback(dag, task_list, blocking_task_list, slas, blocking_tis):
    """Called when SLA is missed"""
    print(f"SLA MISS: Tasks {task_list} missed their SLA")
    # send_sla_alert(task_list)

# Default arguments for all tasks
default_args = {
    'owner': 'data-engineering',
    'depends_on_past': False,
    'email': ['alerts@company.com'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
    'retry_exponential_backoff': True,
    'max_retry_delay': timedelta(hours=1),
    'sla': timedelta(hours=2),  # Task should complete within 2 hours
}

with DAG(
    dag_id="production_data_pipeline",
    default_args=default_args,
    start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
    schedule="0 2 * * *",  # 2 AM daily
    catchup=False,
    max_active_runs=3,
    sla_miss_callback=sla_miss_callback,
    tags=["production", "critical"],
    doc_md="""
    # Production Data Pipeline

    Critical pipeline for processing daily business data.

    ## Schedule
    Runs daily at 2 AM UTC

    ## SLA
    Must complete within 2 hours

    ## Alerts
    - Failures: Email + Slack
    - SLA Miss: Email + PagerDuty
    """,
) as dag:

    @task(
        on_failure_callback=on_failure_callback,
        on_success_callback=on_success_callback,
        on_retry_callback=on_retry_callback,
    )
    def extract_critical_data():
        """Extract critical business data"""
        print("Extracting critical data from production database")
        # Simulate extraction
        return {"records": 10000, "timestamp": str(datetime.now())}

    @task(
        on_failure_callback=on_failure_callback,
        execution_timeout=timedelta(minutes=30),  # Task-specific timeout
    )
    def validate_data(data):
        """Validate extracted data"""
        print(f"Validating {data['records']} records")

        # Simulate validation
        if data['records'] < 1000:
            raise ValueError("Too few records - data quality issue!")

        return {"valid": True, "record_count": data['records']}

    @task(
        on_failure_callback=on_failure_callback,
        pool='heavy_compute',  # Use resource pool
        max_active_tis_per_dag=2,  # Limit concurrent instances
    )
    def transform_data(validation_result):
        """Transform validated data"""
        print(f"Transforming {validation_result['record_count']} records")
        # Heavy transformation logic
        return {"transformed_records": validation_result['record_count']}

    load_to_warehouse = BashOperator(
        task_id="load_to_warehouse",
        bash_command="python /scripts/load_to_warehouse.py --date {{ ds }}",
        on_failure_callback=on_failure_callback,
        on_success_callback=on_success_callback,
    )

    @task
    def data_quality_checks():
        """Run data quality checks on loaded data"""
        print("Running comprehensive data quality checks")
        # Run Great Expectations or custom checks
        return {"quality_score": 0.98, "issues": []}

    @task
    def send_success_report(transform_result, quality_result):
        """Send success report"""
        print(f"\n=== PIPELINE SUCCESS REPORT ===")
        print(f"Records Processed: {transform_result['transformed_records']}")
        print(f"Quality Score: {quality_result['quality_score']:.2%}")
        print(f"Issues Found: {len(quality_result['issues'])}")
        # Send to monitoring dashboard

    # Build pipeline
    extracted = extract_critical_data()
    validated = validate_data(extracted)
    transformed = transform_data(validated)
    transformed >> load_to_warehouse

    quality = data_quality_checks()
    load_to_warehouse >> quality

    send_success_report(transformed, quality)

Key Concepts:

• Comprehensive error handling
• Callbacks for monitoring
• SLA configuration
• Retry strategies with exponential backoff
• Resource pools
• Execution timeouts
• Detailed documentation

---

Example 20: Kubernetes Executor Configuration

Configure tasks for Kubernetes executor with custom resources.

from airflow.sdk import DAG
from airflow.decorators import task
from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator
from kubernetes.client import models as k8s
import pendulum

with DAG(
    dag_id="kubernetes_executor_example",
    start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
    schedule="@daily",
    catchup=False,
    tags=["kubernetes", "production"],
) as dag:

    # Task with GPU requirement
    @task(
        executor_config={
            "pod_override": k8s.V1Pod(
                spec=k8s.V1PodSpec(
                    containers=[
                        k8s.V1Container(
                            name="base",
                            resources=k8s.V1ResourceRequirements(
                                requests={"memory": "4Gi", "cpu": "2"},
                                limits={"memory": "8Gi", "cpu": "4", "nvidia.com/gpu": "1"}
                            )
                        )
                    ]
                )
            )
        }
    )
    def train_ml_model():
        """Train ML model with GPU"""
        print("Training model on GPU")
        # ML training code
        return {"model_accuracy": 0.95}

    # Heavy memory task
    @task(
        executor_config={
            "pod_override": k8s.V1Pod(
                spec=k8s.V1PodSpec(
                    containers=[
                        k8s.V1Container(
                            name="base",
                            resources=k8s.V1ResourceRequirements(
                                requests={"memory": "16Gi", "cpu": "4"},
                                limits={"memory": "32Gi", "cpu": "8"}
                            )
                        )
                    ]
                )
            )
        }
    )
    def process_large_dataset():
        """Process large dataset requiring lots of memory"""
        print("Processing large dataset")
        return {"records_processed": 10000000}

    # Standard Kubernetes Pod Operator
    spark_job = KubernetesPodOperator(
        task_id="run_spark_job",
        name="spark-job",
        namespace="airflow",
        image="my-registry/spark:3.4.0",
        cmds=["spark-submit"],
        arguments=["--master", "k8s://https://kubernetes.default.svc", "/app/job.py"],
        resources=k8s.V1ResourceRequirements(
            requests={"memory": "8Gi", "cpu": "4"},
            limits={"memory": "16Gi", "cpu": "8"}
        ),
    )

    # Dependencies
    model_result = train_ml_model()
    data_result = process_large_dataset()

    [model_result, data_result] >> spark_job

Key Concepts:

• Kubernetes executor configuration
• Custom resource requirements
• GPU allocation
• Memory and CPU limits
• KubernetesPodOperator for complex workloads

---

Continued in next examples...

---

Document Version: 1.0.0 Last Updated: January 2025 Total Examples: 20+ Coverage: ETL, Dynamic Tasks, Sensors, Assets, Branching, XComs, TaskFlow, Production Patterns

Apache Airflow Orchestration Skill

Master Apache Airflow for workflow orchestration, data pipeline automation, and production-grade task scheduling.

Overview

Apache Airflow is the industry-standard platform for programmatically authoring, scheduling, and monitoring workflows. This comprehensive skill teaches you how to build robust, scalable data pipelines using Airflow's powerful features including DAGs, operators, sensors, XComs, dynamic task mapping, and asset-based scheduling.

What is Apache Airflow?

Apache Airflow is an open-source workflow orchestration platform that allows you to:

• Define workflows as code: Write pipelines in Python with full version control
• Schedule complex workflows: Use cron expressions, timedeltas, or event-driven triggers
• Monitor execution: Rich UI for visualizing DAG structure and task status
• Handle dependencies: Model complex task relationships and data dependencies
• Scale horizontally: Execute tasks across distributed clusters
• Integrate everything: Extensive provider ecosystem for databases, cloud services, and tools

Key Capabilities

Core Features

1. DAG Development

• Define workflows as Directed Acyclic Graphs
• Set task dependencies with intuitive operators
• Organize tasks with TaskGroups
• Branch execution based on conditions
• Label edges for clarity

2. Rich Operator Library

• BashOperator for shell commands
• PythonOperator for Python functions
• Provider operators for AWS, GCP, Azure, databases, and more
• Custom operators for specialized tasks
• EmailOperator for notifications

3. Sensors for Waiting

• ExternalTaskSensor for cross-DAG dependencies
• FileSensor for file availability
• TimeDeltaSensor for time-based waits
• BigQueryTableSensor for data warehouse tables
• Custom sensors for any condition
• Deferrable sensors for efficient resource usage

4. XCom Communication

• Pass data between tasks
• Automatic handling with TaskFlow API
• Template variables for dynamic values
• Best practices for data size management

5. Dynamic Workflows

• Generate tasks programmatically
• Dynamic task mapping for parallel processing
• Loop-based task creation
• Conditional task generation
• Map entire TaskGroups

6. Scheduling Patterns

• Cron expressions for complex schedules
• Timedelta-based intervals
• Preset schedules (@daily, @hourly, etc.)
• Asset-based event-driven scheduling
• Manual triggering with parameters

7. Production Features

• Retry logic with exponential backoff
• Task-level and DAG-level concurrency control
• SLA monitoring and alerting
• Callbacks for success, failure, and retry
• Docker and Kubernetes deployment
• Structured logging and metrics

Architecture Overview

Core Components

┌─────────────────────────────────────────────────────────────┐
│                        Airflow Architecture                  │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  ┌──────────────┐         ┌──────────────┐                 │
│  │   Web UI     │────────▶│   Metadata   │                 │
│  │  (Flask)     │         │   Database   │                 │
│  └──────────────┘         └──────────────┘                 │
│         │                        ▲                          │
│         │                        │                          │
│         ▼                        │                          │
│  ┌──────────────┐                │                          │
│  │  Scheduler   │────────────────┘                          │
│  │ (DAG Parser) │                                           │
│  └──────────────┘                                           │
│         │                                                    │
│         │  Submits Tasks                                    │
│         ▼                                                    │
│  ┌──────────────────────────────────────┐                  │
│  │          Executor                    │                  │
│  │  (Local/Celery/Kubernetes/Dask)      │                  │
│  └──────────────────────────────────────┘                  │
│         │                                                    │
│         │  Executes                                         │
│         ▼                                                    │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐     │
│  │   Worker 1   │  │   Worker 2   │  │   Worker N   │     │
│  │   (Task)     │  │   (Task)     │  │   (Task)     │     │
│  └──────────────┘  └──────────────┘  └──────────────┘     │
│                                                              │
└─────────────────────────────────────────────────────────────┘

Execution Flow

1. DAG Parsing: Scheduler parses DAG files to find tasks 2. Task Scheduling: Scheduler determines which tasks to run based on dependencies and schedule 3. Task Submission: Scheduler submits tasks to executor 4. Task Execution: Workers execute tasks and report status 5. State Management: Metadata database stores all state information 6. UI Monitoring: Web UI displays DAG structure and task status

Quick Start Guide

Installation

Using pip:

# Install Airflow
pip install apache-airflow

# Initialize database
airflow db init

# Create admin user
airflow users create \
    --username admin \
    --firstname Admin \
    --lastname User \
    --role Admin \
    --email admin@example.com

Using Docker:

# Download docker-compose.yaml
curl -LfO 'https://airflow.apache.org/docs/apache-airflow/2.7.0/docker-compose.yaml'

# Initialize
docker-compose up airflow-init

# Start services
docker-compose up

Your First DAG

Create a file in ~/airflow/dags/my_first_dag.py:

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

with DAG(
    dag_id='my_first_dag',
    start_date=datetime(2023, 1, 1),
    schedule='@daily',
    catchup=False,
) as dag:

    task1 = BashOperator(
        task_id='print_date',
        bash_command='date'
    )

    task2 = BashOperator(
        task_id='print_hello',
        bash_command='echo "Hello Airflow!"'
    )

    task1 >> task2

Start Airflow:

# Terminal 1: Start scheduler
airflow scheduler

# Terminal 2: Start webserver
airflow webserver --port 8080

Access UI:

• Open browser to http://localhost:8080
• Login with your admin credentials
• Find your DAG and trigger it

Simple ETL Pipeline

from datetime import datetime
from airflow import DAG
from airflow.decorators import task

@DAG(
    dag_id='simple_etl',
    start_date=datetime(2023, 1, 1),
    schedule='@daily',
    catchup=False,
)
def simple_etl_dag():

    @task
    def extract():
        """Extract data from source"""
        import json
        data = '{"orders": [{"id": 1, "amount": 100}, {"id": 2, "amount": 200}]}'
        return json.loads(data)

    @task
    def transform(data):
        """Transform the data"""
        total = sum(order['amount'] for order in data['orders'])
        return {'total_amount': total, 'order_count': len(data['orders'])}

    @task
    def load(summary):
        """Load results"""
        print(f"Processed {summary['order_count']} orders")
        print(f"Total amount: ${summary['total_amount']}")

    # Build pipeline
    data = extract()
    summary = transform(data)
    load(summary)

# Instantiate the DAG
simple_etl_dag()

When to Use Airflow

Ideal Use Cases

✅ Perfect For:

• Batch data processing workflows
• ETL/ELT pipelines
• Machine learning training pipelines
• Data warehouse maintenance
• Multi-step transformations with dependencies
• Scheduled report generation
• Data validation and quality checks
• Cross-system orchestration
• Event-driven data workflows
• Backfilling historical data

❌ Not Ideal For:

• Real-time streaming (use Kafka, Flink instead)
• Simple cron jobs (Airflow adds complexity)
• Low-latency requirements (< 1 second)
• Infinitely running services
• Single-task workflows

Airflow vs Alternatives

Airflow vs Luigi:

• Airflow: Better UI, more features, larger community
• Luigi: Simpler, lighter weight

Airflow vs Prefect:

• Airflow: More mature, larger ecosystem
• Prefect: More modern, cloud-native design

Airflow vs Dagster:

• Airflow: Workflow orchestration focus
• Dagster: Data asset management focus

Airflow vs Cron:

• Airflow: Complex dependencies, monitoring, retries
• Cron: Simple time-based scheduling

Core Workflow Patterns

Pattern 1: Linear Pipeline

extract >> transform >> load

Use when: Simple sequential processing

Pattern 2: Fan-Out / Fan-In

extract >> [transform_a, transform_b, transform_c] >> combine >> load

Use when: Parallel processing with final aggregation

Pattern 3: Branching

check >> [process_a, process_b]  # Only one runs based on condition

Use when: Conditional execution paths

Pattern 4: Asset-Based

# Producer DAG updates asset
producer_task(outlets=[Asset("data.csv")])

# Consumer DAG triggered by asset
with DAG(schedule=[Asset("data.csv")]):
    consumer_task()

Use when: Event-driven workflows based on data availability

Pattern 5: Dynamic Mapping

@task
def process_file(filename):
    # Process single file
    pass

# Dynamically create tasks for each file
process_file.expand(filename=["file1.csv", "file2.csv", "file3.csv"])

Use when: Number of tasks depends on runtime data

Development Workflow

1. Design Phase

• Identify workflow steps and dependencies
• Determine schedule or trigger mechanism
• Plan data flow between tasks
• Identify external dependencies

2. Implementation

• Create DAG file in dags/ folder
• Define tasks using operators or TaskFlow
• Set dependencies
• Add error handling and retries
• Configure alerts and monitoring

3. Testing

• Unit test individual functions
• Test DAG structure (no import errors)
• Test task execution with sample data
• Verify idempotency

4. Deployment

• Deploy DAG to production DAGs folder
• Monitor first few runs
• Validate outputs
• Set up alerts

5. Maintenance

• Monitor execution metrics
• Handle failures
• Optimize performance
• Update for changing requirements

Common Configuration

airflow.cfg Essentials

[core]
dags_folder = /path/to/dags
executor = LocalExecutor  # or CeleryExecutor, KubernetesExecutor
sql_alchemy_conn = postgresql+psycopg2://user:pass@localhost/airflow
parallelism = 32  # Max tasks across all DAGs
max_active_runs_per_dag = 16

[scheduler]
dag_dir_list_interval = 300  # How often to scan for new DAGs
catchup_by_default = False

[webserver]
web_server_port = 8080
base_url = http://localhost:8080

Environment Variables

export AIRFLOW_HOME=~/airflow
export AIRFLOW__CORE__EXECUTOR=LocalExecutor
export AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:airflow@localhost/airflow

Production Deployment Considerations

High Availability

• Run multiple schedulers (Airflow 2.0+)
• Use external database (PostgreSQL, MySQL)
• Load balance web servers
• Use message broker for Celery (Redis, RabbitMQ)

Scalability

• Choose appropriate executor (Kubernetes for large scale)
• Configure task concurrency limits
• Optimize DAG parsing
• Use connection pooling

Security

• Enable RBAC
• Use secrets backend (AWS Secrets Manager, Vault)
• Encrypt connections
• Implement audit logging
• Secure webserver with HTTPS

Monitoring

• Set up metrics collection (StatsD, Prometheus)
• Configure alerting (email, Slack, PagerDuty)
• Monitor task duration and failure rates
• Track queue sizes and worker health
• Set up log aggregation (CloudWatch, Datadog)

Resource Management

• Set appropriate resource limits for tasks
• Use pools to limit concurrent resource usage
• Configure task queues for heterogeneous workers
• Implement task priority weights

Integration Ecosystem

Airflow integrates with 1000+ services through providers:

• Cloud Platforms: AWS, GCP, Azure, Alibaba Cloud
• Databases: PostgreSQL, MySQL, MongoDB, Cassandra, Snowflake, Redshift, BigQuery
• Data Processing: Spark, Flink, Databricks
• Container Orchestration: Kubernetes, Docker
• Message Queues: Kafka, RabbitMQ, SQS
• Data Tools: dbt, Great Expectations, Airbyte
• Monitoring: Datadog, New Relic, Prometheus
• Notifications: Slack, Email, PagerDuty, MS Teams

Learning Path

Beginner

1. Understand DAG concepts 2. Create simple BashOperator and PythonOperator tasks 3. Set task dependencies 4. Use the web UI 5. Understand XComs basics

Intermediate

1. TaskFlow API 2. Sensors and external dependencies 3. Dynamic task mapping 4. Error handling and retries 5. Asset-based scheduling 6. TaskGroups for organization

Advanced

1. Custom operators and sensors 2. Kubernetes executor 3. Performance optimization 4. Complex dynamic workflows 5. Production deployment patterns 6. Monitoring and alerting 7. Security hardening

Best Practices Summary

1. DAG Design: Keep DAGs focused and simple 2. Idempotency: Make tasks safe to re-run 3. Resource Management: Set appropriate concurrency limits 4. Error Handling: Use retries and callbacks 5. Monitoring: Implement comprehensive logging and metrics 6. Testing: Test DAGs before deploying to production 7. Documentation: Document DAG purpose and task logic 8. Version Control: Keep DAGs in Git 9. Secrets: Never hardcode credentials 10. Performance: Optimize heavy tasks, use appropriate executors

Getting Help

• Documentation: https://airflow.apache.org/docs/
• GitHub Issues: https://github.com/apache/airflow/issues
• Stack Overflow: Tag apache-airflow
• Slack Community: https://apache-airflow.slack.com
• Mailing Lists: dev@airflow.apache.org

Next Steps

1. Review the comprehensive SKILL.md for detailed concepts 2. Explore EXAMPLES.md for 18+ real-world patterns 3. Build your first production DAG 4. Join the Airflow community 5. Contribute to the ecosystem

---

Version: 1.0.0 Airflow Compatibility: 2.0+ Last Updated: January 2025