Apache Airflow for Data Pipelines – Complete Python Guide 2026

Apache Airflow is the industry-standard tool for orchestrating data pipelines. Whether you’re running daily ETL jobs, triggering ML model retraining, or coordinating microservices, Airflow lets you define complex workflows as code, schedule them, and monitor their execution from a rich web UI. This guide gets you productive with Airflow in 2026.

What Is Airflow?

Airflow is a workflow orchestration platform where you define pipelines as DAGs (Directed Acyclic Graphs) in Python. Each node in the DAG is a task, and edges define dependencies between tasks. Airflow schedules tasks based on their dependencies and a configurable schedule (cron-style), retries failures automatically, and provides a web UI to monitor execution history, logs, and task state.

Installing Airflow

pip install "apache-airflow==2.9.0" --constraint   "https://raw.githubusercontent.com/apache/airflow/constraints-2.9.0/constraints-3.11.txt"

# Initialize database
airflow db init

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

# Start web server and scheduler (separate terminals)
airflow webserver --port 8080
airflow scheduler

Your First DAG

# dags/my_first_dag.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'dataexpertise',
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
    'email_on_failure': False,
}

with DAG(
    dag_id='daily_etl_pipeline',
    default_args=default_args,
    schedule='0 6 * * *',       # Run daily at 6 AM
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=['etl', 'daily'],
) as dag:

    def extract(**context):
        print("Extracting data from source...")
        data = {"records": 1000, "date": context['ds']}
        return data  # stored in XCom

    def transform(**context):
        data = context['ti'].xcom_pull(task_ids='extract')
        print(f"Transforming {data['records']} records for {data['date']}")
        return {"cleaned_records": data['records'] - 50}

    def load(**context):
        data = context['ti'].xcom_pull(task_ids='transform')
        print(f"Loading {data['cleaned_records']} records to warehouse")

    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)
    notify_task    = BashOperator(task_id='notify',
                                  bash_command='echo "Pipeline complete: {{ ds }}"')

    extract_task >> transform_task >> load_task >> notify_task

XComs – Passing Data Between Tasks

def push_data(**context):
    context['ti'].xcom_push(key='result', value={"rows": 500, "quality": 0.98})

def use_data(**context):
    result = context['ti'].xcom_pull(task_ids='push_task', key='result')
    print(f"Received: {result}")

XComs are stored in Airflow’s metadata database. Keep them small — use them for status/metadata, not large datasets. For large data, write to S3/GCS and pass only the path via XCom.

Sensors – Wait for External Events

from airflow.sensors.filesystem import FileSensor
from airflow.sensors.http_sensor  import HttpSensor

# Wait for a file to appear before processing
wait_for_file = FileSensor(
    task_id='wait_for_data_file',
    filepath='/data/input/daily_dump_{{ ds }}.csv',
    poke_interval=60,   # check every 60 seconds
    timeout=3600,       # fail after 1 hour
    mode='reschedule',  # release worker slot while waiting
)

# Wait for an API to return 200
wait_for_api = HttpSensor(
    task_id='wait_for_api',
    http_conn_id='my_api',
    endpoint='/health',
    poke_interval=30,
)

Branching – Conditional Workflows

from airflow.operators.python import BranchPythonOperator

def choose_branch(**context):
    hour = datetime.now().hour
    return 'full_refresh' if hour < 6 else 'incremental_load'

branch = BranchPythonOperator(task_id='choose_strategy', python_callable=choose_branch)
full_refresh    = PythonOperator(task_id='full_refresh', ...)
incremental     = PythonOperator(task_id='incremental_load', ...)

branch >> [full_refresh, incremental]

TaskFlow API – Cleaner DAG Definition

from airflow.decorators import dag, task

@dag(schedule='@daily', start_date=datetime(2026, 1, 1), catchup=False)
def modern_etl():

    @task
    def extract() -> dict:
        return {"records": 1000}

    @task
    def transform(data: dict) -> dict:
        return {"cleaned": data["records"] - 20}

    @task
    def load(data: dict):
        print(f"Loading {data['cleaned']} records")

    load(transform(extract()))

modern_etl()

The TaskFlow API (Airflow 2.x) uses Python decorators to define tasks. Data is automatically passed between tasks via XCom with no manual push/pull needed.

Production Best Practices

Use Docker Compose or Kubernetes for production Airflow deployment (the official Docker Compose setup handles all components). Always set catchup=False unless you explicitly need backfill runs. Use Airflow Connections to manage database credentials, API keys, and cloud credentials — never hardcode them in DAG files. Set meaningful retries and retry_delay on every task. Use mode='reschedule' on sensors to avoid blocking worker slots. Tag your DAGs for easy filtering in the UI.

Conclusion

Apache Airflow transforms ad-hoc scripts into managed, monitored, production-grade pipelines. The key concepts — DAGs, operators, sensors, XComs, and the schedule — are learnable in a day. The investment pays off immediately: you get automatic scheduling, retry logic, a monitoring UI, and a code-first approach to pipeline definition that plays well with version control and CI/CD.