Idempotent SPC DAG Design for Reproducible Limits

A control limit that changes when you rerun the same window is not a control limit — it is a liability. Backfills, retries, and manual re-triggers are routine in an Airflow SPC pipeline, and every one of them must produce byte-identical limits from identical inputs, or the archived chart an auditor pulls will not match the one the operator saw. This how-to makes reruns deterministic — a calculation_hash, versioned partition writes, and a guarded Phase I baseline — as a focused step within orchestrating SPC pipelines with Apache Airflow.

Idempotency here means a task can run any number of times on the same inputs and leave the system in the same state: the same hash, the same limits, the same file. Getting there requires removing every source of nondeterminism — wall-clock timestamps, unstable ordering, appended writes — and treating the frozen Phase I baseline as immutable once established.

The stakes are higher than developer convenience. A control limit is the reference against which every out-of-control decision is judged, and if that reference silently shifts between the run an operator saw and the run an auditor reproduces, the entire chart loses its evidentiary value. Reruns are not exceptional events to be avoided — retries after a transient fault, backfills to fill a gap, and manual re-triggers after a fix are all normal operations. The design below makes them safe by construction rather than by hoping nobody presses the button twice.

Prerequisites

Confirm these before hardening the DAG:

  • Apache Airflow 2.4+ or 3.x with the compute task already producing a centerline, UCL, and LCL
  • Python 3.10+ with pandas >= 2.0 and pyarrow (pip install "pandas>=2.0" pyarrow)
  • A stable code version string for the compute logic (a git tag or release label, e.g. "2026.07")
  • The AIAG control-chart constant table (A2/D3/D4/d2) pinned at published precision, versioned alongside the code
  • A partitioned store (Parquet on object storage or a mounted volume) that supports overwrite, keyed by process_id and timestamp_window
  • A designated, verified Phase I baseline window per process that must not mutate on rerun
  • The vectorized calculation engine as the single source of limit math, so the hash covers one implementation

Step 1 — Hash the inputs that determine the limits

A rerun is reproducible only if the limits depend on nothing but the input rows, the constant table, and the code version. Fingerprint exactly those three and nothing else — no datetime.now(), no run id, no machine name. Use a content hash of the sorted measurement values so row order cannot change the result.

from __future__ import annotations

import hashlib
import pandas as pd

# AIAG constants, versioned with the code. The values are part of the hash.
A2_TABLE: dict[int, float] = {2: 1.880, 3: 1.023, 4: 0.729, 5: 0.577, 6: 0.483}


def calculation_hash(df: pd.DataFrame, subgroup_size: int, code_version: str) -> str:
    """Deterministic fingerprint of (input rows, constant table, code version).

    Excludes wall-clock time, run id, and host so identical inputs always hash
    identically across retries, backfills, and workers.
    """
    if "measurement" not in df.columns:
        raise ValueError("Frame needs a 'measurement' column to hash.")
    # Sort by a stable key so upstream ordering never changes the fingerprint.
    ordered = df.sort_values(["timestamp", "measurement"]).reset_index(drop=True)
    row_digest = pd.util.hash_pandas_object(
        ordered["measurement"], index=False
    ).values.tobytes()
    constants = f"n={subgroup_size}|A2={A2_TABLE[subgroup_size]}|v={code_version}"
    return hashlib.sha256(row_digest + constants.encode()).hexdigest()[:16]

Verify determinism immediately: hash the same frame twice and assert equality, then shuffle the rows and assert the hash is unchanged.

df = pd.DataFrame({"measurement": [10.1, 9.8, 10.4, 10.0, 9.9, 10.2],
                   "timestamp": pd.date_range("2026-07-16", periods=6, freq="min")})
h1 = calculation_hash(df, 3, "2026.07")
h2 = calculation_hash(df.sample(frac=1.0, random_state=1), 3, "2026.07")
assert h1 == h2                     # row order must not change the hash

Step 2 — Write limits to a versioned, overwrite-safe partition

The second source of nondeterminism is the write itself. An appended write turns a rerun into a duplicate; a filename that includes the wall clock turns a rerun into an orphaned artifact. Key the path by process_id, timestamp_window, and calculation_hash, and overwrite rather than append. A rerun with identical inputs targets the identical path and rewrites identical content.

from pathlib import Path

LIMITS_ROOT = Path("/data/limits")


def write_limits(limits: dict[str, float | str], process_id: str,
                 window: str, calc_hash: str) -> Path:
    """Idempotent write: same inputs -> same path -> same content."""
    out_dir = LIMITS_ROOT / process_id / f"window={window}"
    out_dir.mkdir(parents=True, exist_ok=True)
    # The hash in the filename makes a changed input a NEW artifact, and an
    # unchanged input an overwrite of the same file — never a silent append.
    out_path = out_dir / f"limits_{calc_hash}.parquet"
    pd.DataFrame([{**limits, "calculation_hash": calc_hash}]).to_parquet(
        out_path, index=False
    )
    return out_path

If a rerun produces a different hash for the same window, that is a signal, not a nuisance: the inputs, constants, or code genuinely changed, and the new file sits beside the old one so the change is auditable rather than destructive. Verify by writing twice and confirming a single file with stable content.

Partitioning by window= in the path is deliberate: it lets a query engine prune to a single window without scanning the whole process history, and it makes the retention story simple, since aging out a window is a directory delete rather than a row-level operation. Keep the hash in the filename rather than only inside the file, because a downstream reader can then confirm which calculation produced a chart from the path alone, without opening every candidate Parquet.

Step 3 — Guard the Phase I baseline against mutation

Phase I establishes the limits from a verified stable window of at least 20–25 subgroups; Phase II monitors new data against those frozen limits. A backfill that recomputes and overwrites the Phase I baseline silently moves the goalposts for every chart that referenced it. Guard the baseline so it can only be established once and never mutated by a routine rerun.

def freeze_phase_one_baseline(process_id: str, baseline_limits: dict[str, float],
                              calc_hash: str) -> Path:
    """Establish the Phase I baseline exactly once; refuse silent overwrites."""
    baseline_path = LIMITS_ROOT / process_id / "phase_i_baseline.parquet"
    if baseline_path.exists():
        existing = pd.read_parquet(baseline_path)
        if existing["calculation_hash"].iloc[0] != calc_hash:
            # A different hash means someone is trying to redefine the baseline.
            raise ValueError(
                f"Phase I baseline for {process_id} already frozen with a "
                f"different hash. Redefining it requires a change order, not a rerun."
            )
        return baseline_path            # identical hash: idempotent no-op
    pd.DataFrame([{**baseline_limits, "calculation_hash": calc_hash}]).to_parquet(
        baseline_path, index=False
    )
    return baseline_path

A rerun over the baseline window with unchanged inputs is a harmless no-op; a rerun that would change the baseline raises, forcing the change through a formal engineering change order instead of a quiet overwrite. Verify by re-freezing with the same hash (must succeed silently) and with a different hash (must raise).

Step 4 — Wire the guarded, idempotent write into the DAG task

Compose the three pieces into the compute task so the whole stage is idempotent end to end. The task reads validated rows, computes limits, hashes the determining inputs, and writes through the guarded path.

from typing import Any


def compute_and_persist_limits(process_id: str, subgroup_size: int,
                               is_phase_one: bool, **context: Any) -> dict[str, Any]:
    """Idempotent compute task: deterministic limits, guarded persistence."""
    ti = context["task_instance"]
    df = pd.read_parquet(ti.xcom_pull(task_ids="validate"))
    vals = df["measurement"].to_numpy(dtype=float)
    n = subgroup_size
    n_groups = len(vals) // n
    if n_groups == 0:
        raise ValueError("Not enough measurements for a single subgroup.")
    groups = vals[: n_groups * n].reshape(n_groups, n)
    x_bar_bar = float(groups.mean(axis=1).mean())
    r_bar = float((groups.max(axis=1) - groups.min(axis=1)).mean())
    a2 = A2_TABLE[n]
    limits = {"center": x_bar_bar, "ucl": x_bar_bar + a2 * r_bar,
              "lcl": x_bar_bar - a2 * r_bar}

    calc_hash = calculation_hash(df, n, code_version="2026.07")
    window = context["ds"]
    if is_phase_one:
        freeze_phase_one_baseline(process_id, limits, calc_hash)
    write_limits(limits, process_id, window, calc_hash)
    return {**limits, "calculation_hash": calc_hash, "window": window}

Verification

Idempotency has a crisp test: run the task twice and assert the hash and limits are identical.

df = pd.DataFrame({"measurement": [10.1, 9.8, 10.4, 10.0, 9.9, 10.2,
                                   10.3, 9.7, 10.1, 10.0, 9.9, 10.2],
                   "timestamp": pd.date_range("2026-07-16", periods=12, freq="min")})

h1 = calculation_hash(df, 3, "2026.07")
h2 = calculation_hash(df, 3, "2026.07")      # rerun, same inputs
assert h1 == h2, "reruns must yield an identical calculation hash"

# A code-version bump is the only way the hash should move for fixed data.
h3 = calculation_hash(df, 3, "2026.08")
assert h3 != h1, "a code version change must produce a new, traceable hash"
print("idempotent SPC compute verified")

Expected output: idempotent SPC compute verified. The two properties that matter: identical inputs across a retry or backfill yield the identical hash and identical limits, and the only things that move the hash are a genuine change to the rows, the constant table, or the code version — each of which is an auditable event, not an accident.

Root-Cause Table

Symptom Cause Fix
Rerun produces slightly different limits Wall-clock time or unstable row order entered the computation Hash only rows, constants, and code version; sort by a stable key first (Step 1)
Backfill duplicates a window's limits Append write instead of overwrite Key the path by process, window, and hash and overwrite the same file (Step 2)
Phase I limits drift after a routine rerun Backfill recomputed and overwrote the frozen baseline Guard the baseline so a differing hash raises and forces a change order (Step 3)
Hash changes for no obvious reason An unpinned constant table or library changed the math silently Version the constant table and code, and include both in the hash (Steps 1, 4)
Auditor cannot reproduce an archived chart The exact inputs and code version were not recorded Persist the calculation hash with the limits so the run is reproducible (Steps 2, 4)

FAQ

Why exclude the timestamp and run id from the calculation hash?

Because they change on every run while the statistical result does not. If wall-clock time or the Airflow run id entered the hash, two runs over identical data would produce different fingerprints, defeating the whole purpose — you could never assert that a rerun reproduced a prior result. The hash must depend only on the things that genuinely determine the limits: the input rows, the constant table, and the code version. Everything else is metadata you store alongside the limits, not something you fold into their identity.

What should happen when a backfill would change a Phase I baseline?

It should raise and stop, not overwrite. A Phase I baseline is established once from a verified stable window and becomes the fixed reference for Phase II monitoring. A backfill that quietly recomputes it moves the goalposts for every chart that referenced the old limits, which is exactly the silent drift that idempotency exists to prevent. Redefining a baseline is a legitimate engineering decision, but it belongs in a formal change order with a new hash and a documented rationale, not in a routine rerun.

How does the calculation hash support an audit?

The hash is a compact, reproducible proof that a given chart came from a given set of inputs and code. When an auditor points at an archived chart, you look up its calculation_hash, recover the exact rows, constants, and code version that produced it, and recompute the limits to confirm they match. That closes the reproducibility requirement of ISO 9001:2015 clause 7.5.3 and gives IATF 16949 change-order traceability a concrete anchor, since any limit change is a new hash tied to a new code or constant version.

Up one level: Orchestrating SPC Pipelines with Apache Airflow. The calculation_hash and versioned writes satisfy the reproducibility ISO 9001:2015 clause 7.5.3 requires and the change-order traceability of IATF 16949.