# Continuous Query Migration to Tasks

Moving from InfluxDB 1.x continuous queries (CQs) to 2.x tasks is where a lot of previously invisible behavior suddenly becomes your responsibility. A CQ was a database-managed background process: it inferred its execution interval from a GROUP BY time() clause, backfilled missing intervals on its own, absorbed clock skew silently, and logged failures only to the daemon journal. A task is an explicit Flux script with a declared schedule, an explicit read window, and per-run status you can query. That shift is powerful, but it means every implicit convenience the CQ engine gave you for free must now be written out — and if you miss one, the failure is usually silent data corruption rather than an error. This page is the reference for making that translation correctly within downsampling & aggregation pipeline design, covering the semantic gaps between the two engines, runnable translations, the boundary math that prevents duplicates and gaps, and how to prove the migrated task matches the CQ it replaced.

# The failure scenario this solves

A team runs a 1.x CQ that downsamples per-second temperature to hourly means: CREATE CONTINUOUS QUERY "cq_1h" ON "iot" BEGIN SELECT mean("value") INTO "iot"."long"."temp_1h" FROM "temp" GROUP BY time(1h) END. It has run untouched for two years. During the 2.x cutover an engineer translates it to what looks like the obvious Flux equivalent: a task with every: 1h and range(start: -1h), aggregating with aggregateWindow(every: 1h, fn: mean).

In staging it looks identical. In production three things go wrong at once. First, the task fires at the top of each hour and reads the hour that just closed — but edge gateways buffer readings during cellular gaps and flush them minutes late, so the last few minutes of every hour are aggregated before the late points land, and those points are never picked up because the next run’s window starts after them. Second, when a run is retried after an outage, the naive range(start: -1h) window is anchored to wall-clock now, not to the missed period, so the backfill the CQ engine used to do automatically simply does not happen — there is a permanent hole. Third, an engineer “fixes” the late-data problem by widening the window to range(start: -90m), and now consecutive runs overlap by 30 minutes and write two aggregates for the same hour with slightly different timestamps, inflating every dashboard that reads the rollup.

None of these raise an error. The task history shows every run succeeding. The fix is not one flag — it is understanding that the CQ engine’s implicit interval, backfill, and boundary alignment must all be re-expressed as explicit Flux, and that the read window and the write boundary have to be made deterministic so re-processing overwrites rather than duplicates. The rest of this page makes that concrete.

# Prerequisites

• A running InfluxDB 1.8+ source (to inventory and read the existing CQs) and an InfluxDB 2.7+ or 3.x target with the Flux task engine enabled.
• Flux 0.x (bundled with the 2.x/3.x target; no separate install).
• Destination buckets provisioned before the first task run, each with an explicit retention period — in 1.x the CQ inherited the target retention policy, but a 2.x task will happily write into a bucket whose lifecycle you never set. Sizing follows retention policy design.
• An operator or all-access token with read/write on the source and destination buckets and write on _tasks.
• For programmatic migration and validation: Python 3.9+ with influxdb-client 1.36+.
• A frozen inventory of every CQ (SHOW CONTINUOUS QUERIES) captured before any writes are redirected, so parity can be checked point-for-point.

# Core concept: what the CQ engine did implicitly

A continuous query bundled four behaviors that a task does not. Naming them is the whole migration:

1. Interval inference. The CQ derived its run cadence from GROUP BY time(1h). A task declares cadence separately in option task via every or cron, and declares the aggregation window separately in aggregateWindow(every: ...). These are two independent knobs in Flux; conflating them is the most common translation bug.
2. Automatic backfill. If the daemon was down, the CQ engine re-ran missed intervals against historical data when it came back. A task’s range() is relative to the run’s fire time by default, so a missed run is simply lost unless you either widen the window or replay explicitly.
3. Epoch-aligned boundaries. InfluxQL snapped windows to Unix-epoch multiples of the group duration. aggregateWindow also aligns to the epoch by default, but any location/timezone handling and any non-default offset must be stated, or globally distributed sensors land in subtly different buckets than they did under 1.x.
4. Silent late-data absorption within the window. A CQ re-evaluating an interval would sweep in whatever had arrived. A task reads its window once, at fire time plus offset, and moves on.

The migration is therefore a mapping exercise: every InfluxQL construct maps to an explicit Flux stage, and every implicit behavior maps to a deliberate parameter.

This decomposition aligns with the explicit, version-controlled model that the rest of the pipeline assumes: transformation logic written as Flux scripting for task automation, triggered by deterministic cron & interval scheduling logic, and staged through dependency mapping & DAG construction when one CQ has to become several ordered tasks.

# Step-by-step implementation

# 1. Inventory and semantically map every CQ

Extract the full definition set from the 1.x instance and record, for each query, the source measurement, the aggregate function(s), the grouping tags, the group duration, the destination, and the target retention policy. Do this before redirecting any writes so you have a frozen baseline to check parity against.

SHOW CONTINUOUS QUERIES

Build a mapping row per CQ using the table above. Pay special attention to grouping tags and to any CQ that used a RESAMPLE EVERY ... FOR ... clause — that clause encoded both a cadence and a backfill window, and it maps to a task cadence plus a widened read window, not to a single Flux stage.

# 2. Translate one CQ to a boundary-safe task

Rewrite the InfluxQL SELECT ... GROUP BY time() into a declarative Flux pipeline built from from(), range(), filter(), aggregateWindow(), and to(). The window below is the part most translations get wrong: it reads exactly one cadence, ending one offset in the past, so consecutive runs tile the timeline contiguously with neither overlap nor gap.

option task = {name: "temp_1h_downsample", every: 1h, offset: 10m}

from(bucket: "iot/long")
    |> range(start: -task.every - task.offset, stop: -task.offset)
    |> filter(fn: (r) => r._measurement == "temp" and r._field == "value")
    |> aggregateWindow(every: 1h, fn: mean, createEmpty: false)
    |> to(bucket: "temp_1h")

Three parameters carry the correctness of this task. offset: 10m delays execution ten minutes past the top of the hour so buffered gateway batches have landed before the window is read. The range(start: -task.every - task.offset, stop: -task.offset) spans exactly the hour that ended ten minutes ago, so run N and run N+1 meet at a shared boundary rather than overlapping or skipping. And createEmpty: false stops aggregateWindow from emitting null rows for sensors that reported nothing, which would otherwise inflate cardinality in the destination — a trade-off examined for sparse fleets in fallback chains for missing data.

# 3. Preserve grouping tags and multi-field CQs

A CQ with GROUP BY "host", time(1h) produced one series per host. Flux retains tags through the pipe automatically, but if you need the 1.x column shape (fields as columns rather than as _field/_value pairs) for a downstream consumer, pivot explicitly:

import "influxdata/influxdb/v1"

option task = {name: "temp_humidity_1h", every: 1h, offset: 10m}

from(bucket: "iot/long")
    |> range(start: -task.every - task.offset, stop: -task.offset)
    |> filter(fn: (r) => r._measurement == "environment")
    |> filter(fn: (r) => r._field == "temperature" or r._field == "humidity")
    |> aggregateWindow(every: 1h, fn: mean, createEmpty: false)
    |> v1.fieldsAsCols()
    |> to(bucket: "environment_1h")

If the numeric precision of the aggregate matters — for example when a CQ used mean on high-frequency sensor data whose rounding must stay stable across the rollup — align the task with precision mapping & rounding strategies before decommissioning the CQ, because Flux’s row-oriented arithmetic can differ at the last significant digit from the 1.x engine.

# 4. Replace automatic backfill with an explicit historical task

The CQ engine backfilled missed intervals; a scheduled task does not. To recover a gap, run a one-off task (or a manual run of the same script) whose range() covers the historical window explicitly. Snapping the aggregate to fixed boundaries makes this replay idempotent — re-running it overwrites the same series/timestamp keys instead of appending duplicates.

option task = {name: "temp_1h_backfill", every: 1h}

from(bucket: "iot/long")
    // Explicit historical window — replace with the gap you are recovering.
    |> range(start: 2026-06-01T00:00:00Z, stop: 2026-06-02T00:00:00Z)
    |> filter(fn: (r) => r._measurement == "temp" and r._field == "value")
    |> aggregateWindow(every: 1h, fn: mean, createEmpty: false)
    |> to(bucket: "temp_1h")

Delete the backfill task once it completes so it does not keep re-running against a frozen historical window.

# 5. Provision the migrated tasks as code

Create tasks through the client rather than the UI so definitions live in version control and can be diffed across environments. This is the on-ramp to the broader Python client orchestration patterns, including validating a run before it goes live.

import os
from influxdb_client import InfluxDBClient, TaskCreateRequest

client = InfluxDBClient(
    url=os.environ["INFLUX_URL"],
    token=os.environ["INFLUX_TOKEN"],
    org=os.environ["INFLUX_ORG"],
)

flux = """
option task = {name: "temp_1h_downsample", every: 1h, offset: 10m}

from(bucket: "iot/long")
    |> range(start: -task.every - task.offset, stop: -task.offset)
    |> filter(fn: (r) => r._measurement == "temp" and r._field == "value")
    |> aggregateWindow(every: 1h, fn: mean, createEmpty: false)
    |> to(bucket: "temp_1h")
"""

tasks_api = client.tasks_api()
created = tasks_api.create_task(
    task=TaskCreateRequest(
        org_id=os.environ["INFLUX_ORG_ID"],
        flux=flux,
        description="Migrated from 1.x cq_1h",
        status="active",
    )
)

# Trigger one manual run to validate before trusting the schedule.
run = tasks_api.run_manually(task_id=created.id)
print(f"Task {created.id} created; validation run {run.id} status {run.status}")

The end-to-end mechanics of a single CQ conversion — every InfluxQL clause and its Flux counterpart — are worked in detail in the companion walkthrough on migrating legacy continuous queries to InfluxDB 2.x tasks.

# Configuration reference

# Common failure modes and fixes

1. Cadence and window conflated into one number. Symptom: the migrated rollup has the right shape but wrong values — sums are too high or means are averaged over the wrong span. Root cause: the translator assumed GROUP BY time(1h) maps to a single Flux knob, so every and aggregateWindow(every:) were set inconsistently (e.g. every: 1h with aggregateWindow(every: 15m)). Fix: set the aggregation window to the CQ’s group duration and the task cadence independently.

// Wrong: cadence and window disagree
option task = {name: "roll", every: 1h}
// ... aggregateWindow(every: 15m, fn: mean)  <- produces 15m buckets, not hourly

// Right: window matches the old GROUP BY time(1h)
option task = {name: "roll", every: 1h, offset: 10m}
// ... aggregateWindow(every: 1h, fn: mean, createEmpty: false)

2. Late IoT data dropped because offset is zero. Symptom: the most recent window is systematically low compared to the 1.x output, only under real network conditions. Root cause: the task reads at the top of the period, before buffered gateway batches flush. Fix: measure the 99th-percentile arrival lag and set offset (and, if needed, widen the read window) to exceed it — the offset sizing logic lives in cron & interval scheduling logic.

3. Duplicate aggregates after widening the window. Symptom: dashboard values roughly double after someone extended range() to catch late data. Root cause: overlapping windows re-emit points that append instead of overwrite. Fix: snap emitted points to a deterministic boundary with aggregateWindow(every: 1h, ...) so re-processing overwrites the identical measurement/tag/field/timestamp key. Overlap is only safe when the write is idempotent.

4. Silent gaps where the CQ used to backfill. Symptom: holes in the rollup after any period the task engine was down or paused. Root cause: a scheduled task’s window is relative to fire time, so missed runs are simply not re-attempted. Fix: run an explicit historical task (Step 4) over the gap; do not rely on the schedule to self-heal.

5. Bucket has no retention, so raw and rolled data never expire. Symptom: storage grows without bound after cutover even though the old CQ target had a retention policy. Root cause: the destination bucket was created with the org default (often infinite) instead of inheriting a retention policy the way the 1.x target did. Fix: set an explicit retentionPeriod at provisioning, coordinated with the tiering model in bucket architecture & tiering boundaries.

# Verification and testing

Never decommission a CQ until the task’s output matches it point-for-point over an overlapping window. Run both engines in parallel for a validation period and compare counts and values in the destination series before cutting over.

Confirm the run itself succeeded and inspect schedule-to-start latency (the earliest sign that a heavy query is outlasting its interval) by reading the _tasks system bucket rather than trusting the UI’s green checkmark:

from(bucket: "_tasks")
    |> range(start: -24h)
    |> filter(fn: (r) => r._measurement == "runs")
    |> filter(fn: (r) => r.taskID == "TASK_ID_HERE")

Cross-check parity by comparing the aggregated point volume against the raw source over the same window — a straightforward count() on each side should reconcile once late-data offsets are accounted for:

from(bucket: "temp_1h")
    |> range(start: -24h)
    |> filter(fn: (r) => r._measurement == "temp")
    |> count()

Add a deadman check so a migrated task that stops producing output raises an alert instead of failing silently — the classic risk when replacing a CQ that no one had watched in years:

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

from(bucket: "temp_1h")
    |> range(start: -2h)
    |> filter(fn: (r) => r._measurement == "temp")
    |> monitor.deadman(t: experimental.subDuration(from: now(), d: 2h))
    |> filter(fn: (r) => r.dead == true)

From the CLI, confirm the migrated task exists and is active before redirecting production reads:

influx task list --org "$INFLUX_ORG"

# Integration points

Migration is the entry point into the 2.x pipeline, not the end of it. Once a CQ becomes a task, its trigger is governed by cron & interval scheduling logic and its transformation by Flux scripting for task automation — the same separation of when-versus-what that the rest of automated task scheduling & orchestration is built on. A single CQ that fed several downstream rollups usually becomes an ordered set of tasks whose sequencing is modeled with dependency mapping & DAG construction, rather than one script doing everything. The buckets the migrated tasks write into are governed by retention policy design, whose expiration windows must be at least as long as the slowest rollup’s coverage. And where the aggregate feeds anomaly detection or capacity alerts, the boundary between raw and rolled resolution is set by threshold tuning for aggregation.

# FAQ

# Does a task automatically backfill missed intervals like a continuous query did?

No. A continuous query re-ran missed intervals against historical data; a scheduled task’s range() is relative to its fire time, so a missed run is simply lost. Recover gaps by running an explicit historical task whose range() covers the missing window, and snap the aggregate to fixed boundaries so the replay overwrites rather than duplicates.

# How does GROUP BY time(1h) map to Flux?

It maps to the window argument of aggregateWindow(every: 1h, ...), not to the task cadence. The task’s every (or cron) is a separate, independent knob that controls how often the script runs. For a straight rollup they usually share the same duration, but conflating them is the most common migration bug.

# Will the migrated task produce identical values to my CQ?

Usually, but validate rather than assume. Flux’s row-oriented arithmetic can differ from the 1.x engine at the last significant digit, and window alignment or timezone handling can shift points between buckets. Run both engines in parallel over an overlapping window and reconcile counts and values before decommissioning the CQ.

# Does InfluxDB retry a failed task automatically?

The option task block does not expose per-task retries. Tasks emit structured run status (success, failed, canceled) that you can query, and retry logic is implemented externally — for example by polling the runs API and re-triggering failed runs from a Python orchestrator.

# My destination bucket keeps growing after migration — why?

A 1.x CQ inherited the target’s retention policy; a 2.x bucket does not inherit anything and defaults to the org’s retention (often infinite). Set an explicit retentionPeriod when you provision the destination bucket so rolled data ages out on the schedule you intend.

# Related

• Migrating legacy continuous queries to InfluxDB 2.x tasks — the full clause-by-clause conversion walkthrough.
• Cron & interval scheduling logic — sizing the cadence, offset, and window a migrated task fires on.
• Flux scripting for task automation — writing the replay-safe transformation the task runs.
• Precision mapping & rounding strategies — keeping aggregate arithmetic stable across the rollup.
• Retention policy design — replacing the 1.x target retention policy with explicit bucket lifecycles.

Up one level: Downsampling & Aggregation Pipeline Design