03-influxdb2时序数据库flux查询语言

## 一、Flux查询语言

Flux 可归类为一门新的编程语言，这里仅介绍针对查询需要的一些常用函数和操作，了解更多Flux见[Flux官方文档](https://links.jianshu.com/go?to=https%3A%2F%2Fdocs.influxdata.com%2Fflux%2Fv0.x%2F)

大多数Flux查询遵循如下几个步骤

- 指定数据源
- 过滤数据
- 对数据整形（重新组织数据结构）
- 数据内容计算

```jsx
//例
from(bucket: "example-bucket")            // ── Source
  |> range(start: -1d)                    // ── Filter on time
  |> filter(fn: (r) => r._field == "foo") // ── Filter on column values
  |> group(columns: ["sensorID"])         // ── Shape
  |> mean()                               // ── Process
```

flux查询语言看起来会像下面这样

```jsx
from(bucket: "wxm-influx")
  |> range(start: 0)
  |> filter(fn:(r)=>(r.owner=="wxm"))
  |> window(every: 1s)
  |> mean()
```

其中 `from`,`range`,`filter`等以下称作查询函数。flux查询语言允许使用`|>`分割符将上一个函数的输出传递给下一个函数，类似于linux的管道

### 基础规定

**每个查询语句必须包含数据源，时间区间和过滤器**。对应关键字分别为 `from`,`range`，`filter`。实际上不包含`filter`也是可以的，可以理解为`filter`过滤条件就是all

### Flux 标准库（部分）

该部分列举常用flux标准库函数，根据用法可举一反三使用所有标准库函数

#### buckets

buckets()返回当前组中所有的桶

#### from

指定从那个bucket查询数据

```csharp
from(bucket:"bucket名称")
```

#### range

指定时间区间，range接收两个参数，`start`和`stop` 其中`stop`可以省略，缺省值是当前时间。`start`不能省略，否则会报语法错误。

```
start和stop既可以使用相对的时间区间，也可以使用绝对的时间戳。相对时间区间格式为 时间间隔h/m/s例如 -1h,-5m。时间戳格式2021-01-01T00:00:00Z
```

```csharp
#示例1
from(bucket:"bucket名称")
  |> range(start: -1h)
#示例2
from(bucket:"bucket名称")
  |> range(start: 2021-01-01T00:00:00Z, stop: 2021-01-01T12:00:00Z)
```

#### filter

对数据进行过滤，接收一个判断函数。类似于java中的lamda,它看起来像是下面这样

```dart
(r) => (r.recordProperty comparisonOperator comparisonExpression)
```

判断函数返回一个boolean类型值。只有符合条件的记录才会被返回。

```csharp
#示例
from(bucket: "wxm-influx")
  |> range(start: -24h)
  |> filter(fn:(r)=>(r.owner=="wxm"))
#取最近24小时，含有owner标签且值为wxm的记录
```

当filter内含有多个条件时可用 `and` 或 `or` 连接各个条件

```csharp
#示例
from(bucket: "wxm-influx")
  |> range(start: -24h)
  |> filter(fn:(r)=>(r.owner=="wxm" or r.key=="indoor"))
#取最近24小时，含有owner标签且值为wx或含有标签key且值为indoor的记录
```

#### contains

**参数**

`value` 需要比较的值

`set`被比较的集合

判断`value`是否包含在`set`中。

```csharp
#示例
fiels = ["wxm","ali"]
from(bucket: "wxm-influx")
  |> range(start: 2022-03-12T08:14:04Z,stop:2022-03-13T08:19:05Z)
  |> filter(fn:(r)=>contains(value:r.owner,set:fiels))
```

#### window

window函数根据时间将数据分组.

> 参数

`every`:

every 指定每个时间窗口的时间，例如 every = 5 那么 时间窗口可以是 0m-5m，5m-10m，10m-15m。默认为 period 的值

```
period
```

period 明确在每个时间窗口中需要从时间窗口起始到多久的数据，例如时间窗口为10m-15m period = 3 则只取10m-13m的数据，13m-15m的数据会被抛弃。默认为 every 的值

```
offset
```

offset 指定每个时间窗口的时间偏移量，例如时间窗口为10m-15m offset = 3 那么会取 13m-15m的数据，10m-13m的数据会被抛弃，但注意当offset = every 时 offset不生效。且offset大于every 时 生效offset = offset % every

#### min

> 参数

```
column` 求最小值的列，默认为`_value
```

取每个分组中最小值

#### max

> 参数

```
column` 求最大值的列，默认为`_value
```

取每个分组最大值

#### mean

> 参数

```
column` 求平均值的列，默认为`_value
```

对每个分组的数据求平局值

#### bottom

> 参数

`columns` 按那些列进行排序

`n` 返回最上层 n 条数据

根据columns进行排序并返回最上层 n 条记录

#### count

> 参数

`columns` 按那列统计个数

统计数据个数

#### cumulativeSum

> 参数

`columns` 按那列进行累加

根据给定列对该列数据进行累加，模拟输出忽略 _start, _stop 等列:

原始数据：

| _measurement | _field |  62  |
| :----------- | :----: | :--: |
| temperature  |  wxm   |  62  |
| temperature  |  wxm   |  65  |
| temperature  |  wxm   | 81.5 |
| temperature  |  wxm   | 81.5 |

cumulativeSum函数执行后：

| _measurement | _field |  62   |
| :----------- | :----: | :---: |
| temperature  |  wxm   |  62   |
| temperature  |  wxm   |  127  |
| temperature  |  wxm   | 208.5 |
| temperature  |  wxm   |  290  |

#### group

> 参数

`columns` 按那些列进行分组操作 默认值为[]

`mode`:

mode 可以是以下值

"by",按 columns 内指定的列进行分组

"expect",按除了columns 列中指定的列进行分组

#### columns

> 参数

`column`存放列标签名的列。

返回所有列标签名并存储在指定的列中默认为`_value` 通过`column`参数指定

#### derivative

> 参数

`unit` 求多长时间内的变化速率。变化速率=`（下一个值-上一个值）/（下一个时间-上一个时间）* unit` 默认值1s

`nonNegative`变化速率是否可以是负值，如果是负数influxdb会假定前一个值为0 默认值true

`columns`指定计算变化速率的列 默认值 ["_value"]

`timeColumn`手动指定时间列 默认值 "_time"

#### difference

> 参数

`nonNegative`是否允许差值为负数，如果是负数influxdb会假定前一个值为0 默认值false

计算相邻两行的差值（next-pre）

`keepFirst`是否保留第一行，默认值为false

#### distinct

> 参数

`column` 指定按那列去重

按指定列去重