时序性数据库InfluxDB初探与优化

简介

InfluxDB是一个由InfluxData开发的开源时序型数据库。它由Go写成，着力于高性能地查询与存储时序型数据。时序型数据是指那些随时间变化而变化的数据，比如温度，湿度，股票价格，网站访问量等。这些数据通常具有高频率，高精度，高复杂度和高价值的特点。InfluxDB可以帮助我们有效地管理和分析这些数据，为我们提供实时的洞察和预测。

为了让你更好地理解InfluxDB的优势和特色，我用一个表格来比较它和MySQL等传统关系型数据库的区别：

特点	InfluxDB	MySQL
数据模型	基于时间序列的键值对	基于表格的行列
数据结构	灵活，无需预定义模式	固定，需要预先定义模式
数据类型	主要支持数值类型	支持多种类型
查询语言	使用Flux语言，专为时序型数据设计	使用SQL语言，通用但不针对时序型数据
写入性能	高效，支持批量写入和压缩存储	一般，需要事务处理和索引维护
查询性能	快速，支持多维聚合和窗口函数	慢速，需要多表连接和分组排序
可扩展性	易于水平扩展和分布式部署	难以水平扩展和分布式部署

从上表可以看出，InfluxDB比MySQL更适合处理时序型数据，因为它有更高的性能，更灵活的数据模型，更强大的查询语言和更好的可扩展性。

版本

官网: InfluxDB OSS 2.3 Documentation (influxdata.com)

如果你想更深入地了解InfluxDB的功能和用法，我强烈建议你去多看官方文档。官方文档是InfluxDB的最权威和最全面的资料来源，它包含了InfluxDB的安装，配置，管理，查询，分析，可视化等方面的详细说明和示例。官方文档还提供了一些常见问题和解决方案，以及一些最佳实践和技巧。官方文档是你学习和使用InfluxDB的最好的伙伴。

安装

配置文件

在InfluxDB中,配置文件默认是不存在的,需要执行以下命令进行生成,流入数据库配置选项 |InfluxDB OSS 2.3 文档 (influxdata.com)

docker run --rm influxdb:2.3.0 influx server-config > config.yml

具体内容如下:

{
        "assets-path": "",
        "bolt-path": "/var/lib/influxdb2/influxd.bolt",
        "e2e-testing": false,
        "engine-path": "/var/lib/influxdb2/engine",
        "feature-flags": null,
        "flux-log-enabled": false,
        "hardening-enabled": false,
        "http-bind-address": ":8086",
        "http-idle-timeout": 180000000000,
        "http-read-header-timeout": 10000000000,  #服务器应尝试读取新请求的 HTTP 标头的最长持续时间。设置为 0 表示没有超时
        "http-read-timeout": 0, # 读超时
        "http-write-timeout": 0, # 写超时 
        "influxql-max-select-buckets": 0,
        "influxql-max-select-point": 0,
        "influxql-max-select-series": 0,
        "instance-id": "",
        "log-level": "info", # 日志输出级别
        "metrics-disabled": false,
        "nats-max-payload-bytes": 0,
        "nats-port": 4222,
        "no-tasks": false,
        "pprof-disabled": false,
        "query-concurrency": 1024,  # 允许并发执行的查询数。
        "query-initial-memory-bytes": 0,  #  为查询分配的初始内存字节数。
        "query-max-memory-bytes": 0, #  允许查询的最大内存总字节数。 等于查询并发×单次查询内存字节数
        "query-memory-bytes": 0,  # 单个查询允许的最大内存字节数。必须大于或等于 query-initial-memory-bytes 。
        "query-queue-size": 1024,  # 执行队列中允许的最大查询数。当达到队列限制时，新查询将被拒绝。
        "reporting-disabled": false,
        "secret-store": "bolt",
        "session-length": 60,
        "session-renew-disabled": false,
        "sqlite-path": "/var/lib/influxdb2/influxd.sqlite",
        "storage-cache-max-memory-size": 1073741824,  #  控制存储引擎内存缓存的大小，以字节为单位。
        "storage-cache-snapshot-memory-size": 26214400, # 控制存储引擎快照使用的内存大小，以字节为单位。
        "storage-cache-snapshot-write-cold-duration": "10m0s",  # 控制存储引擎冷数据快照写入的时间。
        "storage-compact-full-write-cold-duration": "4h0m0s",  # 控制存储引擎数据压缩的时间。
        "storage-compact-throughput-burst": 50331648,
        "storage-max-concurrent-compactions": 0,
        "storage-max-index-log-file-size": 1048576,
        "storage-no-validate-field-size": false,
        "storage-retention-check-interval": "30m0s",
        "storage-series-file-max-concurrent-snapshot-compactions": 0,
        "storage-series-id-set-cache-size": 0,
        "storage-shard-precreator-advance-period": "30m0s",
        "storage-shard-precreator-check-interval": "10m0s",
        "storage-tsm-use-madv-willneed": false,
        "storage-validate-keys": false,
        "storage-wal-fsync-delay": "0s",
        "storage-wal-max-concurrent-writes": 0,
        "storage-wal-max-write-delay": 600000000000,
        "storage-write-timeout": 10000000000,
        "store": "disk",
        "testing-always-allow-setup": false,
        "tls-cert": "",
        "tls-key": "",
        "tls-min-version": "1.2",
        "tls-strict-ciphers": false,
        "tracing-type": "",
        "ui-disabled": false,
        "vault-addr": "",
        "vault-cacert": "",
        "vault-capath": "",
        "vault-client-cert": "",
        "vault-client-key": "",
        "vault-client-timeout": 0,
        "vault-max-retries": 0,
        "vault-skip-verify": false,
        "vault-tls-server-name": "",
        "vault-token": ""
}

部署InfluxDB

我是基于Docker部署的InfluxDB2.3版本,docker-compose配置文件如下:

version: "3.1"
services:
  influxdb:
    image: influxdb:2.3.0
    container_name: influxdb
    ports:
      - "8086:8086"
    volumes:
      - influxdbv2:/var/lib/influxdb2
      - config.yml:/home/config.yml      
    environment:
      - DOCKER_INFLUXDB_INIT_MODE=setup
      - DOCKER_INFLUXDB_INIT_USERNAME=admin
      - DOCKER_INFLUXDB_INIT_PASSWORD=admin123
      - DOCKER_INFLUXDB_INIT_ORG=myorg
      - DOCKER_INFLUXDB_INIT_BUCKET=mybucket
      - DOCKER_INFLUXDB_INIT_RETENTION=1w
      - DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=mytoken
    restart: always

这个文件定义了一个名为influxdb的服务，它使用influxdb:2.3这个镜像来创建一个容器，它将容器的8086端口映射到主机的8086端口，它将容器的/var/lib/influxdb2目录挂载到主机的influxdbv2这个卷上，它设置了一些环境变量来初始化InfluxDB的用户，组织，存储桶，保留策略和令牌。

配置优化

[http]：这个部分控制了InfluxDB的HTTP服务，包括绑定地址，端口，超时时间，日志级别等。你可以根据你的网络环境和安全需求来调整这些参数。
[storage]：这个部分控制了InfluxDB的存储引擎，包括缓存大小，快照频率，压缩策略，索引大小等。你可以根据你的硬件资源和性能需求来调整这些参数。
[retention]：这个部分控制了InfluxDB的数据保留策略，包括检查间隔，启用状态等。你可以根据你的数据量和存储空间来调整这些参数。
[query]：这个部分控制了InfluxDB的查询引擎，包括并发度，队列大小，内存限制等。你可以根据你的查询负载和内存资源来调整这些参数。

针对内存占用空间过大的问题,可以尝试以下方法:

减少写入数据的频率和精度。如果你不需要实时或高精度的数据，你可以降低写入数据的频率和精度，从而减少数据量和索引大小。
增加数据压缩比率。如果你不需要高速查询或写入数据，你可以增加数据压缩比率，从而减少数据占用的空间。你可以通过修改storage-compact-throughput-burst参数来调整压缩比率。
启用数据保留策略。如果你不需要永久保存数据，你可以启用数据保留策略，从而定期删除过期或无用的数据。你可以通过修改retention-check-interval参数来调整保留策略。

GC优化

参考: HTSI 模式内存使用率高

- GODEBUG=madvdontneed=1
- GOGC=80

GODEBUG=madvdontneed=1可以减少InfluxDB的虚拟内存占用，而GOGC=80可以减少InfluxDB的实际内存占用。

入门

时序型数据在InfluxDB中是以一种叫做行协议（line protocol）的格式来表示的。行协议是一种文本格式，它由四个部分组成：

度量（measurement）：表示数据的类别或名称，比如温度，湿度，电压等。
标签（tag）：表示数据的元数据或属性，比如设备ID，位置，类型等。标签可以用来过滤或分组数据。
字段（field）：表示数据的值或内容，比如数值，字符串，布尔值等。字段可以用来计算或分析数据。
时间戳（timestamp）：表示数据的时间点或区间，比如秒，毫秒，纳秒等。时间戳可以用来排序或窗口化数据。

行协议的格式如下：

measurement,tag_key=tag_value field_key=field_value timestamp

例如，一个表示温度的行协议可以是这样的：

temperature,device_id=123,location=room1 value=25.6 1631377078000000000

这表示在1631377078000000000纳秒（2023-09-11T15:17:58Z）时，设备ID为123，位置为room1的温度值为25.6摄氏度。

设计原则

为了使数据模型易于查询且性能高效，需要遵循以下设计原则：

将常用的元数据或属性存储为标签，将变化的数值存储为字段。标签是被索引的，字段不是。这意味着通过标签过滤或分组数据比通过字段更快。例如，在查询某个位置的电表读数时，可以通过location标签快速定位到相关的数据点，而不需要扫描所有的字段值。
将每个属性存储为单独的标签，而不是将多个属性拼接在一起。这样可以避免在查询时使用正则表达式来解析复杂的标签值。正则表达式会降低查询性能和可读性。例如，在查询某个区域的水表读数时，可以通过region标签直接过滤出相关的数据点，而不需要从location标签中提取出区域信息。
避免在测量，标签和字段中存储数据。如果在这些要素中存储数据，会导致高基数（cardinality）问题。基数是指不同的测量和标签组合的数量。高基数会消耗更多的内存和CPU资源，并降低查询性能。例如，在存储电表读数时，应该将设备ID作为标签而不是测量或字段。
避免使用保留关键字或特殊字符作为标签和字段的名称。如果使用了Flux关键字作为名称，则需要在查询时用双引号括起来。如果使用了非字母数字字符作为名称，则需要在查询时用方括号表示法。
避免在同一个架构中使用相同的名称作为标签和字段。如果有一个标签和一个字段有相同的名称，则查询结果可能会出现不可预测的情况。

接收数据

收集数据是InfluxDB的第一个步骤，它指的是如何将时序型数据从不同的来源传输到InfluxDB中。时序型数据可以来自于各种设备，应用程序，服务，监控系统，日志系统等。为了方便地收集数据，InfluxDB提供了多种方法，包括：

使用Telegraf代理。Telegraf是一个开源的插件驱动的代理，它可以从各种输入源收集时序型数据，并且将其发送到各种输出目标。Telegraf支持超过200个插件，覆盖了常见的数据来源和目标。你可以使用Telegraf来轻松地将数据从你的设备或应用程序发送到InfluxDB中。
使用Flux语言。Flux是一种功能性的数据脚本语言，它可以用来查询，转换，分析和操作时序型数据。Flux也可以用来从其他数据库或服务中导入时序型数据到InfluxDB中。你可以使用Flux来编写自定义的数据收集逻辑，并且定期执行它。
使用HTTP API。HTTP API是InfluxDB提供的一种基于HTTP协议的接口，它可以用来执行各种操作，包括写入和查询数据。你可以使用HTTP API来直接向InfluxDB发送时序型数据，或者从其他服务中获取时序型数据并转发到InfluxDB中。
使用客户端库。客户端库是一些封装了InfluxDB的HTTP API的编程语言库，它们可以让你在你喜欢的编程语言中使用InfluxDB。客户端库支持多种语言，如Java, Python, Go, Ruby等。你可以使用客户端库来在你的代码中集成InfluxDB，并且向其发送时序型数据。

每种方法都有其优缺点，你可以根据你的需求和场景来选择合适的方法。下面我会简单地比较一下这些方法：

方法	优点	缺点
Telegraf	简单易用，支持多种插件，高效稳定	不够灵活，不能处理复杂的逻辑
Flux	强大灵活，支持多种函数和操作符，易于调试	需要学习新的语言，性能可能受限
HTTP API	直接高效，支持多种格式和参数，兼容多种工具	需要编写HTTP请求，不够友好
客户端库	方便集成，支持多种语言，提供高级功能	需要安装依赖，可能存在兼容性问题

写入数据

这里我提供一个基于Java的实现!

1.导入依赖

<dependency>
  <groupId>com.influxdb</groupId>
  <artifactId>influxdb-client-java</artifactId>
  <version>3.1.0</version>
</dependency>

2.创建连接

package example;

import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.InfluxDBClientFactory;

public class InfluxDB2Example {

  public static void main(final String[] args) {

    // You can generate an API token from the "API Tokens Tab" in the UI
    String token = "1231231243214234==";
    String bucket = "wangwang";
    String org = "wangwang";

    InfluxDBClient client = InfluxDBClientFactory.create("http://127.0.0.1:8086", token.toCharArray());
  }
}

3.写入数据

import com.influxdb.client.WriteApiBlocking;
import com.influxdb.client.write.Point;
import java.time.Instant;

// ...

Point point = Point
  .measurement("mem")
  .addTag("host", "host1")
  .addField("used_percent", 23.43234543)
  .time(Instant.now(), WritePrecision.NS);

WriteApiBlocking writeApi = client.getWriteApiBlocking();

writeApi.writePoint(bucket, org, point);

查询数据

在InfluxDB2.3版本中,用的是Flux语法,这个在以前1.x的版本中用的是InfluxQL语法.

查询InfluxDB的数据主要是以下步骤:

定义你的数据源。你需要使用from()函数来指定你要查询的存储桶的名称或ID。
定义你的时间范围。你需要使用range()函数来指定你要查询的时间范围，比如最近一小时，最近一天等。
定义你的数据过滤条件。你需要使用filter()函数来指定你要查询的数据的条件，比如度量名称，字段名称，标签名称或值等。
定义你的数据操作或分析方法。你可以使用各种Flux或InfluxQL函数来对数据进行操作或分析，比如求平均值，求标准差，求百分比，绘制图表等。

示例

这个查询会从example-bucket存储桶中获取最近一小时内度量为temperature的数据，并且对_value列求平均值。

from(bucket: "example-bucket")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == "temperature")
  |> mean(column: "_value")

InfluxDB提供了很多函数,可以参考官方文档:使用 Flux 查询数据,或者文章结尾参考的网址!

在官方的控制面板,也可以看到各个API的说明!

删除数据

POST http://localhost:8086/api/v2/delete

参数	类型	说明	示例
请求方法	字符串	指定HTTP请求的方法，必须为POST	POST
标题	字典	指定HTTP请求的头部信息，包括授权和内容类型	{“Authorization”: “Token mytoken”, “Content-Type”: “application/json”}
查询参数	字典	指定HTTP请求的查询参数，包括组织名称或ID和存储桶名称或ID	{“org”: “myorg”, “bucket”: “mybucket”}
请求正文	JSON对象	指定HTTP请求的正文内容，包括开始时间，结束时间和删除条件语句	{“start”: “1970-01-01T00:00:00Z”, “stop”: “2023-09-11T15:17:58Z”, “predicate”: “_measurement=”temperature” AND value>30”}

删除具有特定标签值的特定点

curl --request POST http://localhost:8086/api/v2/delete?org=example-org&bucket=example-bucket \
  --header 'Authorization: Token YOUR_API_TOKEN' \
  --header 'Content-Type: application/json' \
  --data '{
    "start": "2020-03-01T00:00:00Z",
    "stop": "2020-11-14T00:00:00Z",
    "predicate": "_measurement=\"example-measurement\" AND exampleTag=\"exampleTagValue\""
  }'

删除指定时间范围内的说有点

curl --request POST http://localhost:8086/api/v2/delete?org=example-org&bucket=example-bucket \
  --header 'Authorization: Token YOUR_API_TOKEN' \
  --header 'Content-Type: application/json' \
  --data '{
    "start": "2020-03-01T00:00:00Z",
    "stop": "2020-11-14T00:00:00Z"
  }'

如果请求成功，你应该看到一个空白的响应。如果请求失败，你应该看到一个包含错误信息的响应。

定时任务

InfluxDB中的定时任务是一种使用Flux语言编写的定期执行的脚本，它可以对输入的数据流进行修改或分析，然后将修改后的数据写回到InfluxDB或执行其他操作。定时任务的功能包括：

数据降采样：数据降采样是一种减少数据量和存储空间的方法，它可以将高精度或高频率的数据转换为低精度或低频率的数据，比如将每秒的数据转换为每分钟或每小时的数据。数据降采样可以提高查询效率，降低存储成本，增加数据保留期限。你可以使用定时任务来定期对你的数据进行降采样，并且将降采样后的数据存储在一个新的存储桶中。
数据清洗：数据清洗是一种提高数据质量和可用性的方法，它可以将不完整，不准确，不一致或无关的数据进行修改或删除，比如将空值填充，将异常值替换，将重复值去除，或者将无用值删除。数据清洗可以提高分析准确性，减少错误率，增加信任度。你可以使用定时任务来定期对你的数据进行清洗，并且将清洗后的数据写回到原来的存储桶中。
数据分析：数据分析是一种从数据中提取有价值信息和知识的方法，它可以将原始数据进行转换，计算，统计，聚合或可视化，比如求平均值，求标准差，求百分比，绘制图表等。数据分析可以提高决策支持，发现规律，预测趋势。你可以使用定时任务来定期对你的数据进行分析，并且将分析结果存储在一个新的存储桶中或发送到其他服务中。
数据告警：数据告警是一种根据数据状态或变化触发通知或动作的方法，它可以将实时或历史数据与预设的阈值或条件进行比较，并且在满足条件时发送邮件，短信，电话等通知方式，或者执行其他操作，比如调整参数，启动备份等。数据告警可以提高监控效果，及时响应，防止危机。你可以使用定时任务来定期对你的数据进行告警，并且将告警信息发送到指定的目标中

示例

// Task options
option task = {name: "downsample_5m_precision", every: 1h, offset: 0m}

// Data source
from(bucket: "example-bucket")
    |> range(start: -task.every)
    |> filter(fn: (r) => r._measurement == "mem" and r.host == "myHost")
    // Data processing
    |> aggregateWindow(every: 5m, fn: mean)
    // Data destination
    |> to(bucket: "example-downsampled")

备份还原数据

备份

influx backup <backup-path> -t <root-token>

还原

所有数据

influx restore /backups/2020-01-20_12-00/

特定桶数据

influx restore \
  /backups/2020-01-20_12-00/ \
  --bucket example-bucket

# OR

influx restore \
  /backups/2020-01-20_12-00/ \
  --bucket-id 000000000000