Get started with InfluxDB tasks

An InfluxDB task is a scheduled Flux script that takes a stream of input data, modifies or analyzes it in some way, then writes the modified data back to InfluxDB or performs other actions.

This article walks through writing a basic InfluxDB task that downsamples data and stores it in a new bucket.

Components of a task

Every InfluxDB task needs the following components. Their form and order can vary, but they are all essential parts of a task.

• Task options
• A data source
• Data processing or transformation
• A destination

Skip to the full example task script

Define task options

Task options define specific information about the task. The example below illustrates how task options are defined in your Flux script:

option task = {name: "downsample_5m_precision", every: 1h, offset: 0m}

See Task configuration options for detailed information about each option.

Define a data source

1. Use from() to query data from InfluxDB . Use other Flux input functions to retrieve data from other sources.
2. Use range() to define the time range to return data from.
3. Use filter() to filter data based on column values.

data = from(bucket: "example-bucket")
    |> range(start: -task.every)
    |> filter(fn: (r) => r._measurement == "mem" and r.host == "myHost")

Process or transform your data

Tasks automatically process or transform data in some way at regular intervals. Data processing can include operations such as downsampling data, detecting anomalies, sending notifications, and more.

The task example below downsamples data by calculating the average of set intervals. It uses aggregateWindow() to group points into 5 minute windows and calculate the average of each window with mean().

option task = {name: "downsample_5m_precision", every: 1h, offset: 5m}

from(bucket: "example-bucket")
    |> range(start: -task.every)
    |> filter(fn: (r) => r._measurement == "mem" and r.host == "myHost")
    |> aggregateWindow(every: 5m, fn: mean)

See Common tasks for examples of tasks commonly used with InfluxDB.

Define a destination

In most cases, you’ll want to send and store data after the task has transformed it. The destination could be a separate InfluxDB measurement or bucket.

The example below uses to() to write the transformed data back to another InfluxDB bucket:

// ...
    |> to(bucket: "example-downsampled", org: "my-org")

To write data into InfluxDB, to() requires the following columns:

• _time
• _measurement
• _field
• _value

You can also write data to other destinations using Flux output functions.

Full example task script

Below is a task script that combines all of the components described above:

// Task options
option task = {name: "downsample_5m_precision", every: 1h, offset: 0m}

// Data source
from(bucket: "example-bucket")
    |> range(start: -task.every)
    |> filter(fn: (r) => r._measurement == "mem" and r.host == "myHost")
    // Data processing
    |> aggregateWindow(every: 5m, fn: mean)
    // Data destination
    |> to(bucket: "example-downsampled")

To learn more about InfluxDB tasks and how they work, watch the following video: