Data Pipelines skill for Claude Code

A pipeline is production-ready when any run can be re-executed for the same interval and produce the same result without duplicates, a backfill is one command, transient failures retry on their own while real failures alert a human, and no task depends on the wall clock.

Ground rules

• Every task processes an explicit partition (the run’s data interval or partition key), never “since last run” or now().
• Writes are idempotent: overwrite the partition, MERGE on a key, or delete-then-insert in one transaction. Never blind INSERT/append.
• Writes are atomic: write to a staging location, then swap/rename/commit.
• Orchestrators orchestrate. Heavy compute runs in the warehouse, Spark, or a container, not in the scheduler’s worker process. Pass references (paths, table names), not data.
• Retries only for transient errors (network, throttling, lock timeouts). Validation failures should fail fast and alert.
• Backfills over large ranges or deletes of existing partitions in production require user confirmation with the exact range.

Workflow

- [ ] 1. Define the dataset contract: grain, partition key, freshness target, owner
- [ ] 2. Split into tasks: extract -> load raw -> transform -> validate -> publish
- [ ] 3. Make each task idempotent and partition-scoped
- [ ] 4. Configure retries, timeouts, concurrency, schedule, catchup
- [ ] 5. Add data checks and alerting
- [ ] 6. Test locally (single partition), then backfill a small range
- [ ] 7. Document runbook: rerun, backfill, common failures

Orchestrator choice

Situation	Pick
Existing Airflow estate, many operators/integrations, time-based schedules	Airflow (3.x)
Asset-centric thinking, lineage, partitions, strong local dev and testing	Dagster
Python-first teams, dynamic workflows, minimal infrastructure	Prefect (3.x)
Only dbt on a schedule	dbt Cloud job or a simple scheduler; an orchestrator may be overkill

Stay with what the team runs unless there is a concrete pain the other tool fixes.

3. Idempotent write patterns

Target	Pattern
Partitioned table (BigQuery, Snowflake, Hive/Iceberg)	Overwrite partition dt = {{ ds }}
Table with business key	MERGE ... ON key WHEN MATCHED UPDATE WHEN NOT MATCHED INSERT
Postgres	BEGIN; DELETE ... WHERE dt = $1; INSERT ...; COMMIT; or INSERT ... ON CONFLICT
Object storage	Write s3://bucket/table/dt=2026-10-18/_tmp/..., then move/commit; or table format (Iceberg/Delta) transactions
External APIs (side effects)	Idempotency keys; record sent IDs; never resend on retry without checking

Airflow 3 example (TaskFlow)

from datetime import datetime, timedelta
from airflow.sdk import dag, task

@dag(
    schedule="@daily",
    start_date=datetime(2026, 1, 1),
    catchup=False,                 # default in Airflow 3; backfill explicitly instead
    max_active_runs=1,
    default_args={
        "retries": 3,
        "retry_delay": timedelta(minutes=5),
        "retry_exponential_backoff": True,
        "execution_timeout": timedelta(hours=1),
    },
    tags=["orders"],
)
def orders_daily():
    @task
    def extract(data_interval_start=None, data_interval_end=None) -> str:
        path = f"s3://lake/raw/orders/dt={data_interval_start:%Y-%m-%d}/orders.parquet"
        export_orders(start=data_interval_start, end=data_interval_end, dest=path)   # overwrite
        return path

    @task
    def load(path: str, ds=None) -> None:
        run_sql("delete from raw.orders where dt = %s", [ds])          # same transaction as insert
        copy_into("raw.orders", path, partition=ds)

    @task
    def validate(ds=None) -> None:
        n = scalar("select count(*) from raw.orders where dt = %s", [ds])
        if n == 0:
            raise ValueError(f"no orders loaded for {ds}")            # fail, do not retry forever

    validate_task = validate()
    load(extract()) >> validate_task

orders_daily()

Airflow 3 notes: imports come from airflow.sdk; schedule_interval is gone (use schedule); execution_date is replaced by logical_date; Datasets are now Assets. Use pools and max_active_tis_per_dag to protect shared systems.

Dagster example (partitioned assets)

import dagster as dg

daily = dg.DailyPartitionsDefinition(start_date="2026-01-01")

@dg.asset(partitions_def=daily, retry_policy=dg.RetryPolicy(max_retries=3, delay=60))
def raw_orders(context: dg.AssetExecutionContext) -> None:
    day = context.partition_key
    export_orders_for_day(day, overwrite=True)

@dg.asset_check(asset=raw_orders)
def raw_orders_not_empty() -> dg.AssetCheckResult:
    n = count_latest_partition()
    return dg.AssetCheckResult(passed=n > 0, metadata={"rows": n})

Prefect example

from prefect import flow, task

@task(retries=3, retry_delay_seconds=[10, 60, 300], timeout_seconds=3600)
def load_day(day: str) -> int:
    return merge_orders_for_day(day)

@flow(log_prints=True)
def orders(day: str):
    rows = load_day(day)
    print(f"{day}: {rows} rows merged")

The partition (day) is a parameter, so a backfill is a loop of runs with explicit days.

4. Scheduling and concurrency

• Schedule after upstream data is complete (sensor/asset trigger or a buffer), not at midnight when the source is still writing.
• max_active_runs=1 for pipelines whose runs must not overlap (stateful targets).
• Set execution_timeout on every task; hung tasks block pools forever otherwise.
• Use data-aware triggers (Airflow Assets, Dagster asset sensors/auto-materialize) instead of guessing cross-DAG timing.

6. Test and backfill

airflow dags list-import-errors
airflow dags test orders_daily 2026-10-18           # one full run locally, no scheduler
airflow tasks test orders_daily load 2026-10-18
airflow backfill create --dag-id orders_daily --from-date 2026-09-01 --to-date 2026-09-07   # Airflow 3; confirm range
dagster asset materialize -m my_pkg.definitions --select raw_orders --partition 2026-10-18

Backfill procedure: run one partition, verify the output, run a small range, check warehouse load and costs, then the full range with concurrency limits. Re-run one already-processed partition and confirm the row count does not change (idempotency proof).

Unit-test the transformation functions without the orchestrator; add a DAG integrity test (imports cleanly, no cycles, required tags/owners) in CI.

Failure handling

Failure	Response
Timeout / 429 / 503 from source	Retry with exponential backoff; respect Retry-After
Schema change in source	Fail with a clear message; alert owner; do not coerce silently
Empty or tiny partition	Fail validation; alert (source outage looks like “success” otherwise)
Duplicate keys	Fail; investigate source or merge logic
Partial write	Atomic write pattern prevents it; rerun the partition

Alert on: task failure after retries, SLA/freshness miss, validation failure. Route to the owning team with the DAG/asset, partition, and log link.

Gotchas

• Using datetime.now() in a task makes reruns and backfills compute the wrong window.
• Top-level code in Airflow DAG files runs on every parse; no DB calls or API requests at import time.
• XCom/return values are for small metadata; large payloads bloat the metadata DB.
• catchup=True on a DAG with an old start_date launches hundreds of runs at deploy.
• Retrying a non-idempotent task (append, email, payment) multiplies side effects.
• Timezones: schedules in UTC; data intervals in UTC; convert at presentation time.
• Changing a DAG’s start_date or schedule after runs exist confuses run history; create a new DAG ID for big changes.

Output

**Pipeline:** orders_daily (Airflow 3), partition = data interval (UTC day)
**Tasks:** extract (overwrite S3 partition) -> load (delete+insert in txn) -> validate (row count > 0, keys unique)
**Reliability:** 3 retries exp. backoff on transient errors, 1 h timeouts, max_active_runs=1
**Idempotency proof:** reran 2026-10-18 twice; raw.orders count stayed 48,211
**Backfill:** 2026-09-01..2026-09-30 completed (confirmed range), 30 runs, 0 failures
**Runbook:** docs/pipelines/orders_daily.md