InfluxDB tasks vs Airflow for time-series pipelines
You run a telco radio-access pipeline: roughly 4,800 cell towers each stream RRC setup-success ratio, PRB utilisation, and downlink throughput into a cell_kpi_raw bucket every 15 seconds, and the network operations centre needs a 5-minute rollup plus a congestion alert when PRB utilisation crosses 85% for two consecutive windows. The rollup and the alert are trivial in isolation — the hard call is where the scheduler lives. A native InfluxDB task runs the Flux inside the database next to the data; an Airflow DAG runs the same logic from an external Python worker that reaches in over the network. That single architectural choice decides your end-to-end latency, your on-call surface, and how painful a three-day backfill is. This page builds the identical workflow both ways and compares them on the axes that actually matter, extending the trade-off analysis in Native vs External Orchestration: A Decision Framework.
Prerequisites
Solution walkthrough
1. Build it natively: a Flux task inside the database
The native path is a single scheduled Flux script. It reads the last window from cell_kpi_raw, writes 5-minute means to cell_kpi_5m, and — in the same engine — evaluates the congestion rule with monitor.check. No process leaves the database, and the scheduler is the InfluxDB task engine itself.
import "influxdata/influxdb/monitor"
import "influxdata/influxdb/schema"
option task = {name: "cell-kpi-5m-rollup", every: 5m, offset: 45s}
src = from(bucket: "cell_kpi_raw")
|> range(start: -task.every)
|> filter(fn: (r) => r._measurement == "radio_kpi")
// 1. Downsample to the aggregate tier.
src
|> aggregateWindow(every: 5m, fn: mean, createEmpty: false)
|> to(bucket: "cell_kpi_5m", org: "network-ops")
// 2. Evaluate the congestion threshold on the fresh rollup.
check = {_check_id: "prb-congestion", _check_name: "PRB congestion", _type: "threshold", tags: {}}
src
|> filter(fn: (r) => r._field == "prb_util")
|> aggregateWindow(every: 5m, fn: mean, createEmpty: false)
|> monitor.check(
crit: (r) => r._value >= 85.0,
ok: (r) => r._value < 85.0,
messageFn: (r) => "cell ${r.cell_id} PRB ${string(v: r._value)}%",
data: check,
)
The offset: 45s holds the window open long enough for late 15-second samples to land before the task fires — the same late-data reasoning covered across automated task scheduling and orchestration. Deploy it with influx task create --file cell-kpi-5m-rollup.flux. Statuses land in the _monitoring bucket where a downstream notification rule can pick up crit transitions. There is nothing else to run: the scheduler, the compute, and the data share one process.
2. Build it externally: the same logic as an Airflow DAG
The Airflow path moves the schedule and the compute outside the database. A TaskFlow DAG runs on a */5 cron, queries the raw window over HTTP, and issues the same downsample and threshold logic from a worker. The database becomes a passive query target; Apache Airflow owns orchestration, retries, and history.
from datetime import datetime, timedelta
from airflow.decorators import dag, task
from influxdb_client import InfluxDBClient
import os
@dag(
schedule="*/5 * * * *",
start_date=datetime(2026, 3, 1),
catchup=False,
default_args={"retries": 3, "retry_delay": timedelta(seconds=30)},
max_active_runs=1,
tags=["influxdb", "cell-kpi"],
)
def cell_kpi_5m_rollup():
@task
def downsample(data_interval_start=None, data_interval_end=None):
flux = f'''
from(bucket: "cell_kpi_raw")
|> range(start: {data_interval_start.isoformat()}, stop: {data_interval_end.isoformat()})
|> filter(fn: (r) => r._measurement == "radio_kpi")
|> aggregateWindow(every: 5m, fn: mean, createEmpty: false)
|> to(bucket: "cell_kpi_5m", org: "network-ops")
'''
with InfluxDBClient(url=os.environ["INFLUX_URL"],
token=os.environ["INFLUX_TOKEN"],
org="network-ops") as c:
c.query_api().query(flux) # to() executes server-side
@task
def alert(data_interval_start=None, data_interval_end=None):
flux = f'''
from(bucket: "cell_kpi_5m")
|> range(start: {data_interval_start.isoformat()}, stop: {data_interval_end.isoformat()})
|> filter(fn: (r) => r._field == "prb_util" and r._value >= 85.0)
'''
with InfluxDBClient(url=os.environ["INFLUX_URL"],
token=os.environ["INFLUX_TOKEN"],
org="network-ops") as c:
breaches = c.query_api().query(flux)
if breaches:
notify_noc(breaches) # your webhook / PagerDuty call
alert().set_upstream(downsample())
cell_kpi_5m_rollup()
Binding the query to data_interval_start/data_interval_end rather than now() is what makes the DAG idempotent and backfill-safe — a rerun of the 09:05 interval always reprocesses the same window. The trade-off is now visible in the code: two network round-trips, an external worker to keep alive, and connection/token handling you own. Detailed connection, secret, and sensor patterns for this variant live in Airflow + InfluxDB Integration.
3. Compare on the axes that decide it
Both pipelines produce the identical cell_kpi_5m rollup and the identical congestion signal. They diverge on operational properties. End-to-end alert latency decomposes as
$$ L_{e2e} = t_{sched} + t_{query} + t_{compute} + t_{network} $$
where the native task drives $t_{network} \approx 0$ because compute runs beside the data, while the Airflow path pays two t_{network} legs plus worker scheduling slack.
| Dimension | Native InfluxDB task | Airflow DAG |
|---|---|---|
| Where compute runs | Inside the database engine | External worker, queries over HTTP |
| Added infrastructure | None | Scheduler, metadata DB, workers |
| End-to-end latency | Lowest (no network legs) | Higher (2 round-trips + queue slack) |
| Backfill / catchup | Manual re-run per window | First-class (catchup, date ranges) |
| Retries & history | Task run log only | Rich retries, XCom, run history |
| Cross-system steps | Flux-only; hard to leave InfluxDB | Trivial (S3, warehouse, Slack in one DAG) |
| Observability | _tasks/_monitoring buckets |
Web UI, per-task logs, SLA misses |
| Failure blast radius | Contained to one task | Worker/scheduler outage stops all DAGs |
| Ops overhead | Near zero | Real (a service to run and patch) |
Observability deserves its own line because it flips the intuition. The native task is simpler but quieter: run history lives in the _tasks bucket and you inspect it with Flux, so a skipped run is something you have to query for rather than something that pages you. Airflow is heavier but louder: the web UI surfaces missed SLAs, per-task logs, and retry counts without extra work, which is worth a great deal once dozens of pipelines share the platform. For a two-tower KPI job the native task’s quietness is fine; for a fleet of forty interdependent workflows, Airflow’s built-in visibility is often the deciding factor on its own.
The rule that falls out: keep it native when the whole job begins and ends in InfluxDB and low latency matters — a single-bucket downsample-and-alert like this one is the textbook native case. Reach for Airflow when the workflow spans systems (page a cold partition to S3, then load a warehouse, then notify), when you need heavy backfills over historical ranges, or when a platform team already runs Airflow and wants one control plane. A useful smoke test: if the workflow’s data-flow diagram has exactly one system in it, default to native and only escalate to Airflow when a concrete cross-system or backfill requirement forces the move. For the two-engine comparison among external Python schedulers instead, see Prefect vs Dagster for InfluxDB orchestration.
Gotchas and edge cases
Double-scheduling the same rollup. If you prototype the native task and then bring up the Airflow DAG without disabling the task, both write to cell_kpi_5m on overlapping cadences. Because aggregateWindow + to() is deterministic the values match, but you double the read load on cell_kpi_raw and burn task-engine slots for nothing. Pick one runner per target bucket and set influx task update --status inactive on the loser before the other goes live.
Airflow now() drift breaks idempotency. A DAG that queries range(start: -5m) against wall-clock now() instead of data_interval_start reprocesses a different window on every retry and on every backfill, so a rerun silently rolls up the wrong five minutes. Always bind the Flux range to the run’s data interval; that single change is what makes catchup and manual reruns safe.
Retention outrunning an external schedule. The native task lives in the same engine as retention, so it fires reliably ahead of expiry. An Airflow worker outage, by contrast, can stall the rollup while cell_kpi_raw’s 7-day retention keeps sweeping — a long enough outage drops raw windows before the DAG ever reads them. Size the source retention to exceed your worst-case orchestrator downtime, and alert on DAG staleness, not just on task failure.
Token scope creep on the external worker. It is tempting to hand the Airflow worker an all-access token so one credential covers every DAG, but that worker is a separate, internet-adjacent process and a leaked all-access token exposes the whole org. Issue the DAG a token scoped to read cell_kpi_raw and write cell_kpi_5m and nothing else; the native task needs no external credential at all because it executes inside the engine under the task owner’s authority. This asymmetry — external runners multiply the credential surface — is a real, if often overlooked, cost on the Airflow side of the ledger.
Verification
Prove both runners produce the same rollup by counting 5-minute points landed in the aggregate bucket for the last hour — you expect roughly 12 windows per active cell:
from(bucket: "cell_kpi_5m")
|> range(start: -1h)
|> filter(fn: (r) => r._measurement == "radio_kpi" and r._field == "prb_util")
|> group(columns: ["cell_id"])
|> count()
|> group()
|> mean()
|> yield(name: "avg_windows_per_cell_should_be_~12")
For the native task specifically, confirm the engine is actually running it on cadence rather than silently erroring:
influx task list --org network-ops --token $INFLUX_TOKEN
# Note the task ID for cell-kpi-5m-rollup, then inspect recent runs:
influx task run list --task-id <TASK_ID> --org network-ops --token $INFLUX_TOKEN
# Every run should show status 'success'; a run gap means the scheduler skipped it.
A steady ~12 windows per cell and an unbroken run of success statuses (or, for the DAG, green task instances with no missed intervals) confirms whichever runner you chose is keeping pace with ingest.
FAQ
Can I run the downsample natively and the alerting in Airflow?
Yes, and it is a common hybrid: let the native task own the low-latency, single-bucket rollup, and let an Airflow DAG read cell_kpi_5m to fan alerts out to PagerDuty, a warehouse, and Slack. You get native latency where it matters and Airflow’s cross-system reach for the notification stage. Just keep exactly one writer per target bucket to avoid the double-scheduling trap above.
Does Airflow add meaningful latency to a 5-minute pipeline?
For a 5-minute rollup the extra network legs and worker-queue slack add seconds, which is usually irrelevant. It only bites when you need near-real-time alerting (sub-minute), where the native task’s in-engine execution and lack of round-trips give it a clear edge. Match the runner to your latency budget rather than to fashion.
Which is cheaper to operate?
For a workflow that never leaves InfluxDB, the native task is cheaper on every axis — no scheduler, metadata database, or worker fleet to run, patch, and monitor. Airflow earns its operating cost once you have many cross-system pipelines that benefit from one shared control plane, rich backfills, and a mature UI; running it for a single downsample task is over-engineering.
Related
- Native vs External Orchestration: A Decision Framework — the broader decision model this head-to-head instantiates.
- Airflow + InfluxDB Integration — connections, secrets, and sensor patterns for the external variant.
- Automated Task Scheduling & Orchestration — the native scheduling, Flux, and Python patterns the first path builds on.