Apache Airflow: DAG Orchestration, Setup & Best Practices

Apache Airflow is the industry-standard open-source platform for authoring, scheduling, and monitoring data pipelines as code. Define your workflows as Python DAGs and get full visibility into every task run, retry, and dependency.

What is Apache Airflow?

Apache Airflow is a platform to programmatically author, schedule, and monitor workflows. Originally created at Airbnb in 2014 and donated to the Apache Software Foundation in 2016, it has become the dominant open-source workflow orchestration tool for data engineering teams worldwide. Airflow represents pipelines as Directed Acyclic Graphs (DAGs) written in Python, which means pipelines can be dynamically generated, version-controlled in Git, and tested like any other code.

The Airflow scheduler continuously monitors all DAGs, triggers task instances when their dependencies are met, and distributes work to an executor. The web UI provides a graphical view of DAG runs, task logs, dependency graphs, and Gantt charts, making it straightforward to diagnose failures and monitor pipeline health in production.

Core Concepts

• DAG — a Python file defining tasks and their dependencies as a directed graph
• Operator — a template for a task: BashOperator, PythonOperator, SqlOperator, S3ToRedshiftOperator, etc.
• Sensor — a special operator that polls for a condition (file exists, API responds, S3 key present) before proceeding
• Task Instance — a specific run of an operator within a DAG run, with states: queued → running → success/failed
• XCom — mechanism for passing small values between tasks via the metadata database
• Connection — stored credentials for external systems (DB, API, cloud), referenced by conn_id
• Variable — key-value store for DAG configuration, optionally JSON-serialized
• Executor — the mechanism for running tasks: Local, Celery, or Kubernetes

DAG Definition with Operators & Sensors

# dags/etl_pipeline.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.providers.http.sensors.http import HttpSensor

default_args = {
    'owner': 'data-team',
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
    'email_on_failure': True,
}

def extract_data(**context):
    import requests
    response = requests.get('https://api.example.com/data')
    context['ti'].xcom_push(key='raw_data', value=response.json())

def transform_data(**context):
    raw = context['ti'].xcom_pull(key='raw_data', task_ids='extract')
    transformed = [{'id': r['id'], 'value': r['v'] * 2} for r in raw]
    context['ti'].xcom_push(key='transformed', value=transformed)

with DAG(
    dag_id='etl_pipeline',
    default_args=default_args,
    schedule_interval='@daily',
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['etl', 'production'],
) as dag:

    check_api = HttpSensor(
        task_id='check_api',
        http_conn_id='api_default',
        endpoint='/health',
        timeout=20,
        poke_interval=5,
    )

    extract = PythonOperator(
        task_id='extract',
        python_callable=extract_data,
    )

    transform = PythonOperator(
        task_id='transform',
        python_callable=transform_data,
    )

    load = BashOperator(
        task_id='load',
        bash_command='python /opt/scripts/load_to_warehouse.py',
    )

    check_api >> extract >> transform >> load

Docker Compose Quickstart

# docker-compose.yml (Airflow quickstart)
version: '3.8'
x-airflow-common: &airflow-common
  image: apache/airflow:2.9.0
  environment:
    AIRFLOW__CORE__EXECUTOR: LocalExecutor
    AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
    AIRFLOW__CORE__FERNET_KEY: ''
    AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: 'true'
    AIRFLOW__API__AUTH_BACKENDS: airflow.api.auth.backend.basic_auth
  volumes:
    - ./dags:/opt/airflow/dags
    - ./logs:/opt/airflow/logs
    - ./plugins:/opt/airflow/plugins
  depends_on:
    postgres:
      condition: service_healthy

services:
  postgres:
    image: postgres:15
    environment:
      POSTGRES_USER: airflow
      POSTGRES_PASSWORD: airflow
      POSTGRES_DB: airflow
    healthcheck:
      test: ["CMD", "pg_isready", "-U", "airflow"]
      interval: 5s

  airflow-webserver:
    <<: *airflow-common
    command: webserver
    ports:
      - "8080:8080"

  airflow-scheduler:
    <<: *airflow-common
    command: scheduler

  airflow-init:
    <<: *airflow-common
    command: version
    entrypoint: /bin/bash
    environment:
      _AIRFLOW_DB_UPGRADE: 'true'
      _AIRFLOW_WWW_USER_CREATE: 'true'
      _AIRFLOW_WWW_USER_USERNAME: admin
      _AIRFLOW_WWW_USER_PASSWORD: admin

XComs, Variables & Connections

# XComs and Connections
from airflow.models import Variable
from airflow.hooks.base import BaseHook

# Read Airflow Variable (encrypted in metastore)
api_key = Variable.get('my_api_key', deserialize_json=False)

# Read a Connection
conn = BaseHook.get_connection('my_postgres')
print(conn.host, conn.schema, conn.login)

# Push/pull XCom values between tasks
def producer(**context):
    context['ti'].xcom_push(key='result', value=42)

def consumer(**context):
    val = context['ti'].xcom_pull(
        task_ids='producer_task',
        key='result'
    )
    print(f"Got: {val}")  # 42

Airflow vs Prefect vs Dagster

Best Practices

• Set catchup=False on all production DAGs to prevent unintended backfills on re-deployment
• Keep tasks atomic and idempotent — each task should do one thing and be safely re-runnable
• Never import heavy libraries at DAG file top level — Airflow parses all DAG files frequently; slow imports degrade scheduler performance
• Use XComs only for small values (IDs, counts, paths) — write large datasets to S3/GCS and pass the path
• Store all credentials in Connections or Variables, never hardcoded in DAG files
• Tag DAGs with tags=[] and use consistent naming conventions for easy filtering in the UI