Scheduling Control-Chart Refresh DAGs in Airflow

Refreshing control charts on a fixed cadence sounds like a one-line cron entry until you have a dozen production lines that must each refresh independently, land their own HTML dashboard and PDF audit artifact, and never replay six months of history the first time the DAG deploys. This how-to builds that refresh DAG the right way — one cron schedule, dynamic task mapping across lines, a data-availability sensor, and export to both formats — as a focused step within orchestrating SPC pipelines with Apache Airflow.

The goal is a single DAG that scales from one line to fifty without editing the graph: add a line to a list, and mapped task instances appear automatically on the next parse. Charts refresh on schedule, exports land in object storage keyed by line and window, and catchup=False keeps a fresh deploy from stampeding the scheduler.

Why refresh per line rather than one chart for the whole plant? Each production line has its own frozen limits, its own stability history, and its own audience of operators and quality engineers. Bundling them into one job means a single line's bad batch can fail the export for every other line, and it makes the artifact path ambiguous when an auditor asks for one line's chart at one timestamp. A mapped task per line gives each its own retry budget, its own success state in the grid view, and its own uniquely keyed HTML and PDF — isolation that a monolithic refresh cannot offer.

Prerequisites

Confirm these are in place before wiring the DAG:

  • Apache Airflow 2.4+ or 3.x (dynamic task mapping requires 2.3+; schedule= requires 2.4+)
  • Python 3.10+ with pandas >= 2.0, pyarrow, and plotly >= 5.0 (pip install "pandas>=2.0" pyarrow "plotly>=5.0" kaleido)
  • kaleido installed for static PDF export from Plotly figures (pip install kaleido)
  • The upstream compute step already freezes immutable limits per line — this DAG refreshes charts, it does not calculate limits
  • A list (or Airflow Variable) of active production line ids to map over
  • Object storage or a mounted volume for the exported HTML and PDF artifacts
  • The chart type known per line, since the Plotly rendering layer draws whatever type the compute step selected

Step 1 — Define the DAG with a cron schedule and catchup off

Set the cadence with schedule= (a cron string), pin a start_date, and — this is the load-bearing line — set catchup=False. Without it, a DAG deployed today with a start_date months back will immediately queue every missed interval and hammer the scheduler and the MES at once.

from __future__ import annotations

from datetime import datetime, timedelta
from typing import Any

from airflow import DAG
from airflow.decorators import task

# Active production lines. Swap for Variable.get("spc_lines", deserialize_json=True)
# so operators can add a line without editing the DAG file.
PRODUCTION_LINES: list[str] = ["LINE_A", "LINE_B", "LINE_C"]

default_args: dict[str, Any] = {
    "owner": "spc_engineering",
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
}

dag = DAG(
    dag_id="spc_chart_refresh",
    default_args=default_args,
    schedule="0 */2 * * *",     # refresh every 2 hours; Airflow 2.4+/3.x
    start_date=datetime(2026, 1, 1),
    catchup=False,              # never replay history on deploy
    max_active_runs=1,          # one refresh window at a time
    tags=["spc", "refresh"],
)

Verify the DAG parses and shows the intended next run: airflow dags list | grep spc_chart_refresh and airflow dags next-execution spc_chart_refresh should report the upcoming interval, not a backlog.

Step 2 — Wait for the window's data with a sensor

A refresh that runs before the window's measurements have landed charts an empty frame. Gate the mapped work behind a PythonSensor in reschedule mode so it releases the worker slot between pokes instead of holding it for the whole wait.

from airflow.sensors.python import PythonSensor
from pathlib import Path

STAGING_ROOT = Path("/data/staging")


def _all_lines_landed(**context: Any) -> bool:
    """Poke True once every active line's window partition exists."""
    window = context["data_interval_start"]
    return all(
        (STAGING_ROOT / line / f"{window:%Y%m%dT%H%M}.parquet").exists()
        for line in PRODUCTION_LINES
    )


with dag:
    wait_for_data = PythonSensor(
        task_id="wait_for_data",
        python_callable=_all_lines_landed,
        poke_interval=120,      # check every 2 minutes
        timeout=60 * 20,        # give up after 20 minutes rather than hang
        mode="reschedule",      # free the slot between pokes
    )

Verify by deleting one line's staged partition and confirming the sensor stays in up_for_reschedule rather than passing — the refresh must not proceed on partial data.

Step 3 — Map the refresh task dynamically over every line

Rather than a hand-written task per line, use dynamic task mapping (.expand()) so Airflow creates one mapped instance per line at run time. Add a line to PRODUCTION_LINES and the graph grows on the next parse with no code change.

import pandas as pd

LIMITS_ROOT = Path("/data/limits")


@task
def refresh_line_chart(line_id: str, **context: Any) -> dict[str, str]:
    """Build a refreshed Plotly figure for one line from its frozen limits.

    Consumes immutable limits computed upstream; it never recalculates them.
    Returns the export paths so a downstream task can index the artifacts.
    """
    window = context["data_interval_start"]
    staged = STAGING_ROOT / line_id / f"{window:%Y%m%dT%H%M}.parquet"
    df = pd.read_parquet(staged)
    limits = pd.read_parquet(sorted((LIMITS_ROOT / line_id).glob("*.parquet"))[-1])
    if df.empty:
        raise ValueError(f"No data to refresh for {line_id} at {window}.")

    # Hand immutable limits to the renderer; see the Plotly rendering guide.
    import plotly.graph_objects as go
    fig = go.Figure(go.Scatter(x=df["timestamp"], y=df["measurement"],
                               mode="lines+markers", name="measurement"))
    fig.add_hline(y=float(limits["ucl"].iloc[0]), line_dash="dash", annotation_text="UCL")
    fig.add_hline(y=float(limits["lcl"].iloc[0]), line_dash="dash", annotation_text="LCL")
    fig.add_hline(y=float(limits["center"].iloc[0]), annotation_text="CL")
    fig.update_layout(title=f"{line_id} control chart — {window:%Y-%m-%d %H:%M}")
    return {"line_id": line_id, "figure_ready": "true"}


with dag:
    # One mapped instance per line, created automatically at run time.
    refreshed = refresh_line_chart.expand(line_id=PRODUCTION_LINES)
    wait_for_data >> refreshed

Verify in the grid view that expanding refresh_line_chart shows one mapped index per line. Adding "LINE_D" to the list and re-parsing should produce a fourth mapped instance with no edit to the dependency wiring.

Step 4 — Export HTML dashboards and PDF audit artifacts

Each refreshed figure serves two consumers: an interactive HTML dashboard for the floor and a frozen PDF for the audit packet. Export both, keyed by line and window so artifacts never collide and an auditor can find the exact chart.

import plotly.io as pio

EXPORT_ROOT = Path("/data/exports")


def export_artifacts(fig, line_id: str, window: datetime) -> dict[str, str]:
    """Write an interactive HTML dashboard and a static PDF for one line."""
    out_dir = EXPORT_ROOT / line_id
    out_dir.mkdir(parents=True, exist_ok=True)
    stamp = f"{window:%Y%m%dT%H%M}"

    html_path = out_dir / f"{stamp}.html"
    # include_plotlyjs='cdn' keeps the HTML small for web embeds.
    pio.write_html(fig, file=str(html_path), include_plotlyjs="cdn")

    pdf_path = out_dir / f"{stamp}.pdf"     # requires kaleido; byte-stable artifact
    pio.write_image(fig, file=str(pdf_path), format="pdf", scale=2)

    return {"html": str(html_path), "pdf": str(pdf_path)}

The PDF is the compliance artifact: it is byte-stable regardless of browser and belongs in the IATF 16949 evidence packet, while the HTML is the operator's interactive view. Fold this call into refresh_line_chart so each mapped instance produces both files for its line.

Verification

Run the DAG for one interval and confirm the mapped instances and their exports:

# Offline sanity check of the export contract — no live Airflow required.
from pathlib import Path
import plotly.graph_objects as go
import plotly.io as pio

fig = go.Figure(go.Scatter(x=[1, 2, 3], y=[10.1, 10.4, 9.8]))
paths = export_artifacts(fig, "LINE_A", datetime(2026, 7, 16, 8, 0))

assert Path(paths["html"]).exists() and Path(paths["html"]).stat().st_size > 0
assert Path(paths["pdf"]).exists() and Path(paths["pdf"]).stat().st_size > 0
print("refresh export contract holds")

Expected output: refresh export contract holds, with one HTML and one PDF per line under /data/exports/<line>/. In the Airflow UI, a successful run shows wait_for_data green and refresh_line_chart expanded to exactly len(PRODUCTION_LINES) mapped instances, each succeeded. Trigger airflow dags test spc_chart_refresh 2026-07-16 to run the whole graph end to end without touching the live scheduler.

One mapped instance failing does not fail the others: Airflow marks that index failed and retries it on its own schedule while the rest succeed. That is the practical payoff of mapping — a bad batch on one line quarantines to one line. When you tune the schedule, match the cron cadence to how fast the process actually moves: a slow chemical bath charted every two hours is sensible, but a high-rate stamping press may warrant a tighter interval, and anything approaching per-reading latency belongs in a streaming path rather than a scheduled refresh.

Root-Cause Table

Symptom Cause Fix
Scheduler queues hundreds of runs on first deploy catchup=True (the default) with a back-dated start_date Set catchup=False unless a historical backfill is intended (Step 1)
Refresh charts an empty or partial frame Task ran before all lines' data landed Gate the mapped work behind the reschedule-mode data sensor (Step 2)
Adding a line requires editing the DAG graph Hard-coded one task per line Map the task with .expand() over the line list or an Airflow Variable (Step 3)
PDF export raises ValueError: kaleido Static image backend not installed pip install kaleido; it is required for Plotly write_image (Step 4)
Exports from two windows overwrite each other Artifact path not keyed by window Key HTML/PDF filenames by line id and timestamp window (Step 4)
Refresh recomputes limits and two views disagree Task recalculated limits instead of reading them Read frozen limits from the compute step; the refresh only draws them (Step 3)

FAQ

Why is catchup=False so important for a refresh DAG?

Airflow defaults catchup to True, which means a DAG deployed today with a start_date set months back will immediately schedule every interval it "missed" between then and now. For an hourly or two-hourly refresh that is hundreds of runs queued at once, each hammering the MES and the export storage. A control-chart refresh only cares about the current window, so catchup=False is almost always correct; reserve a deliberate backfill for the idempotent path where reruns are proven deterministic.

Should I use dynamic task mapping or a task per line?

Use dynamic task mapping once you have more than two or three lines. A hand-written task per line means every new line is a code change and a review, and the DAG file grows without bound. Mapping with .expand() over a line list — ideally an Airflow Variable so operators can edit it without a deploy — creates one mapped instance per line at run time, isolates failures per index, and keeps the graph definition constant regardless of how many lines you run.

Why export both HTML and PDF instead of just one?

They serve different consumers. The interactive HTML dashboard lets operators hover for timestamps and drill into subgroups during a live shift, so it belongs on the floor. The PDF is byte-stable regardless of browser and is the artifact an auditor archives for the IATF 16949 evidence packet. Rendering both from the same figure guarantees the archived chart matches what the operator saw, which the interactive view alone cannot promise.

Up one level: Orchestrating SPC Pipelines with Apache Airflow.