Query data with Flux

The following guides walk through both common and complex queries and use cases for Flux.

Example data variable

Many of the examples provided in the following guides use a data variable, which represents a basic query that filters data by measurement and field. data is defined as:

data = from(bucket: "example-bucket")
    |> range(start: -1h)
    |> filter(fn: (r) => r._measurement == "example-measurement" and r._field == "example-field")

Flux query guides

• Query fields and tags
• Group
• Sort and limit
• Window & aggregate
• Explore your schema
• Transform data with math
• Calculate percentages
• Increase
• Moving Average
• Rate
• Histograms
• Fill
• Median
• Percentile & quantile
• Join
• Cumulative sum
• First and last
• Exists
• Custom functions
• Extract scalar values
• Monitor states
• Operate on timestamps
• Query SQL data
• Conditional logic
• Regular expressions
• Geo-temporal data
• Query the Flux version

Query fields and tags

Use filter() to query data based on fields, tags, or any other column value. filter() performs operations similar to the SELECT statement and the WHERE clause in InfluxQL and other SQL-like query languages.

from(bucket: "example-bucket")
    |> range(start: -1h)
    |> filter(fn: (r) => r._measurement == "example-measurement" and r.tag == "example-tag")
    |> filter(fn: (r) => r._field == "example-field")

Read more

Group

Use group() to group data with common values in specific columns.

data
    |> group(columns: ["host"], mode: "by")

Input:

Output:

Read more

Sort and limit

Use sort() to order records within each table by specific columns and limit() to limit the number of records in output tables to a fixed number, n.

data
    |> sort(columns: ["host", "_value"])
    |> limit(n: 4)

Input:

Output:

Read more

Window & aggregate

This guide walks through windowing and aggregating data with Flux and outlines how it shapes your data in the process.

data
    |> aggregateWindow(every: 20m, fn: mean)

Input:

Output:

Read more

Explore your schema

Flux provides functions that let you explore the structure and schema of your data stored in InfluxDB.

import "influxdata/influxdb/schema"
// List buckets
buckets()
// List measurements
schema.measurements(bucket: "example-bucket")
// List field keys
schema.fieldKeys(bucket: "example-bucket")
// List tag keys
schema.tagKeys(bucket: "example-bucket")
// List tag values
schema.tagValues(bucket: "example-bucket", tag: "example-tag")

Read more

Transform data with math

Use map() to remap column values and apply mathematic operations.

data
    |> map(fn: (r) => ({ r with _value: r._value * r._value }))

Input:

Output:

Read more

Calculate percentages

Use pivot() or join() and the map() function to align operand values into rows and calculate a percentage.

data
  |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")
  |> map(
      fn: (r) => ({
          _time: r._time,
          _field: "used_percent",
          _value: float(v: r.used) / float(v: r.total) * 100.0,
      }),
  )

Input:

Output:

Read more

Increase

Use increase() to track increases across multiple columns in a table. This function is especially useful when tracking changes in counter values that wrap over time or periodically reset.

data
    |> increase()

Input:

Output:

Read more

Moving Average

Use movingAverage() or timedMovingAverage() to return the moving average of data.

data
    |> movingAverage(n: 3)

Input:

Output:

data
    |> timedMovingAverage(every: 2m, period: 4m)

Input:

Output:

Read more

Rate

Use derivative() to calculate the rate of change between subsequent values or aggregate.rate() to calculate the average rate of change per window of time. If time between points varies, these functions normalize points to a common time interval making values easily comparable.

data
    |> derivative(unit: 1m, nonNegative: true)

Input:

Output:

import "experimental/aggregate"
data
    |> aggregate.rate(every: 20m, unit: 1m)

Input:

Output:

Read more

Histograms

Use histogram() to create cumulative histograms with Flux.

data
    |> histogram(
        column: "_value",
        upperBoundColumn: "le",
        countColumn: "_value",
        bins: [100.0, 200.0, 300.0, 400.0],
    )

Input:

Output:

Read more

Fill

Use fill() function to replace null values.

data
    |> fill(usePrevious: true)

Input:

Output:

data
    |> fill(value: 0.0)

Input:

Output:

Read more

Median

Use median() to return a value representing the 0.5 quantile (50th percentile) or median of input data.

data
    |> median()

Input:

Output:

Read more

Percentile & quantile

Use the quantile() function to return all values within the q quantile or percentile of input data.

data
    |> quantile(q: 0.99, method: "estimate_tdigest")

Input:

Output:

Read more

Join

This guide walks through joining data with Flux and outlines how it shapes your data in the process.

import "join"
import "sql"
left =
    from(bucket: "example-bucket-1")
        |> range(start: "-1h")
        |> filter(fn: (r) => r._measurement == "example-m")
        |> filter(fn: (r) => r._field == "example-f")
        |> drop(columns: ["_measurement", "_field"])
right =
    sql.from(
        driverName: "postgres",
        dataSourceName: "postgresql://username:password@localhost:5432",
        query: "SELECT * FROM example_table",
    )
join.inner(
    left: left |> group(),
    right: right,
    on: (l, r) => l.sensorID == r.ID,
    as: (l, r) => ({l with expired: r.expired}),
)
    |> group(columns: ["_time", "_value"], mode: "except")

Input:

left

right

Output:

Read more

Cumulative sum

Use the cumulativeSum() function to calculate a running total of values.

data
    |> cumulativeSum()

Input:

Output:

Read more

First and last

Use first() or last() to return the first or last point in an input table.

data
    |> first()

Input:

Output:

data
    |> last()

Input:

Output:

Read more

Exists

Use the Flux exists operator to check if a row record contains a column or if that column’s value is null.

Filter null values

data
  |> filter(fn: (r) => exists r._value)

Read more

Custom functions

Create your own custom Flux functions to transform and operate on data.

multByX = (tables=<-, x) => tables
    |> map(fn: (r) => ({r with _value: r._value * x}))
data
    |> multByX(x: 2.0)

Read more

Extract scalar values

Use Flux dynamic query functions to extract scalar values from Flux query output. This lets you, for example, dynamically set variables using query results.

scalarValue = (tables=<-) => {
    _record = tables
        |> findRecord(fn: (key) => true, idx: 0)
    return _record._value
}

Read more

Monitor states

Flux provides several functions to help monitor states and state changes in your data.

Operate on timestamps

Use Flux to process and operate on timestamps.

Query SQL data

The Flux sql package provides functions for working with SQL data sources. Use sql.from() to query SQL databases like PostgreSQL, MySQL, Snowflake, SQLite, Microsoft SQL Server, Amazon Athena, and Google BigQuery.

import "sql"
sql.from(
    driverName: "postgres",
    dataSourceName: "postgresql://user:password@localhost",
    query: "SELECT * FROM example_table",
)

Read more

Conditional logic

This guide describes how to use Flux conditional expressions, such as if, else, and then, to query and transform data. Flux evaluates statements from left to right and stops evaluating once a condition matches.

if color == "green" then "008000" else "ffffff"

Read more

Regular expressions

This guide walks through using regular expressions in evaluation logic in Flux functions.

data
    |> filter(fn: (r) => r.tag =~ /^foo[1-3]/)

Input:

Output:

Read more

Geo-temporal data

Use the Flux Geo package to filter geo-temporal data and group by geographic location or track.

import "experimental/geo"
sampleGeoData
    |> geo.filterRows(region: {lat: 30.04, lon: 31.23, radius: 200.0})
    |> geo.groupByArea(newColumn: "geoArea", level: 5)

Read more

Query the Flux version

Use runtime.version() to return the version of Flux installed in InfluxDB.

import "array"
import "runtime"
array.from(rows: [{version: runtime.version()}])

Read more

flux query