Pipeline Orchestration — Airflow, DAGs, Scheduling, and Dependency Management

WEBSERVER
  • Flask app serving the UI — graph view, Gantt chart, task logs, run history
  • Reads state from the metadata database (does not execute tasks)

SCHEDULER
  • The brain — runs continuously, parses DAG files every heartbeat (30s default)
  • Creates DagRuns when schedule intervals trigger, queues eligible tasks
  • Airflow 2.x supports multiple scheduler instances for HA

EXECUTOR
  • Receives queued task instances from the scheduler and runs them
  • SequentialExecutor: one task at a time, dev/testing only
  • LocalExecutor:      subprocesses on the scheduler machine, small teams
  • CeleryExecutor:     distributes to workers via Redis/RabbitMQ, horizontal scale
  • KubernetesExecutor: one pod per task, fully isolated, scales to zero — most common in 2026

METADATA DATABASE (PostgreSQL or MySQL)
  • Stores all state: DAG definitions, DagRuns, TaskInstances, XCom, pools
  • Source of truth — if the DB is down, Airflow stops

WORKERS
  • Actually execute the task code, write logs, report success/failure back to the DB

Task execution, start to finish:
1. Scheduler parses DAG file → creates DagRun at schedule time
2. Scheduler evaluates dependencies → marks eligible tasks QUEUED
3. Scheduler sends the TaskInstance to the Executor
4. Executor assigns the task to a Worker
5. Worker runs the task code, writes logs, reports SUCCESS/FAILURE to metadata DB
6. Scheduler sees SUCCESS → queues downstream tasks
7. UI reads state from metadata DB → task shows green

# DAG schedule: '0 6 * * *' (daily at 06:00 UTC)
# The run that executes at 2026-03-17 06:00 UTC has:
#   logical_date:       2026-03-16 06:00:00 UTC
#   data_interval_start: 2026-03-16 06:00:00 UTC
#   data_interval_end:   2026-03-17 06:00:00 UTC
# → at 06:00 on the 17th, all of the 16th's data is complete and safe to process

def process_orders(**context):
    run_date_wrong = datetime.now().strftime('%Y-%m-%d')                 # NOT reproducible on backfill
    run_date_right = context['data_interval_start'].strftime('%Y-%m-%d') # reproducible on backfill

# Jinja templates for the same value:
# {{ ds }}                  → '2026-03-16'
# {{ ds_nodash }}           → '20260316'
# {{ data_interval_start }} → '2026-03-16T06:00:00+00:00'

# Backfilling 2026-02-15 with the WRONG version:
run_date_wrong → today's actual date, no matter what's being backfilled — broken

# Backfilling 2026-02-15 with the RIGHT version:
run_date_right → '2026-02-14'  (logical_date for the Feb 15 run is Feb 14) — correct

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python   import PythonOperator
from airflow.operators.bash     import BashOperator
from airflow.operators.empty    import EmptyOperator
from airflow.utils.task_group   import TaskGroup
from airflow.models             import Variable

default_args = {
    'owner': 'data-team', 'depends_on_past': False,
    'retries': 2, 'retry_delay': timedelta(minutes=3),
    'retry_exponential_backoff': True,
    'execution_timeout': timedelta(minutes=30),
    'email_on_failure': True, 'email': ['data-team@freshcart.com'],
}

with DAG(
    dag_id          = 'freshcart_morning_pipeline',
    default_args    = default_args,
    description     = 'FreshCart daily data platform — Bronze → Silver → Gold',
    schedule        = '0 2 * * *',    # 02:00 UTC daily
    start_date      = datetime(2026, 1, 1),
    catchup         = False,          # do not backfill on deploy
    max_active_runs = 1,              # no concurrent runs
    tags            = ['daily', 'production', 'silver', 'gold'],
) as dag:
    start = EmptyOperator(task_id='start')
    end   = EmptyOperator(task_id='end')

with TaskGroup('extract', tooltip='Extract from all source systems') as extract_group:

    def make_extract_task(source: str, pool_slots: int = 1):
        """Factory for extraction tasks — avoids repetition."""
        def extract_fn(**context):
            from pipelines.extract import run_extraction
            run_date = context['data_interval_start'].strftime('%Y-%m-%d')
            run_extraction(source=source, run_date=run_date)

        return PythonOperator(
            task_id=f'extract_{source}', python_callable=extract_fn,
            pool='source_db_pool', pool_slots=pool_slots,
            sla=timedelta(minutes=15),
        )

    extract_orders     = make_extract_task('orders')
    extract_customers  = make_extract_task('customers')
    extract_payments   = make_extract_task('payments')
    extract_deliveries = make_extract_task('deliveries')
    # these four run in PARALLEL — no dependency between them within the group

dbt_silver = BashOperator(
    task_id='dbt_silver',
    bash_command=(
        'dbt run --target prod --select staging.* silver.* '
        '--vars \'{"run_date": "{{ ds }}"}\' '
        '&& dbt test --target prod --select staging.* silver.*'
    ),
    env={'DBT_PROFILES_DIR': '/etc/dbt', 'SNOWFLAKE_ACCOUNT': Variable.get('snowflake_account')},
    execution_timeout=timedelta(minutes=45), sla=timedelta(minutes=40),
)

dbt_gold = BashOperator(
    task_id='dbt_gold',
    bash_command=(
        'dbt run --target prod --select gold.* --vars \'{"run_date": "{{ ds }}"}\' '
        '&& dbt test --target prod --select gold.*'
    ),
    env={'DBT_PROFILES_DIR': '/etc/dbt'},
    execution_timeout=timedelta(minutes=20), sla=timedelta(minutes=15),
)

def run_quality_checks(**context):
    from pipelines.quality import check_all_gold_tables
    run_date = context['data_interval_start'].strftime('%Y-%m-%d')
    result   = check_all_gold_tables(run_date=run_date)
    if result.has_anomalies:
        raise ValueError(f'Quality checks failed: {result.summary}')
    context['ti'].xcom_push(key='quality_result', value=result.to_dict())

quality_checks = PythonOperator(task_id='quality_checks', python_callable=run_quality_checks,
                                 execution_timeout=timedelta(minutes=5))

def notify_finance(**context):
    from pipelines.notifications import send_daily_summary
    run_date       = context['data_interval_start'].strftime('%Y-%m-%d')
    quality_result = context['ti'].xcom_pull(task_ids='quality_checks', key='quality_result')
    send_daily_summary(run_date=run_date, quality=quality_result)

notify = PythonOperator(task_id='notify_finance', python_callable=notify_finance)

# ── the whole graph in one line ───────────────────────────────────────────────
start >> extract_group >> dbt_silver >> dbt_gold >> quality_checks >> notify >> end

Graph view — freshcart_morning_pipeline, 2026-03-17 02:00 run
start → [extract_orders, extract_customers, extract_payments, extract_deliveries]
      → dbt_silver → dbt_gold → quality_checks → notify_finance → end
Total duration: 38 min 12 s   Status: success

# airflow pools set source_db_pool 5 "Max 5 concurrent source DB connections"
# airflow pools set snowflake_pool  8 "Max 8 concurrent Snowflake queries"
# airflow pools set api_pool        3 "Max 3 concurrent API calls"

extract_orders = PythonOperator(
    task_id='extract_orders', python_callable=extract_orders_fn,
    pool='source_db_pool', pool_slots=1,   # heavy tasks can consume 2+ slots
)

# A REAL POOL STRATEGY FOR A MEDIUM PLATFORM:
# source_db_pool: 5   snowflake_pool: 8   api_pool: 3   dbt_pool: 2   default_pool: 16

schedule = '0 2 * * *'     # daily at 02:00 UTC
schedule = '*/15 * * * *'  # every 15 minutes
schedule = '@daily'        # Airflow shorthand for '0 0 * * *'
schedule = None            # manual trigger only

# ALWAYS use UTC for schedules — a schedule that silently shifts 30 minutes
# with another country's DST change is very hard to debug.

# catchup=True (default in some versions): a DAG paused 3 days creates
#   3 DagRuns for the missed intervals on resume — useful for date-range
#   pipelines, dangerous for high-frequency ones (hundreds of runs).
# catchup=False: only the latest interval runs on resume — set this
#   explicitly on production DAGs, never rely on the default.

with DAG(catchup=False, max_active_runs=1, ...): ...

from airflow import Dataset

ORDERS_SILVER    = Dataset('snowflake://freshcart/silver/orders')
CUSTOMERS_SILVER = Dataset('snowflake://freshcart/silver/customers')
PAYMENTS_SILVER  = Dataset('snowflake://freshcart/silver/payments')

# Producer DAG: declares which dataset a task produces
with DAG('orders_silver_pipeline', schedule='0 2 * * *') as dag:
    load_orders = PythonOperator(task_id='load_orders', python_callable=run_orders_pipeline,
                                  outlets=[ORDERS_SILVER])

# Consumer DAG: triggers when ALL three listed datasets have been updated
with DAG(dag_id='gold_daily_revenue', schedule=[ORDERS_SILVER, CUSTOMERS_SILVER, PAYMENTS_SILVER]) as dag:
    build_gold = PythonOperator(task_id='build_gold_revenue', python_callable=run_gold_pipeline)

from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.sensors.external_task import ExternalTaskSensor

wait_for_vendor_file = S3KeySensor(
    task_id='wait_for_shipfast_weekly_file',
    bucket_key='s3://freshcart-landing/shipfast/weekly_deliveries_{{ ds_nodash }}.csv',
    poke_interval=300, timeout=7200,
    mode='reschedule',   # releases the worker slot between polls — never 'poke'
    soft_fail=True,      # SKIPPED (not FAILED) on timeout — DAG continues
)

wait_for_upstream = ExternalTaskSensor(
    task_id='wait_for_orders_silver', external_dag_id='orders_silver_pipeline',
    external_task_id='load_orders', allowed_states=['success'],
    mode='reschedule', poke_interval=60, timeout=3600,
)

def check_source_row_count(**context) -> bool:
    """Return True once source has >= 1000 rows for today's date."""
    run_date = context['data_interval_start'].strftime('%Y-%m-%d')
    count    = get_source_row_count(run_date)
    if count >= 1000:
        return True
    print(f'Source has {count} rows — waiting for at least 1000')
    return False

wait_for_data = PythonSensor(
    task_id='wait_for_source_data', python_callable=check_source_row_count,
    poke_interval=180, timeout=7200, mode='reschedule',
)

Task log — wait_for_source_data, poll 4 of ~24:
Source has 640 rows — waiting for at least 1000
[reschedule] releasing worker slot, will check again in 180s
...
Source has 1120 rows — condition met, proceeding

airflow dags backfill --dag-id freshcart_morning_pipeline \
    --start-date 2026-01-01 --end-date 2026-03-16 \
    --max-active-runs 3     # run 3 days in parallel

airflow dags backfill --dag-id freshcart_morning_pipeline \
    --start-date 2026-03-15 --end-date 2026-03-15   # single date

airflow dags backfill --dag-id freshcart_morning_pipeline \
    --start-date 2026-01-01 --end-date 2026-03-16 --dry-run   # shows what would run

Backfill: freshcart_morning_pipeline, 2026-01-01 → 2026-03-16 (75 days)
Running with max_active_runs=3: 25 batches
[1/25] 2026-01-01 ... 2026-01-03: RUNNING
[2/25] 2026-01-04 ... 2026-01-06: RUNNING
...
Backfill complete: 75 succeeded, 0 failed

# Clear a specific task and everything downstream of it — re-runs on next heartbeat
airflow tasks clear --dag-id freshcart_morning_pipeline \
    --task-id dbt_gold --start-date 2026-03-17 --downstream

# Manual trigger with custom configuration:
airflow dags trigger --dag-id freshcart_morning_pipeline \
    --conf '{"run_date": "2026-03-17", "force_full_reload": true}'

# Reading the conf in the DAG:
def run_fn(**context):
    conf         = context['dag_run'].conf or {}
    run_date     = conf.get('run_date', context['ds'])
    force_reload = conf.get('force_full_reload', False)

1. IDEMPOTENCY IS ESSENTIAL
   Backfills re-run pipelines for dates that may already be processed.
   A non-idempotent pipeline (plain INSERT) creates duplicates. Use upserts.

2. BACKFILL RATE
   90 days at max_active_runs=3, 10 min/run → 30 batches × 10 min = 5 hours.
   Plan backfills during low-traffic hours.

3. DEPENDENCY ORDERING
   Backfills respect task dependencies WITHIN a DAG, not ACROSS DAGs.
   If Gold depends on Silver, backfill Silver first, then Gold.

4. SOURCE AVAILABILITY
   Historical data must still exist in the source — a CDC pipeline
   backfilling 90 days needs 90 days of WAL, or a separate bulk extract.

from airflow.decorators import task, dag

@dag(dag_id='process_all_stores', schedule='0 6 * * *', start_date=datetime(2026, 1, 1))
def process_all_stores_dag():

    @task
    def get_active_stores() -> list[str]:
        conn = get_db_connection()
        rows = conn.execute("SELECT store_id FROM reference.stores WHERE is_active = TRUE").fetchall()
        return [row[0] for row in rows]   # ['ST001', ..., 'ST010']

    @task
    def process_store_data(store_id: str, **context) -> dict:
        run_date = context['ds']
        result   = run_store_pipeline(store_id=store_id, run_date=run_date)
        return {'store_id': store_id, 'rows_written': result.rows_written}

    @task
    def aggregate_results(store_results: list[dict]) -> None:
        total = sum(r['rows_written'] for r in store_results)
        print(f'All stores complete: {len(store_results)} stores, {total} total rows')

    stores        = get_active_stores()
    store_results = process_store_data.expand(store_id=stores)   # one task per store, in parallel
    aggregate_results(store_results)

dag = process_all_stores_dag()

Graph view: process_all_stores, 2026-03-17
get_active_stores → process_store_data[0..9] (parallel, per store) → aggregate_results
All stores complete: 10 stores, 812,400 total rows

@task
def process_store_category(store_id: str, category: str) -> dict:
    return run_pipeline(store_id=store_id, category=category)

combinations = [
    {'store_id': 'ST001', 'category': 'grocery'},
    {'store_id': 'ST001', 'category': 'beverages'},
    {'store_id': 'ST002', 'category': 'grocery'},
]
results = process_store_category.expand_kwargs(combinations)   # 3 task instances

def extraction_task(**context):
    result = run_extraction(run_date=context['ds'])
    context['ti'].xcom_push(key='rows_extracted', value=result.rows_extracted)
    context['ti'].xcom_push(key='rows_rejected',  value=result.rows_rejected)
    # XCom value limit: ~48 KB default in PostgreSQL VARCHAR — keep it small

def quality_check_task(**context):
    ti = context['ti']
    rows_extracted = ti.xcom_pull(task_ids='extract_orders', key='rows_extracted')
    rows_rejected  = ti.xcom_pull(task_ids='extract_orders', key='rows_rejected')
    if rows_extracted == 0:
        raise ValueError('No rows extracted — possible source outage')

    rejection_rate = rows_rejected / rows_extracted
    if rejection_rate > 0.05:
        raise ValueError(f'Rejection rate {rejection_rate:.1%} exceeds 5% threshold')

from airflow.decorators import task

@task
def extract_orders(run_date: str) -> dict:
    result = run_extraction(run_date=run_date)
    return {'rows_extracted': result.rows_extracted, 'rows_rejected': result.rows_rejected}

@task
def quality_check(extraction_result: dict) -> None:
    if extraction_result['rows_extracted'] == 0:
        raise ValueError('No rows extracted')

# In the DAG:
result = extract_orders(run_date='{{ ds }}')
quality_check(result)   # result is passed as XCom automatically

# BAD — passing 500 MB through the metadata database
@task
def load_data_bad(**context):
    df = pd.read_csv('s3://bucket/orders.csv')
    context['ti'].xcom_push(key='dataframe', value=df.to_dict())   # crashes Airflow

# GOOD — write the data, push only the path
@task
def load_data_good(**context):
    df = pd.read_csv('s3://bucket/orders.csv')
    output_path = f's3://bucket/tmp/run-{context["run_id"]}/orders.parquet'
    df.to_parquet(output_path)
    context['ti'].xcom_push(key='output_path', value=output_path)

ValueError: XCOM value exceeds maximum size (48 KB)
# this is exactly what load_data_bad above triggers — the fix is load_data_good's
# pattern: push the S3 path (a few dozen bytes), not the data itself

The SLA for the morning pipeline is 10:30 PM ET (previous day). It used to complete by 10:10 PM ET. Over the last four weeks it has been completing later: 10:14, 10:21, 10:28, and this week it missed the SLA at 10:34 PM ET. No code was changed. You are asked to investigate.

-- Historical run durations from the Airflow metadata DB:
SELECT DATE(execution_date) run_date, ROUND(duration / 60.0, 1) total_minutes
FROM dag_run WHERE dag_id = 'freshcart_morning_pipeline' AND state = 'success'
  AND execution_date > NOW() - INTERVAL '30 days'
ORDER BY execution_date DESC;
-- 2026-03-17: 64 min ← SLA BREACH   2026-03-10: 51 min   2026-02-24: 38 min ← was fine
-- total duration grew 68% in 3 weeks — something is scaling linearly

-- Break duration down by task:
SELECT task_id, DATE(execution_date) run_date, ROUND(duration / 60.0, 1) minutes
FROM task_instance WHERE dag_id = 'freshcart_morning_pipeline' AND state = 'success'
  AND task_id IN ('extract_orders', 'dbt_silver', 'dbt_gold');
-- extract_orders: 8 → 8 → 8 → 8 min   (stable)
-- dbt_silver:    12 → 15 → 18 → 22 min ← growing linearly
-- dbt_gold:       4 → 4 → 4 → 4 min   (stable)

# dbt_silver's own log shows which model inside it is slow:
# Model staging.stg_orders completed in 42s
# Model silver.orders completed in 1280s   ← THIS ONE

-- Source table growth:
SELECT DATE(created_at) date, COUNT(*) daily_new_orders FROM raw.orders
GROUP BY 1 ORDER BY 1 DESC LIMIT 30;
-- FreshCart is growing: 48k/day → 52k → 56k → 60k

-- silver.orders model SQL:
-- SUM(order_amount) OVER (PARTITION BY store_id, month ORDER BY created_at)
-- This window function reads ALL historical orders on every run —
-- as the table grows, the model gets slower even though only today's rows are new.

Fix: materialise the monthly running total as a separate Gold model.
  silver.orders:              just cleans and validates (fast — only new rows)
  gold.monthly_store_revenue: the full window-function aggregate (slow, but runs once, persisted)

After fix:
2026-03-18  freshcart_morning_pipeline  2340s  39 min  ← back to baseline

The investigation used Airflow’s metadata database to isolate the slow task, dbt logs to isolate the slow model, and SQL analysis to understand the growth pattern. The fix was architectural — moving the expensive computation from an incremental Silver model (runs daily on all data) to a Gold model (runs once, result persisted).