Scheduling InfluxDB downsampling with Airflow DAGs
You run OEE reporting for a stamping press-shop: forty presses each stream parts_good, parts_scrap, and run_seconds counters into a press_raw bucket several times per second, and the plant dashboard reads 15-minute Overall Equipment Effectiveness figures from a press_oee_15m bucket. A native Flux task could do the rollup, but the shop already owns the availability calculation, the shift-boundary logic, and a MES export inside Airflow — so the downsampling belongs in the same orchestrator as everything else it depends on. The trap is that a naive Airflow DAG that fires “every fifteen minutes” and queries “the last fifteen minutes” is not reproducible: a retry, a paused-then-resumed scheduler, or a manual backfill each read a different wall-clock window and produce different aggregates. This page builds a reproducible TaskFlow DAG whose window is derived from the run’s data interval, gated on raw coverage, and written idempotently so a backfill and a live run for the same quarter-hour always land the identical OEE point. It is the hands-on companion to Airflow + InfluxDB integration.
Prerequisites
Solution walkthrough
Step 1 — Derive the window from the run’s data interval
The single decision that makes an Airflow downsampling DAG correct is refusing to look at the wall clock. Every run carries a data_interval_start and data_interval_end — the logical window the run is responsible for — and those bounds are stable across retries and identical when the same logical date is backfilled months later. Set schedule to the same 15-minute cadence as the aggregate so one run maps to exactly one window, enable catchup=True so historical windows are filled deterministically, and cap max_active_runs so concurrent backfill runs cannot race on the same target series.
from __future__ import annotations
import pendulum
from airflow.decorators import dag, task
from airflow.exceptions import AirflowSkipException
from airflow.hooks.base import BaseHook
from airflow.models import Variable
from influxdb_client import InfluxDBClient, Point, WritePrecision
from influxdb_client.client.write_api import SYNCHRONOUS
INFLUX_CONN = "influxdb_pressshop"
SRC_BUCKET = "press_raw"
DST_BUCKET = "press_oee_15m"
ORG = "manufacturing"
WINDOW_SECONDS = 900 # planned time per 15-minute window
def _client() -> InfluxDBClient:
conn = BaseHook.get_connection(INFLUX_CONN)
return InfluxDBClient(url=conn.host, token=conn.password, org=ORG, timeout=30_000)
@dag(
dag_id="pressshop_oee_downsample_15m",
schedule="*/15 * * * *",
start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
catchup=True,
max_active_runs=1,
default_args={"retries": 3, "retry_delay": pendulum.duration(minutes=2)},
tags=["influxdb", "downsampling", "oee"],
)
def pressshop_oee():
@task
def resolve_window(data_interval_start=None, data_interval_end=None) -> dict:
# Airflow injects the run's logical interval; never use datetime.now().
return {
"start": data_interval_start.in_timezone("UTC").isoformat(),
"stop": data_interval_end.in_timezone("UTC").isoformat(),
}
schedule="*/15 * * * *" makes data_interval_end - data_interval_start exactly the 900-second window the rollup emits, so the window boundaries in InfluxDB and the DAG boundaries never drift apart. Aligning the two is the same discipline that native scheduling relies on when choosing between InfluxDB tasks and Airflow for time-series pipelines — the orchestrator changes, the boundary contract does not.
Step 2 — Gate the window on raw coverage before aggregating
Downsampling an under-reported window is worse than skipping it: it writes a confident-looking OEE number built on half the presses. Before transforming, count how many machines actually reported in the window and short-circuit if coverage is below the shift’s expected fleet size. Raising AirflowSkipException marks the run skipped (not failed) so it does not fire the retry storm a real error would, and leaves a visible gap you can backfill once late data lands.
@task
def quality_gate(window: dict, min_machines: int = 30) -> dict:
flux = f'''
from(bucket: "{SRC_BUCKET}")
|> range(start: {window["start"]}, stop: {window["stop"]})
|> filter(fn: (r) => r._measurement == "stamping_press")
|> filter(fn: (r) => r._field == "parts_good")
|> group(columns: ["machine_id"])
|> count()
|> group()
|> count()
'''
with _client() as client:
tables = client.query_api().query(flux, org=ORG)
reporting = int(tables[0].records[0].get_value()) if tables else 0
if reporting < min_machines:
raise AirflowSkipException(
f"only {reporting}/{min_machines} presses reported "
f"in {window['start']} — skipping rollup"
)
return window
The inner count() reduces each machine’s points to a per-machine row; the outer group() + count() collapses that to the number of distinct reporting machines. Returning the same window dict lets the downstream task depend on the gate through the TaskFlow data edge, so the aggregate step simply never runs on a starved window.
Step 3 — Downsample and write the OEE point idempotently
Now extract the counters, compute OEE, and load one point per machine stamped at the window start. OEE is the product of three ratios — availability, performance, and quality:
$$ \text{OEE} = A \times P \times Q = \frac{t_{run}}{t_{plan}} \times \frac{c_{ideal}, n_{total}}{t_{run}} \times \frac{n_{good}}{n_{total}} $$
Writing each point at data_interval_start is what makes re-runs safe: InfluxDB overwrites a point that shares the same measurement, tag set, and timestamp, so a backfill of a window that already ran replaces its own row rather than double-counting it.
@task
def downsample(window: dict) -> int:
ideal_cycle = Variable.get(
"press_ideal_cycle_s", deserialize_json=True
) # {"machine_id": seconds_per_part}
flux = f'''
from(bucket: "{SRC_BUCKET}")
|> range(start: {window["start"]}, stop: {window["stop"]})
|> filter(fn: (r) => r._measurement == "stamping_press")
|> filter(fn: (r) =>
r._field == "parts_good" or r._field == "parts_scrap"
or r._field == "run_seconds")
|> group(columns: ["machine_id", "_field"])
|> sum()
|> pivot(rowKey: ["machine_id"], columnKey: ["_field"], valueColumn: "_value")
'''
ts = pendulum.parse(window["start"])
points = []
with _client() as client:
tables = client.query_api().query(flux, org=ORG)
for rec in (r for t in tables for r in t.records):
mid = rec["machine_id"]
good = float(rec["parts_good"] or 0)
scrap = float(rec["parts_scrap"] or 0)
run_s = float(rec["run_seconds"] or 0)
total = good + scrap
if total == 0 or run_s == 0:
continue
a = run_s / WINDOW_SECONDS
p = (ideal_cycle.get(mid, 1.0) * total) / run_s
q = good / total
points.append(
Point("oee_15m")
.tag("machine_id", mid)
.field("availability", round(a, 4))
.field("performance", round(min(p, 1.0), 4))
.field("quality", round(q, 4))
.field("oee", round(a * min(p, 1.0) * q, 4))
.field("parts_good", int(good))
.time(ts, WritePrecision.S)
)
client.write_api(write_options=SYNCHRONOUS).write(
bucket=DST_BUCKET, org=ORG, record=points
)
return len(points)
The pivot turns three summed fields into one row per machine so OEE is a single arithmetic pass. Clamping performance at 1.0 absorbs the counter noise that otherwise pushes performance above 100% when a press briefly beats its ideal cycle. For the connection reuse, batching, and backoff that a production writer needs — this example uses a synchronous write for clarity — follow the Python client orchestration patterns, and treat this task as one stage of the broader downsampling aggregation pipeline.
Step 4 — Verify the write and wire the dependencies
A rollup that silently writes zero points is a failure that looks like success. Read the target bucket back for the same window and assert the machine count matches what the transform emitted, then declare the task graph so Airflow enforces gate → downsample → verify in order.
@task
def verify(window: dict, expected: int) -> None:
flux = f'''
from(bucket: "{DST_BUCKET}")
|> range(start: {window["start"]}, stop: {window["stop"]})
|> filter(fn: (r) => r._measurement == "oee_15m" and r._field == "oee")
|> group() |> count()
'''
with _client() as client:
tables = client.query_api().query(flux, org=ORG)
written = int(tables[0].records[0].get_value()) if tables else 0
if written != expected:
raise ValueError(f"verify mismatch: wrote {expected}, read {written}")
w = resolve_window()
gated = quality_gate(w)
n = downsample(gated)
verify(w, n)
pressshop_oee()
Because verify reads the same data_interval window it wrote, the check is exact rather than approximate: any off-by-one in the window math surfaces here as a mismatch, on the run that caused it, instead of as a wrong dashboard number a week later.
Gotchas and edge cases
Wall-clock windows destroy backfill reproducibility. If any task calls datetime.now() or queries range(start: -15m), a retry two minutes after the original attempt reads a shifted window and the aggregate changes. Derive every bound from data_interval_start/data_interval_end so the run is a pure function of its logical date — this is the precondition for catchup and manual backfills producing the same numbers as the live run.
Catchup without idempotent writes double-counts history. Turning on catchup=True after the DAG has been paused replays every missed window. That is only safe because Step 3 stamps points at the window start and InfluxDB overwrites same-timestamp series; if you instead appended with server-assigned timestamps, each backfill would add duplicate rows and inflate the totals. Keep the timestamp deterministic and re-runs stay clean.
A short source retention races the scheduler. If press_raw expires at 24 hours but a backfill reaches for a window three days old, the raw data is already gone and quality_gate skips it forever. Size the source retention to exceed your maximum backfill horizon plus the worst-case scheduler lag, and never let the Airflow interval outlive the data it reads.
Verification
Confirm the DAG is producing continuous, complete windows — one OEE series per machine, no gaps, over the last few hours:
from(bucket: "press_oee_15m")
|> range(start: -3h)
|> filter(fn: (r) => r._measurement == "oee_15m" and r._field == "oee")
|> group(columns: ["_time"])
|> count()
|> group()
|> yield(name: "machines_per_window")
Each 15-minute window should report the same machine count you gated on (30+); a window that dips low pinpoints a starved interval the quality gate skipped, and an empty stretch means the scheduler missed runs Airflow should now backfill.
FAQ
Should the schedule interval always equal the downsample window?
Keeping them equal is the simplest correct default: one run owns exactly one aggregate window, so retries and backfills map cleanly. You can run a coarser schedule that loops over several windows per run, but then you own the loop’s idempotency and partial-failure handling manually — rarely worth it below hourly cadences.
How do I keep the InfluxDB token out of the DAG code?
Store it in the Airflow Connection’s password field and read it through BaseHook.get_connection, or back the connection with a secrets manager via an Airflow secrets backend. The DAG file should reference only the connection id, never the token value, so rotation happens outside the code.
What happens to OEE points when late press data arrives after the window ran?
The window was already written from whatever was present, and the raw late points sit in press_raw. Because the write is idempotent, the cleanest fix is a targeted Airflow backfill of the affected logical dates once the data settles — the re-run overwrites the same OEE series with the now-complete numbers.
Related
- Airflow + InfluxDB Integration — the connection, secrets, and operator patterns that this DAG builds on.
- InfluxDB Tasks vs Airflow for Time-Series Pipelines — when to keep a rollup native versus lifting it into Airflow.
- Python Client Orchestration Patterns — reusable-client, batching, and retry practice for the write path used here.