influxdb小试牛刀

发表于 | 分类于

1.8 版本不支持副本机制，目前开源的高可用方案 influxdb-proxy 是使用多个一致性 hash 环，并将数据分别写入每个 hash 环中的一个实例（双写）。如果数据写入失败则会写入缓存文件以便后续进行重写。

2.2 版本及以上（2022年03月发布）默认采用 Flux 语言，并支持副本机制，副本机制为异步复制，所以一定程度上无法保证数据不丢失。

下面操作如果没有权限则需要带上 token，例如

1
2
3
4
5
6
7
influx query --org iot --token 'GPJyXz8xOdSCpDItUfhoq939ZhRLGBRaZdSYocnCvhzGM06-PWISiN2BotpRu8lsQx54Quz3DeVDdjxLD_IeqA==' 'from(bucket: "event_message")
  |> range(start: -3d)
  |> filter(fn: (r) => r["_measurement"] == "event_message_10")
  |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")
  |> drop(columns: ["_start", "_stop"])
  |> group()
  |> limit(n: 1, offset: 0)'

认证

  1. 创建超级管理员用户
1
2
influx setup --org iot --bucket event_message \
    --username wjaiot --password wja123@iot --force
  1. 创建 org
1
influx org create --name iot --description "iot组织"
  1. API Token
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 查询用户token
docker exec influxdb2 influx auth list \
      --user wjaiot \
      --hide-headers | cut -f 3

# 创建针对 org 的 API Token，后续通过这个 API Token 操作这个组织的数据库
influx auth create \
  --all-access \
  --host http://localhost:8086 \
  --org <ORG> \
  --token <TOKEN>
  
  influx auth create \
  --all-access \
  --org iot

bucket

  1. 创建
1
2
influx bucket create --name example-bucket --retention 30d --shard-group-duration 1d \
  --description "bucket description"
  1. 删除
1
2
influx bucket delete --name example-bucket
influx bucket delete --id 06c86c40a9f36000
  1. 更新
1
2
influx bucket update --id 06c86c40a9f36000 --name new-bucket-name \
   --retention 90d --shard-group-duration 1d

write

1
2
3
4
5
6
7
8
9
influx write --bucket example-bucket "
m,host=host1 field1=1.2,field2=5i 1640995200000000000
m,host=host2 field1=2.4,field2=3i 1640995200000000000
"

influx write dryrun --bucket example-bucket "
m,host=host1 field1=1.2,field2=5i 1640995200000000000
m,host=host2 field1=2.4,field2=3i 1640995200000000000
"

query

1
2
3
4
5
6
7
8
9
influx query --org iot --token 'GPJyXz8xOdSCpDItUfhoq939ZhRLGBRaZdSYocnCvhzGM06-PWISiN2BotpRu8lsQx54Quz3DeVDdjxLD_IeqA==' 'from(bucket: "event_message")
  |> range(start: -3d)
  |> filter(fn: (r) => r["_measurement"] == "event_message_10")
  |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")
  |> drop(columns: ["_start", "_stop"])
  |> group()
  |> limit(n: 1, offset: 0)'

influx query 'from(bucket:"example-bucket") |> range(start:-1m)'
1
2
3
4
5
6
from(bucket: "telegraf/autogen")
  |> range(start: v.timeRangeStart)
  |> filter(fn: (r) => r._measurement == "Sessions Count")
  |> distinct(column: "value")
  |> map(fn: (r) => ({new_column: r._value, _value:r._value}))
  |> count()

副本配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 创建连接
influx remote create \
  --name myremote \
  --org-id <OSS org ID> \
  --token <OSS token> \
  --remote-url <remote URL> \
  --remote-api-token <remote token> \
  --remote-org-id <remote ord ID>
  
# 数据复制
influx replication create \
--name myreplication
--local-bucket example-local-bucket
--remote-bucket example-remote-bucket
--remote-id 0ooxX0xxXo0x

查看 series

1
influxd inspect report-db --db-path '/var/lib/influxdb2/engine/data' --detailed

参考文档

influxdb