【InfluxDB】クエリ言語Fluxについての覚書
概要
InfluxDB 2系における時系列データの取得にはFluxというクエリ言語が使用されます。Flux は関数型のパラダイムが入った言語であり、SQLを主に触ってきた身からすると取っ付きづらさがありました。最近はFluxにも慣れてきたのでなんとなくの使い方をメモしておきます。
Fluxの基本
from, range, filterの3つの関数を使用してデータを取り出すのがFluxを使用する時の基本の型となります。
from(bucket: "bucket-name")
|> range(start: -1y)
|> filter(fn: (r) => r.tag == "t1")
このとき、fromで取得したデータはstreamという型を持ちます。streamは異なるgroupKeyを持つテーブルの集合です。次の図のようなイメージをしておくと理解しやすいと思います。

データ整形は、from, range, filterを使用して取得したstreamに対して、さまざまな関数をパイプで繋げていくことで行われます。次の例はstreamに対してsum関数を適用した例です。(公式ドキュメント内の例より抜粋)
import "sampledata"
sampledata.int()
|> sum()
BEFORE
stream内に以下の2つのtableがあるとします。
| _time | _value | tag |
|---|---|---|
| 2021-01-01T00:00:10Z | 10 | t1 |
| 2021-01-01T00:00:00Z | -2 | t1 |
| 2021-01-01T00:00:20Z | 7 | t1 |
| 2021-01-01T00:00:30Z | 17 | t1 |
| 2021-01-01T00:00:40Z | 15 | t1 |
| 2021-01-01T00:00:50Z | 4 | t1 |
| _time | _value | tag |
|---|---|---|
| 2021-01-01T00:00:00Z | 19 | t2 |
| 2021-01-01T00:00:10Z | 4 | t2 |
| 2021-01-01T00:00:20Z | -3 | t2 |
| 2021-01-01T00:00:30Z | 19 | t2 |
| 2021-01-01T00:00:40Z | 13 | t2 |
| 2021-01-01T00:00:50Z | 1 | t2 |
AFTER
2つの1レコードのテーブルを持つstreamが出力されます。
| tag | _value |
|---|---|
| t1 | 51 |
| tag | _value |
|---|---|
| t2 | 53 |
各関数の分類
Fluxの理解度を深めるにはそれぞれの関数のstreamに対する作用を分類しておくことが有効です。個人的に、以下のような分類が有用だと考えました。
※本記事で取り上げている関数はFluxのstreamを受け取りstreamを返す関数の中の、ある程度シンプルまたは使用頻度の高そうな一部であり、網羅性はありません。
A. (原則として)stream内のtableの数を変化させないもの
- 各tableを一行にして統計量を取り出すもの(sumなど)
- 各tableのcolumnに作用するもの(renameなど)
- 各tableのそれぞれのrecordに作用するもの(mapなど)
- 各tableのrecordを抽出するもの(firstなど)
- 各tableのrecordにreducer的に作用するもの(reduceなど)
- 金融などで使用される移動平均系(movingAverageなど)
- その他
B. stream内のtableの数を変化させるもの
1. group
2. pivot
A. stream内のtableの数を変化させないもの
1. 各tableを一行にして統計量を取り出すもの

| 関数名 | 詳細 |
|---|---|
| covariance | table内の2つのカラムの共分散を取得する |
| max | tableごとに最大値を取得する |
| mean | tableごとに平均値を取得する |
| median | tableごとに中央値(median)を取得する |
| min | tableごとに最小値を取得する |
| mode | tableごとに最頻値(mode)を取得する |
| quantile | tableごとに四分位数を取得する |
| skew | tableごとに歪度を取得する |
| spread | tableごとに最大値と最小値の差を取得する |
| stddev | tableごとに標準偏差を取得する |
| sum | tableごとに合計値を取得する |
| timeWeightedAvg | tableごとに時間で重み付けされた平均を取得する |
2.各tableのcolumnに作用するもの

| 関数名 | 詳細 |
|---|---|
| drop | 特定のカラムを削除する |
| duplicate | カラムを複製する |
| keep | 指定のカラムだけを残し、その他のカラムを削除する |
| rename | カラムの名前を変更する |
| set | 均一な初期値を使用して新しいカラムを追加する |
3. 各tableのそれぞれのrecordに作用するもの

| 関数名 | 詳細 |
|---|---|
| fill | recordの_valueがnullとなっている場合、指定した値で埋める |
| map | 各recordを関数を用いて変換する |
| timeshift | 各recordの_timeに加減算を行う |
4. 各tableのrecordを抽出するもの

| 関数名 | 詳細 |
|---|---|
| first | 各tableの最初のnullでないrecordを返す |
| last | 各tableの最後のnullでないrecordを返す |
| limit | 各tableの先頭からN個のレコードを返す |
| tail | 各tableの末尾からN個のレコードを返す |
| sample | 各tableから飛び飛びのN個のレコードを返す |
5. 各tableにreducer的に作用するもの

| 関数名 | 詳細 |
|---|---|
| reduce | 各tableにreducer関数を作用させる |
6. 金融などで使用される移動平均系
とくに、金融系でよく使われそうな関数として、
| 関数名 | 詳細 |
|---|---|
| doubleEMA | DEMAと呼ばれる指標を計算します |
| tripleEMA | TEMAと呼ばれる指標を計算します |
| exponentialMovingAverage | EMAと呼ばれる指標を計算します |
| movingAverage | 移動平均を計算します(SMA) |
| kaufmansAMA | KAMAと呼ばれる指標を計算します |
| kaufmansER | KERと呼ばれる指標を計算します |
| tripleExponentialDerivative | TRIXと呼ばれる指標を計算します |
| relativeStrengthIndex | RSIと呼ばれる指標を計算します |
B. stream内のtableの数を変化させるもの
1. group
group関数では、groupKeyを指定し直すことによりstream内のテーブルを分割・結合を行えます。groupKeyの指定の仕方次第で、下の2つの図のようにテーブルの数を増やすことも減らすこともできます。


| 関数名 | 詳細 |
|---|---|
| group | 指定したcolumnをgroupKeyとして、stream内のtableを再分割します。 |
2. pivot

| 関数名 | 詳細 |
|---|---|
| pivot | rowやcolumn、tagなどを入れ替える |
Discussion