Airflow Data Quality: The Rerun That Loaded Twice!

Every Airflow load task has a rerun problem. The DAG fails, you clear it, it runs again — and if the load appends instead of upserts, the table now holds two copies of everything. This is the most common data-quality failure in production, and it's almost never caught by the DAG's own success status.

Our version: an analytics table got double-loaded by a manual rerun. Nothing failed. The DAG was green, the checks were 'fine', and the dashboards quietly showed every number twice.

The fix is boring and permanent: idempotent loads, quality checks as real tasks, and gates placed before consumers can touch the data.

A green DAG is a promise that the work ran. It says nothing about the data.

1. Idempotency: The Rerun Safety Net

Idempotency means running the same operation twice produces the same end state as running it once. For loads, that's an upsert by business key or a write into a partition boundary that gets replaced.

Airflow reruns are a fact of life: clears, backfills, catchups, manual triggers. Every one of them is safe when the load is idempotent and dangerous when it isn't.

Our load was INSERT INTO analytics.orders SELECT ... FROM staging.orders. Pure append. The second run didn't fail — it duplicated.

An append-only load is a promise that you'll never rerun. You will rerun.

UPSERT_ORDERS_SQL = """
INSERT INTO analytics.orders (order_id, customer_id, total, loaded_at)
SELECT order_id, customer_id, total, %(run_ts)s
FROM staging.orders
ON CONFLICT (order_id)
DO UPDATE SET customer_id = EXCLUDED.customer_id,
              total = EXCLUDED.total,
              loaded_at = EXCLUDED.loaded_at;
"""

load_orders = PostgresOperator(
    task_id="load_orders_upsert",
    postgres_conn_id="analytics_prod",
    sql=UPSERT_ORDERS_SQL,
    parameters={"run_ts": "{{ data_interval_end }}"},
)

2. Quality Checks as First-Class Tasks

A quality check is not a dashboard or a notebook cell. It's a task in the DAG with a dependency edge, and when it fails, the DAG fails.

Row-count parity between staging and target catches lost and duplicate batches. Null-rate checks catch broken transforms. Freshness checks catch stale partitions.

Each check is a task that raises on violation. No raise, no failure — and a check that only logs is a lie wearing a green icon.

The graph view should show it: load, then gate, then consumers. The gate is not optional; it's the reason the DAG exists.

You don't have to hand-roll every gate: the common SQL provider ships six ready-made check operators — SQLCheckOperator, SQLColumnCheckOperator, SQLTableCheckOperator, SQLIntervalCheckOperator, SQLValueCheckOperator, and SQLThresholdCheckOperator. SQLCheckOperator runs any query and fails the task when a returned value is falsy in Python — SELECT COUNT(*) FROM foo fails only when the count is 0. SQLColumnCheckOperator runs predefined column checks (null_check, min, max, distinct_check) from a column_mapping dict, with partition_clause to scope rows and tolerance for fuzzy bounds. Astronomer recommends the column and table operators over the older value and threshold variants.

def verify_row_parity(**context):
    staging = count_rows("staging.orders", conn="analytics_prod")
    target = count_rows("analytics.orders", conn="analytics_prod")
    if staging != target:
        raise AirflowException(
            f"Row-count mismatch: staging={staging}, target={target}"
        )


verify_parity = PythonOperator(
    task_id="verify_row_parity",
    python_callable=verify_row_parity,
)

3. Row-Count and Dedupe Gates

Two gates catch the double-load class of bug specifically: the dedupe gate and the parity gate.

The dedupe gate runs before the load and after it: group the source by business key and fail if any key appears twice. The parity gate compares source and target counts after the load.

Neither is expensive. Both are one SQL statement each, run as tasks, and they turn the rerun failure from a silent double into a loud, green-checked DAG that stops itself.

Double-loaded analytics is exactly what these two gates exist to prevent.

SQLTableCheckOperator is the natural home for the parity rule: pass checks like {"row_count_check": {"check_statement": "COUNT(*) >= 3"}} and the operator fails when any statement evaluates false. For drift that isn't duplication — values slowly creeping out of range — SQLIntervalCheckOperator compares today's metric against the same metric from days ago, catching slow decay that a hard parity gate misses.

SELECT order_id, COUNT(*) AS copies
FROM staging.orders
GROUP BY order_id
HAVING COUNT(*) > 1
LIMIT 10;

4. Lineage for Downstream Trust

Lineage answers the question users actually ask: where did this number come from? Airflow shows task and DAG relationships in the grid view; Airflow 3's assets make the producer-consumer links explicit.

When a downstream team sees a warning on their dashboard, lineage tells them which DAG fed it, which task wrote it, and which gate checked it. That turns 'your data is wrong' into 'the orders DAG failed the parity gate'.

Lineage is also the trust that makes quality gates acceptable to consumers: they can see the gates exist.

A table without lineage is a rumor. A table with lineage is an argument.

5. Data Contracts with Producers

A data contract is the written agreement between the producer of a table and its consumers: schema, key, freshness, volume range. When the producer breaks the contract, the pipeline should fail at ingest — not silently corrupt downstream.

A validation task at the ingest step checks the shape of the incoming batch: required columns, unique keys, null rates within bounds.

Contracts turn quality from a pipeline owner's hobby into a team standard. The producer can't change a column and hope; the contract check catches it.

No contract means every consumer re-verifies. That's the tax you stop paying.

6. The Failure Story: Double-Loaded Analytics

Monday, 09:12 — an engineer cleared the orders load because a transient DB blip failed the last task. The rerun went green. Nothing complained.

Tuesday, 10:00 — the finance dashboard showed revenue at exactly double the previous day's. The team's first instinct was a query bug. It wasn't.

The load was INSERT ... SELECT with no dedupe key. The first run had committed 48,211 rows before failing on a later task; the clear re-ran the whole DAG, and the load inserted the same 48,211 rows again.

Two facts made this inevitable: append-only load, and no gate between load and consumers. One upsert would have made the rerun harmless. One parity gate would have made it loud.

It was a Tuesday afternoon to fix. It took a Monday morning to notice.

7. The Quality-First DAG Pattern

The shape is always the same: extract, load idempotently, gate, then let consumers run. Extract and load stay atomic; gates sit between load and the rest of the world.

This pattern makes clears safe, backfills safe, and Monday-morning fire drills unnecessary. The DAG is a machine that either produces verified data or stops itself.

Ship the gates with the load — they're not a phase-two initiative. The day they ship is the day reruns stop being dangerous.

One deliberate staging exception: a soft gate. Airflow fails a run only when a leaf task ends failed, so a leaf with trigger_rule="all_done" runs regardless of upstream check results — the run stays green while on_failure_callback still fires. Use it where you want a visible warning, not a stop; production gates stay hard.

with DAG(
    dag_id="orders_quality_gated",
    schedule="@daily",
    catchup=False,
    default_args=default_args,
) as dag:
    extract = PythonOperator(
        task_id="extract_orders", python_callable=extract_orders
    )
    load = PostgresOperator(
        task_id="load_orders_upsert",
        postgres_conn_id="analytics_prod",
        sql=UPSERT_ORDERS_SQL,
    )
    check_dedup = PostgresOperator(
        task_id="check_no_duplicates",
        postgres_conn_id="analytics_prod",
        sql=DEDUPE_CHECK_SQL,
    )
    check_parity = PythonOperator(
        task_id="verify_row_parity", python_callable=verify_row_parity
    )

    extract >> load >> check_dedup >> check_parity