influxdb学习笔记

目录标题

一、influxdb简介
（一）InfluxDB 不是 CRUD
（二）InfluxDB 支持两种风格的查询（操作数据库的）方式
- 1. Flux
- 2. InfluxQL
- 3. Flux 与 InfluxQL
（三）dashboard
- 1. 2.0以下：
- 2. 2.0以上：
二、认证
三、java客户端参考
（一）版本1.8
（二）2.0以上版本
四、flux的基本查询

一、influxdb简介

（一）InfluxDB 不是 CRUD

InfluxDB 是一个针对时间序列数据进行优化的数据库。这些数据通常来自分布式传感器组、来自大型网站的点击数据或金融交易列表等来源。

这些数据的一个共同点是它在聚合中更有用。一项阅读说您的计算机的 CPU 在星期二 UTC 时间 12:38:35 的利用率为 12%，这很难得出结论。当与系列的其余部分结合并可视化时，它会变得更加有用。这是随着时间的推移开始显示趋势的地方，可以从数据中得出可操作的见解。另外，时序数据一般只写一次，很少更新。

结果是 InfluxDB 不是一个完整的 CRUD 数据库，而更像是一个 CR-ud，将创建和读取数据的性能优先于更新和销毁，并防止一些更新和销毁行为，使创建和读取的性能更高。

（二）InfluxDB 支持两种风格的查询（操作数据库的）方式

InfluxDB 支持多种查询语言：

Flux
InfluxQL

1. Flux

Flux是一种数据脚本语言，用于查询、分析和处理时间序列数据。从InfluxDB 1.8.0开始，Flux 可与 InfluxQL 一起用于生产。

注意：这个功能低版本可能没有开启，需要修改配置文件进行开启。

# influxdb.conf 配置文件
# 改为true
flux-enabled = true
# 然后启动的时候需要指定influxdb.conf 启动
influxd.exe -config influxdb.conf

Flux示例：

from(bucket:"telegraf/autogen")|> range(start: -15m)|> filter(fn: (r) =>r._measurement == "cpu" andr._field == "usage_system" andr.cpu == "cpu-total")

更多参考官网

2. InfluxQL

这种语法跟sql语法基本保持一致。
InfluxQL 是一种类似于 SQL 的查询语言，用于与 InfluxDB 交互。它经过精心设计，让来自其他 SQL 或类 SQL 环境的人感到熟悉，同时还提供特定于存储和分析时间序列数据的功能。然而InfluxQL不是SQL，缺乏像更高级的操作支持UNION，JOIN并HAVING认为SQL电力用户习惯。此功能可用于Flux。
InfluxQL 的SELECT语句遵循 SQLSELECT语句的形式：

SELECT FROM WHERE

示例:

SELECT * FROM "foodships" WHERE "planet" = 'Saturn'

注意：2.0以上版本视乎移除了InfluxQL。

3. Flux 与 InfluxQL

Flux 是 InfluxQL 和其他类似 SQL 的查询语言的替代品，用于查询和分析数据。Flux 使用函数式语言模式，使其非常强大、灵活，并且能够克服 InfluxQL 的许多限制。本文概述了使用 Flux 而不是 InfluxQL 可能完成的许多任务，并提供了有关 Flux 和 InfluxQL parity 的信息。可以使用 Flux InfluxQL 和 Flux parity 可以使用 Flux Joins Math 跨测量按标签分组按日历月份和年份窗口处理多个数据源 DatePart-like 查询 Pivot Histograms Covariance Cast booleans to integers 字符串操作和数据整形使用地理时间数据连接 InfluxQL 从未支持连接。

可以去官网了解更多

（三）dashboard

注意：在2.0一下的版本是没有可视化的dashboard 功能的

1. 2.0以下：

访问：http://localhost:8086/

当然如果你不习惯使用bash开发，可以下载官方推荐的三方dashboard可视化管理界面。参考地址

使用Chronograf或Grafana仪表板来可视化您的时间序列数据。(推荐第一个)

Chronograf安装教程：1、解压，2、通过cmd 运行chronograf.exe，3、浏览器访问http://localhost:8888/

2. 2.0以上：

访问：http://localhost:8086/

注意这里可以获取令牌。

二、认证

默认情况下influxdb是没有开启认证的。

当我们直接启动，那就会加载默认的配置。默认的配置是没有开启认证的。

.\influxd.exe

我们指定一个配置文件

.\influxd.exe -config influxdb.conf

开启认证

创建管理员用户
1. 当您启用 HTTP 身份验证时，InfluxDB 要求您在与系统交互之前至少创建一个管理员用户。
```
CREATE USER admin WITH PASSWORD '<password>' WITH ALL PRIVILEGES
```
1. 创建另一个管理员用户
```
CREATE USER <username> WITH PASSWORD '<password>' WITH ALL PRIVILEGES
```

三、java客户端参考

（一）版本1.8

导入pom依赖

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

java客户端

package com.lihua;import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.InfluxDBClientFactory;
import com.influxdb.client.WriteApi;
import com.influxdb.client.flux.FluxClient;
import com.influxdb.client.flux.FluxClientFactory;
import com.influxdb.client.flux.FluxConnectionOptions;
import com.influxdb.client.write.Point;
import com.influxdb.query.FluxRecord;
import com.influxdb.query.FluxTable;import java.util.List;/*** @author hs* @date 2021/11/10 17:30*/
public class InfluxdbClientTest18 {public static void main(final String[] args) {//选择数据库，可以通过SHOW DATABASES 命令查询。// 如果没有数据库可以通过CREATE DATABASE "db_name"创建String database = "myinflux";//这是保留策略，注意：from(bucket:"<database>/<retention-policy>") 。String retentionPolicy = "autogen";InfluxDBClient client = InfluxDBClientFactory.createV1("http://localhost:8086","admin","admin123".toCharArray(),database,retentionPolicy);System.out.println("*** Write Points ***");try (WriteApi writeApi = client.makeWriteApi()) {Point point = Point.measurement("mem").addTag("host", "host1").addField("used_percent", 29.43234543);System.out.println(point.toLineProtocol());writeApi.writePoint(point);}System.out.println("*** Query Points ***");String query = String.format("from(bucket: \"%s/%s\") |> range(start: -1h)", database, retentionPolicy);List<FluxTable> tables = client.getQueryApi().query(query);tables.get(0).getRecords().forEach(record -> System.out.println(String.format("%s %s: %s %s",record.getTime(), record.getMeasurement(), record.getField(), record.getValue())));client.close();}
}

（二）2.0以上版本

导入pom依赖

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

java客户端代码

package com.lihua;import com.influxdb.client.*;
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.client.write.Point;
import com.influxdb.query.FluxRecord;
import com.influxdb.query.FluxTable;import java.time.Instant;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.Objects;/*** @author hs* @date 2021/11/8 10:08*/
public class InfluxdbClientTest {//这个是令牌需要先获取令牌，可以在2.0的dashboard可视化页面中创建 private static char[] token = "ehtg-uIR1wqNtH3CDUv5RwHPPb3QTdQ9s4zadE0WUHCRfazV5GT7GLz_tM2g-qFVxI0qjiWdZqGtctoba_waWg==".toCharArray();//组织（公司）private static String org = "test-group";//桶（数据库）private static String bucket = "test";public static void main(String[] args) {InfluxDBClient client = InfluxDBClientFactory.create("http://localhost:8086", token,org,bucket);//// Write data//WriteApiBlocking writeApi = client.getWriteApiBlocking();//// Write by Data Point//Point point = Point.measurement("temperature").addTag("location", "west").addField("value", 55D).time(Instant.now().toEpochMilli(), WritePrecision.MS);System.out.println("开始写："+DateTimeFormatter.ISO_LOCAL_DATE_TIME.format(LocalDateTime.now()));writeApi.writePoint(point);System.out.println("完成："+DateTimeFormatter.ISO_LOCAL_DATE_TIME.format(LocalDateTime.now()));//// Query data//String flux = "from(bucket:\"test\") |> range(start: 0)";System.out.println("开始读："+DateTimeFormatter.ISO_LOCAL_DATE_TIME.format(LocalDateTime.now()));QueryApi queryApi = client.getQueryApi();List<FluxTable> tables = queryApi.query(flux);System.out.println("完成："+DateTimeFormatter.ISO_LOCAL_DATE_TIME.format(LocalDateTime.now()));for (FluxTable fluxTable : tables) {List<FluxRecord> records = fluxTable.getRecords();for (FluxRecord fluxRecord : records) {System.out.println(fluxRecord.getTime() + ": " + fluxRecord.getValueByKey("_value"));}}client.close();}
}

四、flux的基本查询

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

来源——from
Flux输入函数从数据源检索数据。所有输入函数都返回一个表流。也就是要指定查询的桶（桶等于db-name/保留策略）
筛选——range、filter
过滤器函数迭代并评估每个输入行以查看它是否与指定条件匹配。满足条件的行包含在函数输出中。不满足指定条件的行将被删除。
Flux 提供以下主要过滤器功能：
- range(): 根据时间过滤数据。
- filter(): 根据列值过滤数据。 filter()使用参数中定义的谓词函数fn来评估输入行。每一行被传递到谓词函数作为记录，r含有键-值对的行中的每一列中。
形状——group
许多查询需要修改数据的结构以准备处理。常见的数据整形任务包括按列值或按时间重新分组数据或将列值转换为行。
重塑数据的函数包括：
- group(): 修改组键
- window()：修改_start和_stop通过的时间来组数据行的值
- pivot(): 将列值转为行
- drop(): 删除特定列
- keep()：保留特定列并删除所有其他列
过程——mean
处理数据可以采用多种形式，包括以下类型的操作：
- 聚合数据：将输入表的所有行聚合为一行。有关信息，请参阅函数类型和类别 - 聚合。
- 选择特定数据点：从每个输入表中返回特定行。例如，返回第一行或最后一行、具有最高或最低值的行等。有关信息，请参阅函数类型和类别 - 选择器。
- 重写行：用于map()重写每个输入行。使用数学运算转换值、处理字符串、动态添加新列等。
- 发送通知：评估数据并使用 Flux 通知端点函数向外部服务发送通知。有关信息，请参阅函数类型和类别 - 通知端点。

推荐查看官方文档——flux的基本用法

2.0以下版本的Java客户端参考例子
2.0以上的例子