Apache Airflow in Production: DAG Design, Dynamic Tasks, XCom, and Celery Executor

Apache Airflow is the most widely adopted workflow orchestration platform for data pipelines. Its Python-native DAG authoring, rich operator ecosystem, and powerful scheduler make it suitable for everything from nightly ETL jobs to complex ML training pipelines. This guide covers DAG design principles, dynamic task mapping, XCom communication, the Celery executor, and production operations best practices.

Airflow Architecture

Airflow consists of five main components:

• Scheduler: Parses DAGs, schedules task instances, and submits them to the executor
• Executor: Determines how tasks run (LocalExecutor, CeleryExecutor, KubernetesExecutor)
• Workers: Execute task instances (in Celery/Kubernetes mode)
• Metadata Database: Stores DAG definitions, task states, XCom values, and logs (PostgreSQL recommended)
• Web Server: Provides the Airflow UI for monitoring and management

TaskFlow API (Airflow 2.x)

The TaskFlow API uses Python decorators for clean, readable DAG authoring:

from airflow.decorators import dag, task
from airflow.utils.dates import days_ago
from datetime import timedelta

default_args = {
    'owner': 'data-engineering',
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
    'retry_exponential_backoff': True,
    'max_retry_delay': timedelta(minutes=60),
    'email_on_failure': True,
    'email': ['data-alerts@company.com'],
}

@dag(
    dag_id='ecommerce_daily_etl',
    default_args=default_args,
    schedule_interval='0 2 * * *',
    start_date=days_ago(1),
    catchup=False,
    max_active_runs=1,
    tags=['etl', 'ecommerce'],
)
def ecommerce_daily_etl():

    @task()
    def extract_orders(ds: str) -> list:
        from mylib.db import query_postgres
        return query_postgres(
            "SELECT * FROM orders WHERE date(created_at) = %s", params=[ds])

    @task()
    def transform_orders(raw_orders: list) -> list:
        return [
            {'order_id': r['id'],
             'amount_usd': r['amount'] / 100.0,
             'status': r['status'].lower()}
            for r in raw_orders
        ]

    @task()
    def load_orders(transformed: list) -> int:
        from mylib.warehouse import bulk_upsert
        return bulk_upsert('fct_orders', transformed, key='order_id')

    raw = extract_orders()
    transformed = transform_orders(raw)
    load_orders(transformed)

dag_instance = ecommerce_daily_etl()

DAG Design Principles

Idempotency: Every DAG run for the same execution_date must produce identical results. Use ds (the logical date) to scope queries, never NOW().

Atomicity: Each task should do exactly one thing. Avoid monolithic tasks that both extract and load — they are hard to retry and debug.

No top-level I/O: Heavy imports or database connections at the module level execute during every scheduler parse cycle, degrading performance.

# BAD — connection opened every parse cycle
import psycopg2
conn = psycopg2.connect(...)

# GOOD — connection inside the task function
@task()
def my_task():
    import psycopg2
    conn = psycopg2.connect(...)

Avoid dynamic schedule-time logic: Do not compute schedule_interval from external state. The scheduler needs a stable, serializable DAG structure.

Dynamic Task Mapping (Airflow 2.3+)

Dynamic task mapping creates tasks at runtime based on upstream data — essential for processing a variable number of files, partitions, or API pages.

@dag(schedule_interval='@daily', start_date=days_ago(1), catchup=False)
def parallel_file_processing():

    @task()
    def list_files(ds: str) -> list[str]:
        import boto3
        s3 = boto3.client('s3')
        resp = s3.list_objects_v2(Bucket='data-lake', Prefix=f'raw/events/{ds}/')
        return [obj['Key'] for obj in resp.get('Contents', [])]

    @task()
    def process_file(s3_key: str) -> dict:
        rows = do_processing(s3_key)
        return {'key': s3_key, 'rows': rows}

    @task()
    def consolidate(results: list[dict]) -> None:
        total = sum(r['rows'] for r in results)
        print(f"Processed {total} total rows across {len(results)} files")

    files = list_files()
    processed = process_file.expand(s3_key=files)  # creates N parallel tasks
    consolidate(processed)

dag_instance = parallel_file_processing()

XCom for Inter-Task Communication

XCom (Cross-Communication) lets tasks share small values via the metadata database.

@task()
def get_record_count(ds: str) -> int:
    return query_count(ds)  # returns an integer

@task()
def validate_count(count: int) -> None:
    if count < 1000:
        raise ValueError(f"Expected >= 1000 records, got {count}")
    print(f"Validation passed: {count} records")

# TaskFlow automatically passes XCom values between decorated tasks
count = get_record_count()
validate_count(count)

XCom push/pull with traditional operators:

from airflow.operators.python import PythonOperator

def push_value(**context):
    context['ti'].xcom_push(key='record_count', value=42000)

def pull_value(**context):
    count = context['ti'].xcom_pull(task_ids='push_task', key='record_count')
    print(f"Got {count} records")

Important: XCom is stored in the metadata database. Keep values small (< 1 MB). For large> datasets, write to object storage and pass the S3 path via XCom instead.

Celery Executor Setup

The CeleryExecutor is the standard choice for distributed, multi-worker Airflow deployments.

docker-compose excerpt:

services:
  airflow-scheduler:
    image: apache/airflow:2.9.0
    environment:
      AIRFLOW__CORE__EXECUTOR: CeleryExecutor
      AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
      AIRFLOW__CELERY__BROKER_URL: redis://redis:6379/0
      AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql://airflow:airflow@postgres/airflow
    depends_on: [postgres, redis]

  airflow-worker:
    image: apache/airflow:2.9.0
    command: celery worker
    environment:
      AIRFLOW__CORE__EXECUTOR: CeleryExecutor
      AIRFLOW__CELERY__BROKER_URL: redis://redis:6379/0
    deploy:
      replicas: 4  # scale horizontally

Worker queues for task routing:

@task(queue='gpu-workers')
def train_model(config: dict) -> str:
    # This task will only run on workers in the 'gpu-workers' queue
    ...

@task(queue='default')
def lightweight_transform(data: list) -> list:
    ...

Start a worker listening on a specific queue:

airflow celery worker --queues gpu-workers

Sensors and External Triggers

FileSensor — wait for a file to appear before proceeding:

from airflow.sensors.filesystem import FileSensor

wait_for_file = FileSensor(
    task_id='wait_for_daily_dump',
    filepath='/mnt/data/dumps/{{ ds }}/orders.csv',
    poke_interval=300,   # check every 5 minutes
    timeout=7200,        # fail after 2 hours
    mode='reschedule',   # release worker slot while waiting
)

ExternalTaskSensor — wait for another DAG's task to complete:

from airflow.sensors.external_task import ExternalTaskSensor

wait_for_upstream = ExternalTaskSensor(
    task_id='wait_for_upstream_etl',
    external_dag_id='upstream_etl',
    external_task_id='load_complete',
    execution_delta=timedelta(hours=1),
    mode='reschedule',
    timeout=3600,
)

Alerting and SLA Monitoring

Task-level callbacks:

def on_failure_callback(context):
    from mylib.slack import send_alert
    task = context['task_instance']
    send_alert(f"Task {task.task_id} in {task.dag_id} failed on {task.execution_date}")

@task(on_failure_callback=on_failure_callback)
def critical_load() -> None:
    ...

SLA misses:

@dag(
    sla_miss_callback=lambda dag, task_list, blocking_task_list, slas, blocking_tis: notify_sla_breach(slas),
    default_args={'sla': timedelta(hours=3)},
)
def my_dag():
    ...

Performance Tuning

Scheduler tuning for high DAG count environments:

# airflow.cfg
[scheduler]
min_file_process_interval = 30       # parse each DAG file every 30 seconds
dag_dir_list_interval = 60           # scan DAG folder every 60 seconds
max_dagruns_to_create_per_loop = 10
max_callbacks_per_loop = 20

[core]
parallelism = 64                     # max concurrent tasks across all DAGs
max_active_tasks_per_dag = 16

Database optimization: Add indexes on dag_run, task_instance, and xcom tables. Archive old DAG runs regularly:

from airflow.utils.db_cleanup import run_cleanup

run_cleanup(
    clean_before_timestamp=pendulum.now().subtract(days=90),
    table_names=['dag_run', 'task_instance', 'xcom', 'log'],
    dry_run=False,
)

Production Checklist

Before promoting a DAG to production:

1. catchup=False unless backfilling is intentional
2. max_active_runs=1 for stateful pipelines
3. All tasks are idempotent (safe to re-run for the same date)
4. No top-level I/O or heavy imports in DAG files
5. Retry logic and exponential backoff configured
6. Failure callbacks and SLA monitoring set up
7. Large data passed via S3 path, not XCom directly
8. Sensors use mode='reschedule' to release worker slots

Conclusion

Apache Airflow production deployments require disciplined DAG design, idempotent tasks, appropriate use of XCom for small values, dynamic task mapping for variable workloads, a properly sized Celery executor cluster, and comprehensive monitoring via callbacks and SLA miss alerting. These practices produce reliable, observable pipelines that can be safely operated at scale.