# Dependency Mapping & DAG Construction

In high-throughput IoT environments, telemetry pipelines rarely execute as isolated, linear scripts. A single logical pipeline typically spans ingestion, schema validation, downsampling, anomaly scoring, and cold-storage archival — and each of those stages depends on the deterministic completion of the one before it. The moment two or more of these stages are wired together on independent timers, the platform develops a silent class of failures: an aggregation task fires against a window that validation has not finished writing, a rollup overwrites a downstream metric that a late run had already corrected, or a blind retry loop reprocesses a window that was never actually ready. Dependency mapping and DAG construction is the discipline that removes this ambiguity. By modeling pipeline stages as discrete nodes and their execution prerequisites as directed edges, you enforce strict temporal alignment, eliminate race conditions, and maintain verifiable data lineage from edge device to analytical dashboard. Within the InfluxDB ecosystem, this paradigm directly dictates how tasks are chained and monitored, and it is one of the core specializations of Automated Task Scheduling & Orchestration for production time-series workloads.

# The Failure This Solves: Overlapping, Out-of-Order Windows

Consider a common IoT topology. A validation task runs every: 5m to quarantine impossible sensor readings, and an hourly rollup task runs cron: "0 * * * *" to average the validated stream into a reporting bucket. Nothing connects them except the clock. Most of the time it works. Then one afternoon a fleet firmware update floods the ingest path, the 14:00–15:00 validation runs take nine minutes each instead of forty seconds, and the 15:00 rollup fires while the final validation window is still in flight. The rollup reads a half-written hour, computes a mean over 80% of the points, and writes a finalized aggregate that is quietly wrong. No task reports an error. No alert fires. The dashboard shows a plausible number that nobody can reconcile three weeks later during an incident review.

This is the failure dependency mapping exists to prevent. The two tasks are not independent — the rollup has a hard prerequisite on validation completing for the same time window — but nothing in the schedule expresses that. A directed acyclic graph makes the prerequisite explicit and machine-checkable: the rollup node has an inbound edge from the validation node, and it does not run until that edge reports success for the identical temporal slice. The rest of this guide shows how to build, encode, and verify that graph against the InfluxDB task engine.

# Prerequisites

Before implementing the patterns below, confirm your environment:

• InfluxDB 2.7+ (OSS or Cloud) with the native task engine enabled and an org you can create tasks in
• A token with read/write scope on your data buckets plus write access to a checkpoint bucket
• Python 3.9+ for graph validation — graphlib.TopologicalSorter is in the standard library from 3.9 onward; no third-party install is required for the core validator
• Source and target buckets provisioned: raw_telemetry, validated_telemetry, and a dedicated pipeline_checkpoints metadata bucket
• Familiarity with the option task block and window anchoring covered in Flux scripting for task automation

# Core Concepts: Time-Series DAG Semantics

A Directed Acyclic Graph (DAG) applied to time-series data must satisfy three non-negotiable constraints to prevent data corruption and execution deadlocks:

1. Temporal Monotonicity. Downstream transformations must process time windows that are strictly equal to or later than upstream windows. Late-arriving telemetry or historical backfills require explicit compensation logic rather than implicit graph traversal, ensuring that rollups never overwrite finalized aggregates with partial ones.
2. Acyclicity. Circular dependencies between tasks create unbounded retry loops and resource exhaustion. Every directed edge must advance the execution sequence forward, guaranteeing that the pipeline eventually reaches a terminal state. Acyclicity is what makes a valid execution order computable at all.
3. Idempotent Execution. Each node must yield deterministic results when re-executed over an identical time range. Idempotency enables safe automated retries, manual backfills, and parallel scaling without introducing duplicate points or metric drift — the same guarantee developed in writing robust Flux scripts for automated data rollups.

A useful way to reason about a single pipeline run is that every node is bound to one temporal slice [window_start, window_end), and an edge is not a data pipe but a readiness signal: it carries the completion state of the upstream node for that exact slice, not the telemetry itself. The raw data always travels through buckets; the graph only carries permission to proceed.

When architecting IoT telemetry flows, these principles dictate workload partitioning. Raw sensor payloads feed a validation node, which then fans out into parallel branches — a 1-minute downsample and an anomaly-scoring pass — that share a synchronized upstream boundary but execute independently. Downsampling and scoring do not depend on each other, so they can run concurrently; but both depend on validation, and the hourly rollup depends on the downsample. Expressing this as a graph is what lets the orchestrator parallelize the safe branches while serializing the unsafe ones.

# Step-by-Step Implementation

The most robust pattern for InfluxDB pairs an in-database checkpoint layer with an external graph validator. The checkpoint bucket records which nodes have completed for which window; the validator guarantees the graph you deploy is acyclic and computes a safe execution order. The following steps build both.

# Step 1 — Choose a dependency-encoding strategy

InfluxDB’s native task engine executes Flux scripts on cron or interval triggers, but it does not expose a visual DAG builder, so dependencies must be encoded explicitly. Two patterns dominate:

• Implicit window chaining. Task B runs on a fixed interval with a calculated offset. It queries data that Task A materialized in the preceding window, and the offset acts as a temporal buffer guaranteeing A completes before B reads. This is simple and requires no extra state, but it is fragile under variable processing latency — the failure described at the top of this page is an implicit-chaining failure. Tuning that offset correctly is the subject of cron & interval scheduling logic.
• Explicit state tracking. A lightweight metadata bucket stores execution checkpoints. Downstream tasks query this bucket to verify that upstream prerequisites have successfully completed for the target time range before proceeding. This decouples scheduling from execution guarantees and is the correct choice for bursty IoT workloads.

Use implicit chaining only for predictable, low-latency stages; use explicit tracking wherever a stage can overrun its window.

# Step 2 — Create the checkpoint bucket

Provision a dedicated metadata bucket so checkpoint writes never compete with telemetry retention:

influx bucket create --name pipeline_checkpoints --org your-org --retention 30d

The short retention keeps the bucket cheap — you only need enough history to cover your longest backfill window plus an audit margin.

# Step 3 — Emit a checkpoint from the upstream node

Configure the upstream validation task to write a completion record keyed to the window it just processed. The critical parameter is that the checkpoint timestamp equals the window’s stop boundary, not now() — this is what lets a downstream task query “has the 15:00 window completed?” rather than “did anything run recently?”

import "array"

// task: validate_sensor_data
option task = {name: "validate_sensor_data", every: 5m, offset: 15s}

data = from(bucket: "raw_telemetry")
    |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
    |> filter(fn: (r) => r._measurement == "sensor.readings")
    |> filter(fn: (r) => r._field == "temperature")
    |> filter(fn: (r) => r._value > -50.0 and r._value < 150.0)

data |> to(bucket: "validated_telemetry")

// Write a checkpoint stamped at the window boundary, not wall-clock now().
array.from(rows: [{
        _time: v.timeRangeStop,          // <-- keyed to the slice, enables per-window lookups
        _measurement: "task_status",
        task_name: "validate_sensor_data",
        _field: "completed",
        _value: 1,
    }])
    |> to(bucket: "pipeline_checkpoints")

# Step 4 — Gate the downstream node on the checkpoint

The downstream rollup queries the checkpoint bucket for its target window before transforming anything. If the expected record is absent, the task exits cleanly and relies on the scheduler to retry on the next tick — it never writes a partial aggregate.

// task: hourly_rollup
option task = {name: "hourly_rollup", every: 1h, offset: 2m}

// 1. Look for the upstream completion marker for THIS window.
ready = from(bucket: "pipeline_checkpoints")
    |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
    |> filter(fn: (r) => r.task_name == "validate_sensor_data")
    |> filter(fn: (r) => r._field == "completed")
    |> count()
    |> findColumn(fn: (key) => true, column: "_value")

// 2. Only materialize the rollup when the prerequisite is present.
if length(arr: ready) > 0 and ready[0] > 0 then
    from(bucket: "validated_telemetry")
        |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
        |> filter(fn: (r) => r._field == "temperature")
        |> aggregateWindow(every: 1h, fn: mean, createEmpty: false)
        |> to(bucket: "telemetry_hourly")
else
    // No-op: exit and let the scheduler retry on the next interval.
    array.from(rows: [{_time: now(), _measurement: "skipped", _field: "reason", _value: "upstream_not_ready"}])
        |> to(bucket: "pipeline_checkpoints")

This gating pattern scales cleanly when paired with an external control plane — the Python client orchestration patterns guide covers wrapping these tasks in a runner that owns retries and cross-system I/O.

# Step 5 — Validate the graph before you deploy it

Before pushing complex pipelines to production, validate the dependency graph programmatically to catch cycles, orphaned nodes, or a misordered edge. Python’s graphlib.TopologicalSorter does this with zero external dependencies and yields a safe execution order in the same pass:

from graphlib import TopologicalSorter, CycleError

# Nodes mapped to their upstream dependencies (the edges of the DAG).
pipeline_graph = {
    "ingest_raw": [],
    "validate_telemetry": ["ingest_raw"],
    "downsample_1m": ["validate_telemetry"],
    "downsample_5m": ["validate_telemetry"],
    "detect_anomalies": ["downsample_1m", "downsample_5m"],
    "archive_cold": ["detect_anomalies"],
}

def validate_and_sort(graph: dict) -> list:
    ts = TopologicalSorter(graph)
    try:
        ts.prepare()                       # raises CycleError if the graph is cyclic
        return list(ts.static_order())     # a dependency-respecting execution order
    except CycleError as e:
        raise RuntimeError(f"DAG validation failed — cycle detected: {e.args[1]}")

if __name__ == "__main__":
    order = validate_and_sort(pipeline_graph)
    print("Valid DAG. Execution order:", order)

Topological sorting guarantees no task is scheduled before its prerequisites, and prepare() is where a stray back-edge that would create an infinite retry loop is caught at deploy time rather than at 3 a.m. For a fuller control plane that tracks per-window node state and drives the API, see building dependency graphs for multi-stage pipeline execution.

# Configuration Reference

The parameters below carry most of the operational weight when wiring dependent tasks. Get the offset and concurrency wrong and the graph enforces nothing.

# Common Failure Modes and Fixes

1. Offset too small for late IoT data. Symptom: downstream rollups intermittently average fewer points than expected; the tail of each window is missing. Root cause: the downstream offset is shorter than the worst-case upstream processing latency, so the gate reads before validation finishes writing. Fix: size the offset from observed p99 upstream latency, not the mean, and verify it against the checkpoint timestamps rather than guessing.

2. Hidden cycle from a corrective back-edge. Symptom: two tasks retrigger each other forever and CPU on the task engine climbs. Root cause: someone added an edge from a downstream “correction” task back into an upstream node to reprocess data, turning the graph cyclic. Fix: run validate_and_sort() in CI on every task change — prepare() raises CycleError before deploy. Model corrections as a new forward node over the same window, never a back-edge.

3. Checkpoint keyed to now() instead of the window. Symptom: the gate passes even when the correct window never ran, because it only checks that something completed recently. Root cause: the checkpoint _time was set to now(), so per-window lookups collapse into “did anything run.” Fix: stamp the checkpoint with v.timeRangeStop as in Step 3, and have the gate range over the same slice.

4. Duplicate rollup points after a retry. Symptom: aggregated series show doubled values for isolated windows. Root cause: a slow run overlapped its successor because concurrency was left unbounded, and the window was not anchored to logical run time. Fix: set concurrency: 1 on every write-producing node and anchor range() to v.timeRangeStart/v.timeRangeStop so a rewrite overwrites rather than appends.

5. Orphaned node that never fires. Symptom: a stage’s target bucket stays empty and no error appears. Root cause: its gate waits on a checkpoint from an upstream task whose name was renamed or misspelled, so the edge can never resolve. Fix: treat task names as a shared contract; assert every task_name referenced by a gate exists in the deployed graph as part of the same CI validation step.

# Verification and Testing

Validating the graph statically is necessary but not sufficient — you also need runtime confirmation that edges actually resolve in production. Two checks cover this.

First, confirm that each window produces exactly one upstream checkpoint and one downstream materialization. Query the checkpoint bucket and count completions per window:

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)   // surfaces windows with 0 (missed) or >1 (double-run)

Any row this returns is a broken window — a 0 means the node never fired for that slice, a value above 1 means it ran more than once.

Second, add a deadman health check so a silently stalled graph raises an alert instead of decaying quietly. A deadman watches for the absence of a fresh checkpoint, which is the only signal a stopped scheduler emits:

import "influxdata/influxdb/monitor"
import "experimental"

from(bucket: "pipeline_checkpoints")
    |> range(start: -15m)
    |> filter(fn: (r) => r.task_name == "validate_sensor_data")
    |> monitor.deadman(t: experimental.subDuration(d: 15m, from: now()))
    // a `dead: true` row here means no checkpoint arrived in the window -> page on-call

Deadman checks are the highest-value alert in any dependency-mapped pipeline precisely because a stalled node produces no error to catch — only the missing output reveals it.

# Integration Points

Dependency mapping sits at the center of the scheduling discipline and touches every adjacent topic on this site. The gating logic in Steps 3–4 is written in Flux, so the correctness rules for retry-safe, column-pruned scripts in Flux scripting for task automation apply directly to every node. The offset and cadence decisions that keep edges from resolving too early are governed by cron & interval scheduling logic. When the graph outgrows what in-database gating can express — conditional branches, cross-system calls, dynamic fan-out — the control plane moves into Python client orchestration patterns.

Downstream, the aggregation nodes these graphs sequence are designed in downsampling and aggregation pipeline design, and the archival node at the terminal edge is bounded by retention policy design. The official InfluxDB documentation on processing data with tasks provides the underlying reference for task execution and resource allocation.

# Frequently Asked Questions

# Related

• Building dependency graphs for multi-stage pipeline execution — a full Python control plane that tracks per-window node state and drives the InfluxDB API.
• Flux scripting for task automation — write the retry-safe, idempotent Flux each graph node depends on.
• Cron and interval scheduling logic — tune the offsets and cadence that keep edges from resolving too early.
• Python client orchestration patterns — move the control plane outward for branching and cross-system I/O.
• Downsampling and aggregation pipeline design — design the aggregation nodes these graphs sequence.

Up: Automated Task Scheduling & Orchestration — the parent guide covering native and external execution across the full data lifecycle.