InfluxDB - Flux言語を用いたテーブルの結合

InfluxDB Python

時系列DBであるInfluxDB 2.0から採用されたFlux言語を用いたクエリの書き方を調べたので、メモをここに残しておく。

 

Flux言語の公式ドキュメント

Flux built-in functions | InfluxDB OSS 2.0 Documentation

 

 

data1 = from(bucket: "bucketname")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r["_measurement"] == "tablename")
  |> filter(fn: (r) => r["_field"] == "fieldname1")
  |> aggregateWindow(every: 1s, fn: last, createEmpty: false)
  |> keep(columns: ["_time", "_field", "_value"])
  
data2 = from(bucket: "bucketname")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r["_measurement"] == "tablename")
  |> filter(fn: (r) => r["_field"] == "fieldname2")
  |> aggregateWindow(every: 1s, fn: last, createEmpty: false)
  |> keep(columns: ["_time", "_field", "_value"])
  
join(
  tables: {d1: data1, d2: data2},
  on: ["_time"]
)

joinを用いることでテーブルの結合が可能となる。
"_time"を基準にテーブルの結合処理を行うが、どちらのテーブルの列名も"_field"、"_value"となっているため、それらの列名へ付与するsuffixを"d1"、"d2"と支持している。
この場合、結合後のテーブルの列は "_time", "_field_d1", "_field_d2", "_value_d1", "_value_d2"となる。