Airflow Data Quality: The Rerun That Loaded Data Twice

Reruns are inevitable. Someone will click clear, a backfill will overlap, and your load task will run twice with the same data. The only question is whether the second run is a no-op or a disaster.

One team's rerun loaded every row twice. Analytics doubled overnight, the dashboard lied for days, and the cleanup meant reconciling millions of rows by hand. No gate caught it because no gate existed.

You'll build loads that converge on rerun and gates that fail loudly on bad batches. Twice the runs. Same rows.

Idempotency: The Rerun Safety Net

Idempotency means the second run changes nothing. For loads, that means keying every write on the business date plus the natural key and merging instead of appending. Reruns, retries, and backfills all converge on the same table state.

You'll implement it as an upsert in one transaction. Delete the partition's keys, insert the fresh batch, commit together. A crash mid-load leaves the old partition intact instead of a half-written mess.

Test it by running twice. Load Monday's partition, run the same task again, and diff the table. Identical means safe; doubled means the next real rerun becomes an incident.

Quality Checks as First-Class Tasks

Quality checks deserve to be tasks, not notebook queries someone runs monthly. As tasks they block promotion, appear in the Grid view, and page the owner when they fail. Visibility turns checks from folklore into guarantees.

You'll stage first, gate second, promote third. Extract lands in a staging table, gate tasks validate it, and only passing batches merge to serving. Failing batches quarantine in staging with their lineage intact for diagnosis.

Own each gate explicitly. The producing team owns extract freshness, the consuming team owns contract thresholds, and the DAG fails when either side breaks the deal. Shared ownership means shared paging, which means fast fixes.

Use the official GX provider instead of hand-rolled wrappers — it cuts boilerplate roughly 60%. Three operators cover three situations: GXValidateDataFrameOperator for in-memory Spark/pandas frames, GXValidateBatchOperator for data behind a BatchDefinition, and GXValidateCheckpointOperator when you want full Checkpoint power with actions on results. All three take Ephemeral (in-DAG only) or Cloud (persisted, linkable) contexts; only Checkpoint supports File contexts with Data Docs. Set fail_task_on_validation_failure to True on prod gates so bad batches never load — post-hoc warnings get ignored about 60% of the time.

Scale it with dynamic task mapping: keep a tables = [...] list and map one validation task per table instead of copy-pasting DAGs. Validation results land in XCom (use return_json_dict for a serializable dict), so downstream tasks can push null percents to Prometheus or branch on quality scores.

import pendulum
from airflow.sdk import dag, task
from airflow.providers.postgres.hooks.postgres import PostgresHook

@dag(
    dag_id="orders_daily",
    schedule="0 7 * * *",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    tags=["tier-1", "orders"],
)
def orders_daily():
    @task
    def load_partition(ds: str | None = None) -> str:
        hook = PostgresHook(postgres_conn_id="analytics_warehouse")
        hook.run(
            """
            BEGIN;
            DELETE FROM serving.orders WHERE business_date = %(ds)s;
            INSERT INTO serving.orders
            SELECT * FROM staging.orders WHERE business_date = %(ds)s
            ON CONFLICT (business_date, source_id) DO UPDATE
            SET amount = EXCLUDED.amount, status = EXCLUDED.status;
            COMMIT;
            """,
            parameters={"ds": ds},
        )
        return ds or ""

    load_partition()

orders_daily()

Row-Count and Dedupe Gates

Row-count bands catch empty and partial files. Compare today's staged count against the 30-day median; a batch at 1% of median is a broken extract, not a quiet day. You'll tune the band per feed because Black Friday is not a data incident.

Dedupe checks catch double loads directly. Group staging by the natural key and fail on any count above one. This single query would have stopped the incident batch before promotion.

Null-rate checks catch schema drift. When an upstream rename turns customer_id null for 40% of rows, the aggregate flags it in seconds. Each gate is one query, runs in seconds, and saves days.

Steal Soda's double-gate placement even if you don't run Soda. Gate one runs in memory (DuckDB over the inbound batch) before the write — aborts the DAG, stays local to Airflow logs, never spams Slack. Gate two runs against the live table after the write and publishes to Soda Cloud, where stewards, catalogs, and pagers see it. One YAML contract, two enforcement points: pre-prod blocks, production audits. Never publish staging noise to the shared channel or the team mutes quality alerts within a week.

Lean on SodaCL's check vocabulary: freshness (youngest row younger than X), schema (required columns, types), reference (values in table A must exist in table B), plus failed_rows with an expression or query for cross-column rules like ship-date-after-order-date. Give every check both a warn threshold (track the trend) and a fail threshold (stop the run) — warnings for drift you watch, failures for missing keys, invalid statuses, and schema breaks.

-- row-count band: today's batch vs 30-day median volume
WITH history AS (
  SELECT business_date, count(*) AS n
  FROM serving.orders
  WHERE business_date >= current_date - interval '30 days'
  GROUP BY 1
)
SELECT count(*) AS staged_today,
       (SELECT percentile_cont(0.5) WITHIN GROUP (ORDER BY n) FROM history) AS median_n
FROM staging.orders
WHERE business_date = '{{ ds }}';

-- dedupe: must return zero rows or the gate fails
SELECT business_date, source_id, count(*)
FROM staging.orders
WHERE business_date = '{{ ds }}'
GROUP BY 1, 2
HAVING count(*) > 1
LIMIT 20;

Lineage for Downstream Trust

Lineage answers what-loaded-what when counts look wrong. Record dag_id, run_id, and data_interval with every partition write, and the double-load investigation becomes one lookup instead of a week of archaeology.

You'll store it as partition metadata, not tribal knowledge. A small lineage table with partition, run_id, row count, and loaded_at timestamp serves both debugging and compliance audits. Append on every load; never update in place.

Expose it where analysts look. A dashboard column showing the producing run per partition lets the business self-serve the is-this-fresh question. Fewer pings, faster trust.

Data Contracts With Producers

Data contracts turn hallway promises into failing tasks. Producer and consumer agree on volume bands, null ceilings, key uniqueness, and freshness windows, then encode each as a gate both sides can see. Renegotiation happens in the open, not in a postmortem.

You'll version contracts like APIs. A v2 schema with renamed columns ships alongside a v2 gate, and the old gate retires only after consumers migrate. Breaking changes arrive as planned work, not surprises.

Enforce ownership on breach. The contract names the producer contact who gets paged, so a broken extract wakes the team that can fix it. You'll resolve most breaches before analysts finish their coffee.

import pendulum
from airflow.sdk import dag, task
from airflow.providers.postgres.hooks.postgres import PostgresHook

MEDIAN_TOLERANCE = 0.5  # fail if staged count below 50% of median

@dag(
    dag_id="orders_gated",
    schedule="0 7 * * *",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    tags=["tier-1", "orders"],
)
def orders_gated():
    @task
    def gate_row_count(ds: str | None = None) -> str:
        hook = PostgresHook(postgres_conn_id="analytics_warehouse")
        staged = hook.get_first(
            "SELECT count(*) FROM staging.orders WHERE business_date = %(ds)s",
            parameters={"ds": ds},
        )[0]
        median = hook.get_first(
            "SELECT percentile_cont(0.5) WITHIN GROUP (ORDER BY n) FROM "
            "(SELECT count(*) AS n FROM serving.orders "
            "WHERE business_date >= current_date - 30 GROUP BY 1) h",
        )[0] or 0
        if median and staged < median * MEDIAN_TOLERANCE:
            raise ValueError(f"row-count breach: staged={staged} median={median}")
        return f"rows_ok={staged}"

    gate_row_count()

orders_gated()

The Failure Story: Double-Loaded Analytics

The double-load played out in slow motion. A manual rerun appended the full batch, counts doubled, and every task glowed green because appending twice is technically success. Analysts found it; monitoring didn't.

Reconciliation took days. Without a dedupe key the team matched rows by timestamps, without lineage they guessed at run boundaries, and without gates there was no quarantine to roll back to. Each missing piece multiplied the cleanup.

The rebuilt pipeline makes repetition safe. Upserts converge, gates quarantine, lineage explains. You'll rerun with confidence now, which is the whole point: operations should be boring.