# Airflow + InfluxDB Integration

InfluxDB’s native task engine is excellent at running a single Flux script on a fixed cadence, but it has no notion of a task graph, no first-class backfill primitive, and no way to gate one job on the data quality of another. The moment a time-series pipeline needs to say “downsample last night’s raw SCADA, but only proceed to the daily energy-yield rollup once the ten-minute tier passes a completeness check, and re-run any night that failed without double-writing,” you have outgrown the built-in scheduler. This page sits under External Orchestration & Workflow Engines and shows how to wire Apache Airflow in front of InfluxDB as the external orchestrator: modelling the pipeline as a DAG, calling InfluxDB with influxdb-client from a TaskFlow task, gating progress on a data-quality sensor, and running idempotent backfills whose windows line up exactly with InfluxDB’s aggregation boundaries.

# The failure scenario this solves

A renewable-energy operator runs a SCADA fleet of 480 wind turbines across 12 farms. Each turbine writes power output, rotor RPM, nacelle wind speed, and gearbox oil temperature at 1-second resolution into a bucket named scada_raw. Two rollups feed the analytics stack: a ten-minute average (the IEC 61400 standard resource-assessment window) in scada_10m, and a daily energy-yield aggregate in scada_daily that the commercial team uses for availability reporting and warranty claims.

For months this ran as two independent native InfluxDB tasks on cron schedules. Then a firmware push at one farm delayed telemetry by roughly forty minutes on a single night. The ten-minute task ran on schedule against a window that was only two-thirds populated and wrote averages that silently understated output. The daily task ran an hour later, read those thin ten-minute rows, and produced a yield figure 4.1% low for 40 turbines. Nobody caught it until a warranty audit six weeks later, because the two tasks had no dependency between them and no gate: the daily rollup had no idea the tier beneath it was incomplete. Re-running was worse — the naive fix re-executed the ten-minute task, which appended a second set of points for the same timestamps, and the daily average then double-counted them.

The root cause is not InfluxDB and not the firmware. It is that a pipeline with a real dependency, a real quality gate, and a real backfill requirement was being run by a scheduler that models none of those things. Airflow models all three. The rest of this page rebuilds the pipeline as a DAG whose edges encode the dependency, whose sensor encodes the gate, and whose tasks are idempotent so a backfill of any night is safe to run twice.

# Prerequisites

• Apache Airflow 2.7+ (TaskFlow API and the @task.sensor decorator) running with the LocalExecutor, CeleryExecutor, or KubernetesExecutor.
• InfluxDB 2.7+ (or InfluxDB 3.x Cloud/Dedicated) with the buckets scada_raw, scada_10m, and scada_daily already provisioned.
• Python 3.9+ and influxdb-client 1.36+ installed in the Airflow worker image (pip install influxdb-client).
• A per-pipeline InfluxDB token scoped to read scada_raw / scada_10m and write scada_10m / scada_daily — never an all-access token.
• An Airflow Connection or secrets backend entry to hold the InfluxDB URL, org, and token (details below); tokens must not appear in DAG source.
• Timestamps stored in RFC 3339 / UTC so Airflow’s data_interval maps cleanly onto Flux range() bounds; see the RFC 3339 timestamp specification.

# Core concept: the DAG as the source of truth for time windows

Airflow’s scheduler does one thing InfluxDB’s cannot: it treats every run as a bounded data interval, not just a fire time. A scheduled DAG run carries data_interval_start and data_interval_end — the half-open window [start, end) of the data that run is responsible for. When you map that interval directly onto a Flux range(start:, stop:), the orchestrator and the database agree, byte for byte, on which slice of time is being processed. That single alignment is what makes backfills and retries deterministic: re-running the DAG for 2026-03-02 reprocesses exactly the same window, because the window is a property of the run, not of the wall clock at execution time.

The dependency between tiers becomes a DAG edge. The gate becomes a sensor that blocks the edge until the ten-minute tier is complete. And “complete” is measurable: at 1 Hz across a fleet of N turbines, a fully populated window of duration Δt should hold a known number of raw points, so a completeness ratio can gate the downstream rollup.

$$E_{window} = N_{turbines} \times \frac{\Delta t_{window}}{p_{sample}}$$

For the ten-minute window across the full fleet, $E_{window} = 480 \times \frac{600,\text{s}}{1,\text{s}} = 288{,}000$ expected raw points. The data-quality sensor computes the observed count over the same interval and only lets the daily aggregation proceed once the ratio clears a threshold (say 98%), so a firmware-delayed farm parks the pipeline instead of silently corrupting the yield report.

InfluxDB still does the heavy lifting — the aggregation runs inside the database as Flux, not in Python — but Airflow owns when it runs, whether the next stage may proceed, and how a missed night is reprocessed. Modelling that dependency correctly is the subject of dependency mapping & DAG construction; here we apply it against a live InfluxDB backend.

# Step-by-step implementation

# 1. Model the pipeline as a DAG

Start with the graph, not the queries. The DAG declares three tasks and two edges: downsample runs first, the quality sensor gates the middle, and the daily aggregation depends on the gate passing. Disable catchup at first so a fresh deploy does not stampede every historical interval, and cap parallelism so a manual backfill cannot saturate InfluxDB.

from datetime import datetime
from airflow.decorators import dag

@dag(
    dag_id="scada_downsampling",
    schedule="@daily",              # one run per calendar day, window-bounded
    start_date=datetime(2026, 1, 1),
    catchup=False,                  # opt in to backfills deliberately
    max_active_runs=3,              # bound concurrent windows against InfluxDB
    default_args={"retries": 3, "retry_delay": 300},
    tags=["influxdb", "scada"],
)
def scada_pipeline():
    ...

scada_pipeline()

The schedule="@daily" cadence means each run’s data_interval is exactly one calendar day in UTC — the window every downstream Flux query will bind to. Airflow’s model of DAGs, tasks, and data intervals is documented in the Airflow core concepts guide; the mapping from those concepts to InfluxDB is tabulated later on this page.

# 2. Store the InfluxDB connection and token outside the DAG

The token must never be a literal in DAG source. Register an Airflow Connection whose extra carries the org and bucket, and whose password field holds the token — Airflow encrypts it in the metadata DB (or fetches it from a secrets backend such as Vault or AWS Secrets Manager). Create it once from the CLI, not in code:

airflow connections add influxdb_scada \
  --conn-type generic \
  --conn-host "https://influx.internal:8086" \
  --conn-password "$INFLUX_TOKEN" \
  --conn-extra '{"org": "wind-ops", "raw_bucket": "scada_raw", "ten_min_bucket": "scada_10m", "daily_bucket": "scada_daily"}'

At runtime a helper resolves the Connection into an InfluxDBClient, so no task body ever sees a raw credential:

from airflow.hooks.base import BaseHook
from influxdb_client import InfluxDBClient

def influx_client() -> InfluxDBClient:
    conn = BaseHook.get_connection("influxdb_scada")
    return InfluxDBClient(url=conn.host, token=conn.password,
                          org=conn.extra_dejson["org"])

# 3. Write the downsampling task with TaskFlow and influxdb-client

The first task rolls raw SCADA into ten-minute means. Critically, it binds the Flux range() to the run’s data_interval — passed in through the TaskFlow context — rather than to a relative -24h. That is what makes the task reproducible: the same logical day always reads and writes the same window, which is the precondition for safe retries and backfills.

from airflow.decorators import task

@task
def downsample_10m(data_interval_start=None, data_interval_end=None):
    start = data_interval_start.strftime("%Y-%m-%dT%H:%M:%SZ")
    stop = data_interval_end.strftime("%Y-%m-%dT%H:%M:%SZ")
    flux = f'''
        from(bucket: "scada_raw")
          |> range(start: {start}, stop: {stop})
          |> filter(fn: (r) => r._measurement == "turbine")
          |> filter(fn: (r) => r._field == "power_kw" or r._field == "rotor_rpm")
          |> aggregateWindow(every: 10m, fn: mean, createEmpty: false)
          |> set(key: "_measurement", value: "turbine_10m")
          |> to(bucket: "scada_10m", org: "wind-ops")
    '''
    with influx_client() as client:
        client.query_api().query(flux)

Two parameters carry the correctness. stop: is the exclusive interval end, so adjacent days never overlap by a single point, and aggregateWindow(every: 10m) bins on wall-clock boundaries that line up with the IEC 61400 windows the analytics team expects. The Python-side orchestration idioms — client lifecycle, connection reuse, error handling — are developed in Python client orchestration patterns.

# 4. Gate the pipeline on a data-quality sensor

Before the daily rollup runs, a sensor confirms the ten-minute tier is actually complete for the interval. It counts raw points in the window and compares against the expected $E_{window}$; only a ratio above threshold lets the DAG proceed. A reschedule-mode sensor frees the worker slot between pokes instead of holding it.

@task.sensor(poke_interval=300, timeout=7200, mode="reschedule")
def qc_completeness(data_interval_start=None, data_interval_end=None):
    from airflow.sensors.base import PokeReturnValue
    start = data_interval_start.strftime("%Y-%m-%dT%H:%M:%SZ")
    stop = data_interval_end.strftime("%Y-%m-%dT%H:%M:%SZ")
    expected = 480 * 86400          # 480 turbines * 1 Hz * 1 day
    flux = f'''
        from(bucket: "scada_raw")
          |> range(start: {start}, stop: {stop})
          |> filter(fn: (r) => r._measurement == "turbine" and r._field == "power_kw")
          |> count()
          |> sum()
    '''
    with influx_client() as client:
        tables = client.query_api().query(flux)
        observed = tables[0].records[0].get_value() if tables else 0
    ratio = observed / expected
    return PokeReturnValue(is_done=ratio >= 0.98, xcom_value=round(ratio, 4))

If a farm is delayed, the ratio stays below 0.98, the sensor keeps rescheduling until the data lands (or times out and alerts), and the daily aggregate simply never runs on thin data. That gate is the piece the original two-task setup lacked.

# 5. Trigger the dependent daily aggregation

The final task reads the validated ten-minute tier and produces daily energy yield. Because it is wired downstream of the sensor in the DAG body, Airflow will not start it until the gate returns done.

@task
def aggregate_daily(data_interval_start=None, data_interval_end=None):
    start = data_interval_start.strftime("%Y-%m-%dT%H:%M:%SZ")
    stop = data_interval_end.strftime("%Y-%m-%dT%H:%M:%SZ")
    flux = f'''
        from(bucket: "scada_10m")
          |> range(start: {start}, stop: {stop})
          |> filter(fn: (r) => r._measurement == "turbine_10m" and r._field == "power_kw")
          |> aggregateWindow(every: 1d, fn: mean, createEmpty: false)
          |> map(fn: (r) => ({{ r with _value: r._value * 24.0 }}))   // mean kW -> kWh/day
          |> set(key: "_measurement", value: "turbine_daily")
          |> to(bucket: "scada_daily", org: "wind-ops")
    '''
    with influx_client() as client:
        client.query_api().query(flux)

@dag(...)
def scada_pipeline():
    rolled = downsample_10m()
    gate = qc_completeness()
    daily = aggregate_daily()
    rolled >> gate >> daily          # the dependency the original setup lacked

The >> operators are the edges the scheduler enforces. This is where a broader multi-stage yield pipeline connects to the wider downsampling & aggregation pipeline design — Airflow supplies ordering and gating, InfluxDB supplies the transforms.

# 6. Make backfills idempotent and window-aligned

To reprocess history, run airflow dags backfill -s 2026-02-01 -e 2026-02-14 scada_downsampling. Each historical run rebinds its Flux range() to that day’s interval, so windows always align — but a re-run must not append duplicate points. InfluxDB deduplicates on the line-protocol series key plus timestamp, so idempotency is guaranteed only if the rollup writes stable tags and timestamps. aggregateWindow emits deterministic bin timestamps, so a second run overwrites rather than doubles — provided you never add a run-specific tag (like an execution id) to the output.

# Idempotent: same interval -> same bin timestamps + tags -> overwrite, not append.
# Backfill 2026-02-01..2026-02-14 is safe to run twice.
#   airflow dags backfill -s 2026-02-01 -e 2026-02-14 scada_downsampling
#
# NON-idempotent (do NOT do this): tagging output with the run id makes every
# backfill a new series, so the daily mean double-counts.
#   |> set(key: "airflow_run_id", value: "{run_id}")   # <-- breaks dedup

# Configuration reference

# Common failure modes and fixes

1. Airflow logical_date vs InfluxDB window drift. Symptom: rollups appear shifted by exactly one interval — the run labelled 2026-03-02 writes data for March 1st, or a day is skipped entirely. Root cause: using the legacy execution_date/logical_date (the start of the interval) as if it were the end, or mixing it with a relative range(start: -24h) evaluated at execution time. Fix: always bind Flux to data_interval_start and data_interval_end and treat stop: as exclusive.

# WRONG: relative window, evaluated whenever the worker happens to run.
'|> range(start: -24h)'
# RIGHT: the run owns its window; retries and backfills reprocess it identically.
f'|> range(start: {data_interval_start:%Y-%m-%dT%H:%M:%SZ}, stop: {data_interval_end:%Y-%m-%dT%H:%M:%SZ})'

2. Non-idempotent backfill double-writes rollups. Symptom: re-running a night inflates daily yield; a twice-run backfill roughly doubles a tier’s values. Root cause: the output carries a run-specific tag (execution id, timestamp of run) so each backfill creates a new series instead of overwriting the existing points. Fix: write only stable tags and deterministic aggregateWindow timestamps so InfluxDB deduplicates on the series key; verify a re-run is a no-op before trusting the backfill.

3. Secrets baked into DAG code. Symptom: a token appears in the Git history, the Airflow UI’s rendered task source, or a task log. Root cause: the InfluxDBClient(token="...") literal lives in the DAG file. Fix: store the token in an Airflow Connection or secrets backend and resolve it at runtime via a hook; rotate any token that ever touched source control.

# WRONG: token in source, visible in the UI and Git.
client = InfluxDBClient(url="https://influx:8086", token="abc123==", org="wind-ops")
# RIGHT: resolved from the encrypted Connection at runtime.
client = influx_client()

4. Catchup storm saturates InfluxDB. Symptom: the first deploy of a DAG with an early start_date launches hundreds of runs at once and InfluxDB query memory spikes or OOMs. Root cause: catchup=True with no max_active_runs cap replays every historical interval simultaneously. Fix: set catchup=False, cap max_active_runs, and use a dedicated Airflow pool so concurrent Flux queries against InfluxDB stay bounded.

5. Sensor holds a worker slot for hours. Symptom: the pool drains and unrelated DAGs stall while a completeness sensor waits on a delayed farm. Root cause: a poke-mode sensor occupies its worker slot for the full timeout. Fix: run the sensor in mode="reschedule" so it releases the slot between pokes, and set a timeout that alerts rather than blocking indefinitely.

# Verification and testing

Verify three things: that a backfill reprocesses the exact window without duplication, that the ten-minute tier holds the expected point volume, and that a stalled pipeline raises an alert while the raw data still exists to reprocess.

Confirm the ten-minute rollup landed the right shape for a given day directly in Flux:

from(bucket: "scada_10m")
  |> range(start: 2026-03-02T00:00:00Z, stop: 2026-03-03T00:00:00Z)
  |> filter(fn: (r) => r._measurement == "turbine_10m" and r._field == "power_kw")
  |> count()
  |> group()
  |> sum()
// Expect 480 turbines * 144 ten-min bins = 69,120 rows for a full day.

Then add a deadman health check so a downsampling task that silently stops writing pages an operator while scada_raw still holds the source samples to reprocess:

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

from(bucket: "scada_10m")
  |> range(start: -2h)
  |> filter(fn: (r) => r._measurement == "turbine_10m" and r._field == "power_kw")
  |> monitor.deadman(t: experimental.subDuration(from: now(), d: 90m))
  |> filter(fn: (r) => r.dead == true)

If the deadman returns rows, no fresh ten-minute data has been written in 90 minutes — the Airflow task is stuck or failing — and the alert fires while the raw tier is still intact. Finally, prove idempotency in a staging org by running the same backfill twice and asserting the daily yield is unchanged:

from influxdb_client import InfluxDBClient
import os, subprocess

def daily_yield(day: str) -> float:
    with InfluxDBClient(url=os.environ["INFLUX_URL"], token=os.environ["INFLUX_TOKEN"],
                        org="wind-ops") as client:
        flux = f'''
            from(bucket: "scada_daily")
              |> range(start: {day}T00:00:00Z, stop: {day}T23:59:59Z)
              |> filter(fn: (r) => r._measurement == "turbine_daily")
              |> sum()
        '''
        tables = client.query_api().query(flux)
        return sum(r.get_value() for t in tables for r in t.records)

cmd = ["airflow", "dags", "backfill", "-s", "2026-03-02", "-e", "2026-03-02", "scada_downsampling"]
subprocess.run(cmd, check=True)
first = daily_yield("2026-03-02")
subprocess.run(cmd, check=True)          # run the identical backfill again
second = daily_yield("2026-03-02")
assert abs(first - second) < 1e-6, f"NOT idempotent: {first} != {second}"
print(f"[OK] idempotent daily yield: {first:.1f} kWh")

# Integration points

Airflow is the orchestration layer, so it touches most of the pipeline. The Python inside each task — client reuse, batching, retry semantics against a flaky endpoint — follows the platform’s client orchestration patterns, while the DAG edges themselves apply the dependency-mapping techniques covered earlier. Whether Airflow is even the right engine for a given pipeline — versus native tasks, Prefect, or Dagster — is weighed in native vs external orchestration: a decision framework, and for lighter, single-stage jobs the same trade-offs point toward Kubernetes CronJob orchestration instead of a full scheduler. For the concrete DAG wiring of a downsampling job — file layout, TaskFlow imports, and the backfill command end to end — see scheduling InfluxDB downsampling with Airflow DAGs. The influxdb-client query and write endpoints these tasks call are documented in the InfluxDB Python client reference.

# FAQ

# When should I use Airflow instead of native InfluxDB tasks?

Reach for Airflow when the pipeline has dependencies, quality gates, or backfill requirements that InfluxDB’s flat scheduler cannot express. A single unconditional rollup on a fixed cadence belongs in a native task; a graph where the daily aggregation must wait on a completeness check of the ten-minute tier, and where any past day must be reprocessible on demand, is exactly what Airflow’s DAG model, sensors, and backfill command exist for.

# How do I map Airflow’s logical date to an InfluxDB time range?

Bind the Flux range() to data_interval_start and data_interval_end, treating stop: as exclusive, and never use a relative window like -24h. The data interval is a property of the run, so every retry and backfill of that run reprocesses precisely the same window, which is what keeps rollups deterministic instead of drifting by an interval.

# How do I keep backfills from double-writing rollups?

Write only stable tags and let aggregateWindow produce deterministic bin timestamps, so InfluxDB deduplicates on the series key and a re-run overwrites rather than appends. The moment you add a run-specific tag such as the execution id, every backfill becomes a new series and downstream means double-count. Test it by running an identical backfill twice and asserting the aggregate is unchanged.

# Where should the InfluxDB token live?

In an Airflow Connection or an external secrets backend such as Vault or AWS Secrets Manager, resolved at runtime through a hook — never as a literal in the DAG file, which would expose it in Git, the rendered-source view, and task logs. Scope the token to only the buckets the pipeline reads and writes, and rotate any token that has ever appeared in source.

# How do I stop a backfill from overwhelming InfluxDB?

Leave catchup=False, cap max_active_runs, and route the tasks through a dedicated Airflow pool so the number of concurrent Flux queries against InfluxDB stays bounded. Trigger history deliberately with airflow dags backfill over an explicit date range rather than letting a DAG replay years of intervals the moment it is deployed.

# Related

• Dependency Mapping & DAG Construction — the fan-out, join, and gating patterns behind the DAG edges used here.
• Python Client Orchestration Patterns — client lifecycle, batching, and retry idioms for the influxdb-client calls inside each task.
• Scheduling InfluxDB downsampling with Airflow DAGs — the concrete DAG file, TaskFlow layout, and backfill command, step by step.
• Downsampling & Aggregation Pipeline Design — what the ten-minute and daily rollups should actually compute.
• External Orchestration & Workflow Engines — where Airflow fits among the external orchestrators for InfluxDB.

Up one level: External Orchestration & Workflow Engines