Building dependency graphs for multi-stage pipeline execution

IoT telemetry ingestion rarely follows a linear path. Raw sensor payloads traverse validation, schema normalization, windowed aggregation, downsampling, and cold-storage archival before reaching analytical endpoints or a machine-learning feature store, and each stage depends on the deterministic completion of the one before it for the same time window. In-database checkpoint gating, covered under dependency mapping and DAG construction, handles a two- or three-node chain cleanly. But once a pipeline fans out into parallel branches, needs conditional execution, or must coordinate InfluxDB with external systems, you want a single control plane that holds the whole topology, tracks the state of every node per window, and drives the InfluxDB API directly. This walkthrough builds exactly that: a small Python orchestrator that models each processing stage as a node, resolves execution order topologically, and executes one temporal slice at a time with idempotent, isolated retries.

Prerequisites

InfluxDB 2.7+ (OSS or Cloud) with the native task engine enabled and an org you can write tasks to
Python 3.9+ (graphlib.TopologicalSorter ships in the standard library from 3.9; networkx is optional and only used for predecessor lookups)
influxdb-client 1.36+ installed (pip install influxdb-client networkx)
A token with read/write scope on your telemetry buckets
Source and target buckets provisioned: raw_telemetry, validated_telemetry, telemetry_hourly, and a pipeline_checkpoints metadata bucket
Window boundaries expressed in UTC as RFC 3339 timestamps to avoid drift across the control plane and the database

Why an external control plane

InfluxDB’s native task engine schedules each task independently and has no built-in concept of one task depending on another. The gating pattern in the parent guide expresses a dependency by having a downstream Flux task query a checkpoint bucket before it runs — perfect for predictable chains. It gets awkward the moment you need to fan a validated stream out into a downsample branch and an anomaly-scoring branch that run concurrently, then re-join them at a rollup, or when a node has to call something outside InfluxDB. A Python control plane owns the topology as data, computes a safe execution order once, and tracks the state of every (node, window) pair so retries stay surgical. The database keeps doing what it is good at — high-throughput ingestion and Flux computation — while the orchestrator owns ordering, state, and failure routing. This is the same separation that Python client orchestration patterns applies to bursty batch work.

Solution walkthrough

Step 1 — Model each stage as a per-window node

The unit of work is not “a task” but a (node, window) pair. A node is a named Flux transformation; a window is a half-open slice [window_start, window_end). Binding state to that pair — rather than to the node alone — is what lets the 15:00 rollup be SUCCESS while the 16:00 rollup is still PENDING, and it is what keeps a retry scoped to one slice instead of the whole stage. Store each node’s Flux with ${window_start} / ${window_end} placeholders so a single definition serves every window:

python

VALIDATE_FLUX = """
from(bucket: "raw_telemetry")
    |> range(start: ${window_start}, stop: ${window_end})
    |> filter(fn: (r) => r._measurement == "sensor.readings")
    |> filter(fn: (r) => r._value > -50.0 and r._value < 150.0)
    |> to(bucket: "validated_telemetry")
"""

ROLLUP_FLUX = """
from(bucket: "validated_telemetry")
    |> range(start: ${window_start}, stop: ${window_end})
    |> aggregateWindow(every: 1h, fn: mean, createEmpty: false)
    |> to(bucket: "telemetry_hourly")
"""

Every node’s Flux must be idempotent so a retry never doubles a point: explicit range() bindings, deterministic group() keys, and aggregation semantics that collapse duplicates. Because writes are keyed by measurement, tags, field, and timestamp, re-running the identical window overwrites rather than appends. The correctness rules behind this are the subject of writing robust Flux scripts for automated data rollups; the createEmpty: false choice above matters specifically for sparse IoT sensors that skip windows.

Step 2 — Build the graph and validate its topology

Declare the pipeline as node-to-dependencies edges, then validate acyclicity and compute a dependency-respecting order before anything executes. A back-edge added by a well-meaning “correction” task is the classic way a graph turns cyclic and retries loop forever; catching it here means it fails at deploy time, not at 3 a.m.

python

import networkx as nx
from graphlib import TopologicalSorter, CycleError

def build_graph() -> nx.DiGraph:
    g = nx.DiGraph()
    g.add_node("validate", flux=VALIDATE_FLUX)
    g.add_node("downsample_1m", flux=DOWNSAMPLE_FLUX)
    g.add_node("anomaly_score", flux=ANOMALY_FLUX)
    g.add_node("rollup", flux=ROLLUP_FLUX)
    g.add_node("archive", flux=ARCHIVE_FLUX)
    g.add_edge("validate", "downsample_1m")
    g.add_edge("validate", "anomaly_score")   # fans out — runs concurrently
    g.add_edge("downsample_1m", "rollup")
    g.add_edge("rollup", "archive")
    return g

def safe_order(g: nx.DiGraph) -> list[str]:
    deps = {n: list(g.predecessors(n)) for n in g.nodes}
    ts = TopologicalSorter(deps)
    try:
        ts.prepare()                    # raises CycleError on a back-edge
        return list(ts.static_order())  # dependencies always precede dependents
    except CycleError as e:
        raise RuntimeError(f"DAG validation failed — cycle: {e.args[1]}")

downsample_1m and anomaly_score share validate as an upstream boundary but do not depend on each other, so the orchestrator is free to run them concurrently. Expressing the pipeline as a graph is what lets you parallelize the safe branches while serializing the unsafe ones — a decision that cascades into the downsampling and aggregation pipeline design those nodes implement.

Step 3 — Execute one window with per-node state tracking

With a validated order in hand, execute a single window. For each node, confirm every predecessor reached SUCCESS for the same slice before running it; otherwise mark it PENDING and move on. State lives in a dict keyed by (node, window_start), which you should persist to the pipeline_checkpoints bucket so the control plane can resume after a restart:

python

import asyncio
from datetime import datetime
from influxdb_client import InfluxDBClient

class TelemetryDAG:
    def __init__(self, org: str, client: InfluxDBClient, graph: nx.DiGraph):
        self.org = org
        self.query_api = client.query_api()
        self.graph = graph
        self.order = safe_order(graph)
        self.state: dict[tuple[str, datetime], str] = {}

    async def execute_window(self, window_start: datetime, window_end: datetime):
        for node_id in self.order:
            deps = list(self.graph.predecessors(node_id))
            if not all(self.state.get((d, window_start)) == "SUCCESS" for d in deps):
                self.state[(node_id, window_start)] = "PENDING"
                continue

            self.state[(node_id, window_start)] = "RUNNING"
            try:
                flux = self.graph.nodes[node_id]["flux"]
                bound = (flux
                         .replace("${window_start}", window_start.isoformat())
                         .replace("${window_end}", window_end.isoformat()))
                await self._run_flux(bound)
                self.state[(node_id, window_start)] = "SUCCESS"
            except Exception as e:
                self.state[(node_id, window_start)] = "FAILED"
                raise RuntimeError(f"Node {node_id} failed for {window_start}: {e}")

    async def _run_flux(self, flux_script: str):
        # Production wrapper: run via query_api, validate the response, and
        # enforce a timeout so a hung run cannot wedge the whole window.
        loop = asyncio.get_running_loop()
        await loop.run_in_executor(None, self.query_api.query, flux_script, self.org)

Injecting window_start / window_end as ISO 8601 strings keeps each run bound to its slice and prevents late-arriving telemetry from contaminating a neighbouring window. Cadence and the offset that decides when a window is safe to launch are governed by cron and interval scheduling logic — the control plane decides ordering, but the schedule still decides timing.

Step 4 — Isolate partial failures and retry the slice

Partial failures — a transient rate limit, a network partition, one malformed payload — are inevitable in distributed telemetry. The graph turns them from a pipeline-wide break into a localized event: when a node hits FAILED, downstream propagation halts immediately for that window while other windows keep flowing. Retry only the failed (node, window) pair, with exponential backoff, and because the node is idempotent and range-bound the retry reprocesses the exact slice without duplicating data:

python

async def run_node_with_retry(dag, node_id, w_start, w_end, max_attempts=5):
    for attempt in range(max_attempts):
        try:
            flux = dag.graph.nodes[node_id]["flux"] \
                .replace("${window_start}", w_start.isoformat()) \
                .replace("${window_end}", w_end.isoformat())
            await dag._run_flux(flux)
            dag.state[(node_id, w_start)] = "SUCCESS"
            return
        except Exception:
            dag.state[(node_id, w_start)] = "FAILED"
            await asyncio.sleep(min(2 ** attempt, 60))  # capped exponential backoff
    raise RuntimeError(f"{node_id} exhausted retries for {w_start}")

For long windows, checkpoint intermediate aggregation state to a scratch bucket so a retry resumes from the last verified point rather than restarting from raw ingestion.

Gotchas and edge cases

State keyed to the node, not the window. If you track state per node instead of per (node, window), a single SUCCESS marks the stage “done” for every slice, so a retry silently skips windows that never actually ran. Always key state — and the checkpoint you persist — by the window boundary.
Concurrency left unbounded on a write node. Running two overlapping executions of a write-producing node against the same window duplicates points. Serialize write nodes (one in-flight run per node) even while you parallelize independent branches across nodes; only read-only branches like anomaly scoring should fan out freely.
createEmpty: true on sparse sensors. For IoT fleets whose devices skip reporting intervals, leaving aggregateWindow(createEmpty: true) emits null rows that inflate downstream cardinality and skew means. Use createEmpty: false unless a consumer genuinely needs a row per interval.

Verification snippet

Confirm each window produced exactly one materialization per node — a 0 means a node never fired for that slice, a value above 1 means it ran more than once:

flux

from(bucket: "pipeline_checkpoints")
    |> range(start: -6h)
    |> filter(fn: (r) => r._measurement == "task_status" and r._field == "completed")
    |> group(columns: ["task_name"])
    |> aggregateWindow(every: 1h, fn: sum, createEmpty: true)
    |> filter(fn: (r) => r._value != 1)   // any row returned is a broken window

Pair this with a deadman check so a silently stalled control plane pages on-call instead of decaying quietly, and export node latency (time from RUNNING to SUCCESS) and queue depth (pending windows awaiting execution) to your monitoring stack. The InfluxData reference on processing data with tasks documents the underlying task-execution and resource model.

Dependency mapping and DAG construction — the in-database checkpoint-gating approach this control plane builds on.
Python client orchestration patterns — wrap these nodes in a runner that owns retries and cross-system I/O.
Writing robust Flux scripts for automated data rollups — the idempotent Flux each node depends on.

Up: Dependency Mapping & DAG Construction — the parent guide covering time-series DAG semantics, checkpoint gating, and graph validation.

# Building dependency graphs for multi-stage pipeline execution

# Prerequisites

# Why an external control plane

# Solution walkthrough

# Step 1 — Model each stage as a per-window node

# Step 2 — Build the graph and validate its topology

# Step 3 — Execute one window with per-node state tracking

# Step 4 — Isolate partial failures and retry the slice

# Gotchas and edge cases

# Verification snippet

# Related

Related pages

Building dependency graphs for multi-stage pipeline execution

Prerequisites

Why an external control plane

Solution walkthrough

Step 1 — Model each stage as a per-window node

Step 2 — Build the graph and validate its topology

Step 3 — Execute one window with per-node state tracking

Step 4 — Isolate partial failures and retry the slice

Gotchas and edge cases

Verification snippet

Related