Get started processing data

map() takes a single parameter, fn. fn takes an anonymous function that reads each row as a record named r. In the r record, each key-value pair represents a column and its value. For example:

r = {
    _time: 2020-01-01T00:00:00Z,
    _measurement: "home",
    room: "Kitchen",
    _field: "temp",
    _value: 21.0,
}

The fn function modifies the r record in any way you need and returns a new record for the row. For example, using the record above:

(r) => ({ _time: r._time, _field: "temp_F", _value: (r._value * 1.8) + 32.0})

// Returns: {_time: 2020-01-01T00:00:00Z, _field: "temp_F", _value: 69.8}

Notice that some of the columns were dropped from the original row record. This is because the fn function explicitly mapped the _time, _field, and _value columns. To retain existing columns and only update or add specific columns, use the with operator to extend your row record. For example, using the record above:

(r) => ({r with _value: (r._value * 1.8) + 32.0, degrees: "F"})

// Returns:
// {
//     _time: 2020-01-01T00:00:00Z,
//     _measurement: "home",
//     room: "Kitchen",
//     _field: "temp",
//     _value: 69.8,
//     degrees: "F",
// }

map() lets your perform mathematical operations on your data. For example, using the data written in “Get started writing to InfluxDB”:

1. Query the temp field to return room temperatures in °C.
2. Use map() to iterate over each row and convert the °C temperatures in the _value column to °F using the equation: °F = (°C * 1.8) + 32.0.

from(bucket: "get-started")
    |> range(start: 2022-01-01T14:00:00Z, stop: 2022-01-01T20:00:01Z)
    |> filter(fn: (r) => r._measurement == "home")
    |> filter(fn: (r) => r._field == "temp")
    |> map(fn: (r) => ({r with _value: (r._value * 1.8) + 32.0}))

Within a map() function, you can use conditional expressions (if/then/else) to conditionally assign values. For example, using the data written in “Get started writing to InfluxDB”:

1. Query the co field to return carbon monoxide parts per million (ppm) readings in each room.

2. Use map() to iterate over each row, evaluate the value in the _value column, and then conditionally assign a state:

   • If the carbon monoxide is less than 10 ppm, assign the state: ok.
   • Otherwise, assign the state: warning.

   Store the state in a state column.

from(bucket: "get-started")
    |> range(start: 2022-01-01T14:00:00Z, stop: 2022-01-01T20:00:01Z)
    |> filter(fn: (r) => r._measurement == "home")
    |> filter(fn: (r) => r._field == "co")
    |> map(fn: (r) => ({r with state: if r._value < 10 then "ok" else "warning"}))

map() lets you execute more complex operations on a per row basis. Using a Flux block ({}) in the fn function, you can create scoped variables and execute other functions within the context of each row. For example, you can send a message to Slack.

For example, using the data written in “Get started writing to InfluxDB”:

1. Import the slack package.

2. Query the co field to return carbon monoxide parts per million (ppm) readings in each room.

3. Use map() to iterate over each row, evaluate the value in the _value column, and then conditionally assign a state:

   • If the carbon monoxide is less than 10 ppm, assign the state: ok.
   • Otherwise, assign the state: warning.

   Store the state in a state column.

4. Use filter() to return only rows with warning in the state column.

5. Use map() to iterate over each row. In your fn function, use a Flux block ({}) to:

   1. Create a responseCode variable that uses slack.message() to send a message to Slack using data from the input row. slack.message() returns the response code of the Slack API request as an integer.
   2. Use a return statement to return a new row record. The new row should extend the input row with a new column, sent, with a boolean value determined by the responseCode variable.

map() sends a message to Slack for each row piped forward into the function.

import "slack"

from(bucket: "get-started")
    |> range(start: 2022-01-01T14:00:00Z, stop: 2022-01-01T20:00:01Z)
    |> filter(fn: (r) => r._measurement == "home")
    |> filter(fn: (r) => r._field == "co")
    |> map(fn: (r) => ({r with state: if r._value < 10 then "ok" else "warning"}))
    |> filter(fn: (r) => r.state == "warning")
    |> map(
        fn: (r) => {
            responseCode =
                slack.message(
                    token: "mYSlacK70k3n",
                    color: "#ff0000",
                    channel: "#alerts",
                    text: "Carbon monoxide is at dangerous levels in the ${r.room}: ${r._value} ppm.",
                )

            return {r with sent: responseCode == 200}
        },
    )

With the results above, you would receive the following messages in Slack:

Carbon monoxide is at dangerous levels in the Kitchen: 18 ppm.
Carbon monoxide is at dangerous levels in the Kitchen: 22 ppm.
Carbon monoxide is at dangerous levels in the Living Room: 14 ppm.
Carbon monoxide is at dangerous levels in the Kitchen: 26 ppm.
Carbon monoxide is at dangerous levels in the Living Room: 17 ppm.

Using the data written in “Get started writing to InfluxDB”:

1. Query the temp and hum fields.
2. Use group() to group by only the _field column.

from(bucket: "get-started")
    |> range(start: 2022-01-01T08:00:00Z, stop: 2022-01-01T10:00:01Z)
    |> filter(fn: (r) => r._measurement == "home")
    |> filter(fn: (r) => r._field == "temp" or r._field == "hum")
    |> group(columns: ["_field"])

Using the data written in “Get started writing to InfluxDB”:

1. Query the temp and hum fields.
2. Use group() without any parameters to “ungroup” data or group by no columns. The default value of the columns parameter is an empty array ([]).

from(bucket: "get-started")
    |> range(start: 2022-01-01T08:00:00Z, stop: 2022-01-01T10:00:01Z)
    |> filter(fn: (r) => r._measurement == "home")
    |> filter(fn: (r) => r._field == "temp" or r._field == "hum")
    |> group()

Using the data written in “Get started writing to InfluxDB”:

1. Query the temp field. By default, from() returns the data grouped by _measurement, room and _field, so each table represents a room.
2. Use mean() to return the average temperature from each room.

from(bucket: "get-started")
    |> range(start: 2022-01-01T14:00:00Z, stop: 2022-01-01T20:00:01Z)
    |> filter(fn: (r) => r._measurement == "home")
    |> filter(fn: (r) => r._field == "temp")
    |> mean()

Using the data written in “Get started writing to InfluxDB”:

1. Query the temp field.
2. Use group() to ungroup the data into a single table. By default, from() returns the data grouped by_measurement, room and _field. To get the overall average, you need to structure all results as a single table.
3. Use mean() to return the average temperature.

from(bucket: "get-started")
    |> range(start: 2022-01-01T14:00:00Z, stop: 2022-01-01T20:00:01Z)
    |> filter(fn: (r) => r._measurement == "home")
    |> filter(fn: (r) => r._field == "temp")
    |> group()
    |> mean()

Using the data written in “Get started writing to InfluxDB”:

1. Query all fields by simply filtering by the home measurement.
2. The fields in the home measurement are different types. Use toFloat() to cast all field values to floats.
3. Use group() to group the data by room.
4. Use count() to return the number of rows in each input table.

from(bucket: "get-started")
    |> range(start: 2022-01-01T14:00:00Z, stop: 2022-01-01T20:00:01Z)
    |> filter(fn: (r) => r._measurement == "home")
    |> toFloat()
    |> group(columns: ["room"])
    |> count()

Output

Using the data written in “Get started writing to InfluxDB”:

1. Query the temp field.
2. Use first() to return the first row from each table.

from(bucket: "get-started")
    |> range(start: 2022-01-01T14:00:00Z, stop: 2022-01-01T20:00:01Z)
    |> filter(fn: (r) => r._measurement == "home")
    |> filter(fn: (r) => r._field == "temp")
    |> first()

Using the data written in “Get started writing to InfluxDB”:

1. Query the temp field.
2. Use last() to return the last row from each table.

from(bucket: "get-started")
    |> range(start: 2022-01-01T14:00:00Z, stop: 2022-01-01T20:00:01Z)
    |> filter(fn: (r) => r._measurement == "home")
    |> filter(fn: (r) => r._field == "temp")
    |> last()

Using the data written in “Get started writing to InfluxDB”:

1. Query the temp field.
2. Use max() to return the row with the highest value in the _value column from each table.

from(bucket: "get-started")
    |> range(start: 2022-01-01T14:00:00Z, stop: 2022-01-01T20:00:01Z)
    |> filter(fn: (r) => r._measurement == "home")
    |> filter(fn: (r) => r._field == "temp")
    |> max()