Data Pipeline Orchestration Pattern: DAG Scheduling, Retries, and Recovery

Scenario context

Pipeline orchestration is an operational control plane problem that requires explicit dependency, retry, and backfill contracts.

flowchart TD
  A[Scheduler: create interval run] --> B[Sensor: wait for upstream data]
  B --> C[Task A: extract to staging]
  C --> D[Task B: transform partition]
  D --> E{Task B success?}
  E -->|Yes| F[Task C: validate data]
  E -->|No - retry 1| D
  E -->|No - retry 2| D
  E -->|No - retry 3 exhausted| G[Quarantine or skip]
  G --> H[Alert owner]
  F --> I{Validation pass?}
  I -->|Yes| J[Publish and advance watermark]
  I -->|No| K[Quarantine partition]
  K --> H

Observable outcome: 📊 DAG with Retry Logic and Recovery Branches This diagram expands on the basic flow by making retry behavior explicit: Task B retries up to three times before exhausting its budget and routing to quarantine.

Stage 6 — Practical Example: Airflow DAG With Explicit Retry and Publish Control

• This simplified Airflow DAG waits for a settlement file, stages one interval, and publishes only after validation succeeds.
• Why this matters operationally: 1.
• max active runs=1 stops catchup work from colliding with the same interval.
• The sensor timeout creates a clear late-data branch instead of indefinite hanging.

from datetime import timedelta

from airflow import DAG
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from airflow.utils.dates import days_ago

with DAG(
  dag_id="daily_settlement",
  schedule="0 4 * * *",
  start_date=days_ago(1),
  catchup=True,
  max_active_runs=1,
  default_args={
    "retries": 3,
    "retry_delay": timedelta(minutes=10),
  },
  tags=["finance", "freshness-critical"],
) as dag:
  wait_for_card_file = S3KeySensor(
    task_id="wait_for_card_file",
    bucket_name="raw-settlement",
    bucket_key="{{ data_interval_start.strftime('%Y-%m-%d') }}/cards.csv",
    timeout=60 * 60,
    poke_interval=300,
  )

  stage_settlement = SQLExecuteQueryOperator(
    task_id="stage_settlement",
    conn_id="warehouse",
    sql="sql/stage_settlement.sql",
  )

  validate_settlement = SQLExecuteQueryOperator(
    task_id="validate_settlement",
    conn_id="warehouse",
    sql="sql/validate_settlement.sql",
  )

  publish_settlement = SQLExecuteQueryOperator(
    task_id="publish_settlement",
    conn_id="warehouse",
    sql="sql/publish_settlement.sql",
  )

  wait_for_card_file >> stage_settlement >> validate_settlement >> publish_settlement

Observable outcome: The sensor timeout creates a clear late-data branch instead of indefinite hanging.

End-to-end outcome

After working through the stages, you should be able to explain:

• • Dataset freshness is a better alert target than raw task failure count.
• • Backfills need their own capacity plan or they will sabotage current SLAs.

Alternatives considered

Decision Guide: When You Need Full Orchestration

Adopt orchestration first for the datasets that already page humans.

Failure and degraded modes

Trade-offs & Failure Modes: Trade-offs and Failure Modes

• Detection: Orchestration is worth the overhead when multiple dependent datasets must be both correct and recoverable.
• User or operator impact: Orchestration is worth the overhead when multiple dependent datasets must be both correct and recoverable.