InfluxDB Flux

数据源#

Flux查询的第一步是指定数据源，可以使用from关键字来指定数据源。例如：

1
from(bucket: "my-bucket")

这个查询语句指定了数据源为名为my-bucket的桶。

过滤器#

Flux查询中可以使用过滤器来筛选出符合条件的数据。过滤器使用filter()函数来实现，可以传入一个Lambda表达式作为参数，该表达式用于对数据进行过滤。例如：

1
from(bucket: "my-bucket")
2
  |> filter(fn: (r) => r._measurement == "cpu")

这个查询语句指定了数据源为my-bucket桶，并使用filter()函数来筛选出_measurement字段等于cpu的数据。

转换器#

Flux查询中可以使用转换器对数据进行转换和处理。转换器使用map()函数来实现，可以传入一个Lambda表达式作为参数，该表达式用于对数据进行转换。例如：

1
from(bucket: "my-bucket")
2
  |> filter(fn: (r) => r._measurement == "cpu")
3
  |> map(fn: (r) => ({ r with _value: r._value * 2.0 }))

这个查询语句指定了数据源为my-bucket桶，并使用filter()函数筛选出_measurement字段等于cpu的数据，然后使用map()函数将_value字段的值乘以2.0。

聚合器#

Flux查询中可以使用聚合器对数据进行聚合计算。聚合器使用aggregateWindow()函数来实现，可以传入一个Lambda表达式作为参数，该表达式用于指定聚合计算的方式。例如：

1
from(bucket: "my-bucket")
2
  |> filter(fn: (r) => r._measurement == "cpu")
3
  |> aggregateWindow(every: 1m, fn: mean)

这个查询语句指定了数据源为my-bucket桶，并使用filter()函数筛选出_measurement字段等于cpu的数据，然后使用aggregateWindow()函数对数据进行聚合计算，每分钟计算一次平均值。

排序器#

Flux查询中可以使用排序器对数据进行排序。排序器使用sort()函数来实现，可以传入一个Lambda表达式作为参数，该表达式用于指定排序方式。例如：

1
from(bucket: "my-bucket")
2
  |> filter(fn: (r) => r._measurement == "cpu")
3
  |> sort(columns: ["_time"], desc: true)

这个查询语句指定了数据源为my-bucket桶，并使用filter()函数筛选出_measurement字段等于cpu的数据，然后使用sort()函数对数据按照_time字段进行降序排序。

数学操作符#

Flux语法支持常见的数学操作符，例如加减乘除和取模等。例如：

1
x + y
2
x - y
3
x * y
4
x / y
5
x % y

逻辑操作符#

Flux语法支持常见的逻辑操作符，例如与、或和非等。例如：

1
x and y
2
x or y
3
not x

比较操作符#

Flux语法支持常见的比较操作符，例如等于、不等于、大于、小于、大于等于和小于等于等。例如：

1
x == y
2
x != y
3
x > y
4
x < y
5
x >= y
6
x <= y

字符串操作符#

Flux语法支持常见的字符串操作符，例如连接和分割等。例如：

1
x + y
2
strings.split(v: "x,y,z", t: ",")

时间操作符#

Flux语法支持常见的时间操作符，例如时间加减和时间格式化等。例如：

1
time + duration(v: 1h)
2
time - duration(v: 1h)
3
time |> formatTime(fmt: "2006-01-02T15:04:05Z07:00")

使用range()函数限制查询时间范围#

在查询数据时，我们通常只需要查询指定时间范围内的数据。Flux语法中可以使用range()函数来限制查询时间范围。例如：

1
from(bucket: "my-bucket")
2
  |> range(start: -1h, stop: now())

这个查询语句指定了数据源为my-bucket桶，并使用range()函数限制查询时间范围为最近一小时内的数据。

使用filter()函数筛选数据#

在查询数据时，我们通常需要筛选出符合条件的数据。Flux语法中可以使用filter()函数来筛选数据。例如：

1
from(bucket: "my-bucket")
2
  |> range(start: -1h, stop: now())
3
  |> filter(fn: (r) => r._measurement == "cpu")

这个查询语句指定了数据源为my-bucket桶，并使用range()函数限制查询时间范围为最近一小时内的数据，然后使用filter()函数筛选出_measurement字段等于cpu的数据。

使用map()函数对数据进行转换#

在查询数据时，我们通常需要对数据进行转换和处理。Flux语法中可以使用map()函数对数据进行转换。例如：

1
from(bucket: "my-bucket")
2
  |> range(start: -1h, stop: now())
3
  |> filter(fn: (r) => r._measurement == "cpu")
4
  |> map(fn: (r) => ({ r with _value: r._value * 2.0 }))

这个查询语句指定了数据源为my-bucket桶，并使用range()函数限制查询时间范围为最近一小时内的数据，然后使用filter()函数筛选出_measurement字段等于cpu的数据，最后使用map()函数将_value字段的值乘以2.0。

使用aggregateWindow()函数对数据进行聚合计算#

在查询数据时，我们通常需要对数据进行聚合计算。Flux语法中可以使用aggregateWindow()函数对数据进行聚合计算。例如：

1
from(bucket: "my-bucket")
2
  |> range(start: -1h, stop: now())
3
  |> filter(fn: (r) => r._measurement == "cpu")
4
  |> aggregateWindow(every: 1m, fn: mean)

这个查询语句指定了数据源为my-bucket桶，并使用range()函数限制查询时间范围为最近一小时内的数据，然后使用filter()函数筛选出_measurement字段等于cpu的数据，最后使用aggregateWindow()函数对数据进行聚合计算，每分钟计算一次平均值。

使用group()函数对数据进行分组#

在查询数据时，我们通常需要对数据进行分组。Flux语法中可以使用group()函数对数据进行分组。例如：

1
from(bucket: "my-bucket")
2
  |> range(start: -1h, stop: now())
3
  |> filter(fn: (r) => r._measurement == "cpu")
4
  |> aggregateWindow(every: 1m, fn: mean)
5
  |> group(columns: ["_field"])

这个查询语句指定了数据源为my-bucket桶，并使用range()函数限制查询时间范围为最近一小时内的数据，然后使用filter()函数筛选出_measurement字段等于cpu的数据，最后使用aggregateWindow()函数对数据进行聚合计算，每分钟计算一次平均值，最后使用group()函数对数据按照_field字段进行分组。

使用sort()函数对数据进行排序#

在查询数据时，我们通常需要对数据进行排序。Flux语法中可以使用sort()函数对数据进行排序。例如：

1
from(bucket: "my-bucket")
2
  |> range(start: -1h, stop: now())
3
  |> filter(fn: (r) => r._measurement == "cpu")
4
  |> aggregateWindow(every: 1m, fn: mean)
5
  |> sort(columns: ["_value"], desc: true)

这个查询语句指定了数据源为my-bucket桶，并使用range()函数限制查询时间范围为最近一小时内的数据，然后使用filter()函数筛选出_measurement字段等于cpu的数据，最后使用aggregateWindow()函数对数据进行聚合计算，每分钟计算一次平均值，最后使用sort()函数对数据按照_value字段进行降序排序。

使用join()函数连接多个数据源#

在查询数据时，我们通常需要连接多个数据源。Flux语法中可以使用join()函数连接多个数据源。例如：

1
from(bucket: "my-bucket")
2
|> range(start: -1h, stop: now())
3
|> filter(fn: (r) => r._measurement == "cpu")
4
|> aggregateWindow(every: 1m, fn: mean)
5
|> join(tables: {
6
other_table: from(bucket: "my-bucket-2")
7
|> range(start: -1h, stop: now())
8
|> filter(fn: (r) => r._measurement == "memory")
9
|> aggregateWindow(every: 1m, fn: mean)
10
}, on:["_time"])

这个查询语句指定了数据源为my-bucket桶，并使用range()函数限制查询时间范围为最近一小时内的数据，然后使用filter()函数筛选出_measurement字段等于cpu的数据，最后使用aggregateWindow()函数对数据进行聚合计算，每分钟计算一次平均值。同时，使用join()函数连接了另一个数据源my-bucket-2，并对其进行类似的处理，然后按照_time字段进行连接。

以上是一些Flux的常用函数用法示例，希望能够对读者理解和使用Flux语法有所帮助。