Orchestration Patterns for Data Pipelines

When this skill applies

Orchestration is the layer that turns a collection of scripts into a reliable, observable pipeline. Reach for these patterns any time you're wiring up tasks that depend on each other, need to run on a schedule, or must recover gracefully from failures. The right design here saves enormous debugging time downstream.

Choosing an Orchestrator

The biggest split is task-centric (Airflow, Prefect) vs asset-centric (Dagster). Asset-centric thinking — "what data does this job produce?" rather than "what does this job do?" — makes lineage and freshness checks natural. If you're starting fresh and the team can learn, Dagster is worth the investment.

DAG Structure Principles

One task, one responsibility

Each task should do exactly one thing that can be individually retried, skipped, or monitored. Avoid "god tasks" that extract, transform, and load in one function — a failure anywhere forces the whole thing to rerun.

# Airflow — split extract/transform/load into separate operators
extract = PythonOperator(task_id="extract_events", python_callable=extract_events)
transform = PythonOperator(task_id="transform_events", python_callable=transform_events)
load = PythonOperator(task_id="load_events", python_callable=load_events)

extract >> transform >> load

Make tasks idempotent

A task is idempotent if running it twice produces the same result as running it once. This is the single most important property for reliable pipelines — it means retries are safe and backfills are predictable.

def load_events(ds, **context):
    # Delete-then-insert on partition date, not append
    conn.execute(f"DELETE FROM events WHERE event_date = '{ds}'")
    conn.execute(f"INSERT INTO events SELECT * FROM staging_events WHERE event_date = '{ds}'")

Parameterize by logical date, not wall-clock time

Use the DAG's logical execution date (ds in Airflow, scheduled_time in Prefect) rather than datetime.now(). This makes backfills predictable — when you rerun yesterday's DAG, it processes yesterday's data, not today's.

Retry and Failure Patterns

Tiered retry strategy

Not all failures are equal. Transient network errors warrant fast retries; upstream data delays warrant longer waits.

# Airflow default task retry config
default_args = {
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(minutes=60),
}

Use exponential backoff to avoid hammering a struggling upstream service. Cap the maximum delay so a task doesn't silently wait hours before alerting.

Alerting on failure

Failures should page someone or post to Slack — silent failures are worse than loud ones.

# Airflow — send Slack alert on failure
def alert_slack(context):
    SlackWebhookOperator(
        task_id="slack_alert",
        http_conn_id="slack_webhook",
        message=f"Task {context['task_instance'].task_id} failed on {context['ds']}",
    ).execute(context)

default_args = {"on_failure_callback": alert_slack}

SLA misses vs. task failures

An SLA miss means the task didn't finish within the expected window — even if it eventually succeeds. Set SLAs on tasks that feed downstream consumers with time expectations.

# Airflow SLA — alert if task hasn't completed within 2 hours of schedule
PythonOperator(
    task_id="load_daily_summary",
    python_callable=load_summary,
    sla=timedelta(hours=2),
)

Sensors and External Dependencies

Use sensors when a task depends on something outside the DAG — a file landing in S3, a table partition becoming available, or an upstream DAG completing.

# Airflow — wait for upstream partition
from airflow.sensors.external_task import ExternalTaskSensor

wait_for_upstream = ExternalTaskSensor(
    task_id="wait_for_events_dag",
    external_dag_id="events_pipeline",
    external_task_id="load_events",
    timeout=3600,         # give up after 1 hour
    poke_interval=120,    # check every 2 minutes
    mode="reschedule",    # free the worker slot while waiting
)

Prefer mode="reschedule" over mode="poke" for long waits — it frees the worker slot so other tasks can run.

Backfill Strategies

Backfills reprocess historical data. The key is that your tasks must be idempotent (see above) and your DAG must be parameterized by logical date.

# Airflow — backfill a date range
airflow dags backfill my_dag --start-date 2024-01-01 --end-date 2024-01-31

# Run with max concurrency to avoid overwhelming upstream
airflow dags backfill my_dag --start-date 2024-01-01 --end-date 2024-01-31 --max-active-runs 3

For very large backfills, consider running in chunks and checking intermediate results rather than firing off 365 runs at once.

Prefect Patterns

from prefect import flow, task
from prefect.tasks import task_input_hash
from datetime import timedelta

@task(retries=3, retry_delay_seconds=60, cache_key_fn=task_input_hash, cache_expiration=timedelta(hours=1))
def extract(date: str) -> list:
    ...

@task
def transform(records: list) -> list:
    ...

@flow(name="daily-events-pipeline")
def events_pipeline(date: str):
    raw = extract(date)
    clean = transform(raw)
    load(clean)

Cache results with task_input_hash so a failed flow can resume from the last successful task rather than restarting from scratch.

Dagster Asset Patterns

from dagster import asset, AssetIn, FreshnessPolicy

@asset(freshness_policy=FreshnessPolicy(maximum_lag_minutes=60))
def raw_events(context) -> pd.DataFrame:
    ...

@asset(ins={"raw_events": AssetIn()})
def clean_events(raw_events: pd.DataFrame) -> pd.DataFrame:
    ...

Assets make freshness SLAs first-class — Dagster will alert when raw_events is stale. The lineage graph is automatic.

Common Pitfalls

Design Checklist

Before shipping a new DAG:

• [ ] Every task is idempotent — safe to retry or rerun
• [ ] All date references use logical execution date, not now()
• [ ] Retries configured with backoff and a max delay
• [ ] Failure alerts wired up (Slack, PagerDuty, email)
• [ ] SLAs set on tasks with downstream time commitments
• [ ] Sensors use reschedule mode for long waits
• [ ] Backfill tested on at least one historical date range