InfluxDB 2.x
简介
InfluxDB 2.x 是 InfluxData 公司推出的时间序列数据库的最新版本。与 InfluxDB 1.x 相比,InfluxDB 2.x 引入了一些新的特性和改进,旨在提供更强大、灵活和易于使用的时间序列数据存储和查询解决方案。以下是 InfluxDB 2.x 的一些关键特点和概念:
- Flux 查询语言: InfluxDB 2.x 引入了 Flux 查询语言,用于执行强大的数据处理和分析操作。Flux 提供了更灵活的查询能力,支持复杂的数据转换、过滤、聚合和连接操作,以更好地支持用户对时间序列数据的操作和分析。
- 新的数据存储引擎: InfluxDB 2.x 使用 Flux Engine 作为新的数据存储引擎。Flux Engine 具有更好的性能和扩展性,支持高并发写入和高效查询。
- Bucket 和 Organization: InfluxDB 2.x 引入了 Bucket 和 Organization 的概念。Bucket 是数据的逻辑容器,用于组织和存储时间序列数据,而 Organization 用于隔离和管理不同的用户或应用程序。
- 新的用户界面(UI): InfluxDB 2.x 提供了全新的仪表板和可视化工具,使用户能够轻松创建、定制和共享漂亮的图表和仪表板,以直观地展示时间序列数据。
- API 密钥和访问令牌: 为了实现更安全的访问和操作,InfluxDB 2.x 引入了 API 密钥和访问令牌。API 密钥用于身份验证,访问令牌用于授权对数据的访问。
- 支持多租户: InfluxDB 2.x 提供更好的多租户支持,允许在同一实例中隔离和管理多个不同的组织和数据。
- 新的存储结构: InfluxDB 2.x 采用新的存储结构,支持更灵活的标签和更好的查询性能,同时改进了存储效率。
- 云原生和容器化: InfluxDB 2.x 被设计用于云原生环境和容器化部署,支持在 Kubernetes 或其他容器编排工具中运行。
- 更好的持久性查询支持: Flux 查询语言支持对查询结果进行持久性存储,使用户可以创建并定期执行持久性的计算任务。
基本概念
以下是 InfluxDB 2.x 的一些基本概念,这些概念有助于理解和使用 InfluxDB:
- Bucket(桶): Bucket 是 InfluxDB 中的基本存储单元,用于组织和存储时间序列数据。它是数据的逻辑容器,用于区分和隔离不同类型或来源的数据。
- Organization(组织): Organization 是 InfluxDB 中的组织概念,用于隔离和管理用户、应用程序或项目。每个用户可以属于一个或多个组织,Bucket 也属于一个组织。
- Measurement(测量值): Measurement 是 InfluxDB 中的数据表的概念。它类似于关系型数据库中的表,用于存储时间序列数据。每个 Measurement 包含一组字段和标签。
- Field(字段): Field 是 Measurement 中的一种数据类型,用于存储实际的数值数据。例如,温度、湿度等是字段的例子。
- Tag(标签): Tag 是 Measurement 中的一种索引类型,用于标识和过滤数据。标签通常用于存储维度信息,例如地理位置、设备 ID 等。
- Point(数据点): Point 是 InfluxDB 中时间序列数据的基本单位,它包含了时间戳、字段和标签。每个数据点都关联到一个 Measurement。
- API 密钥和访问令牌: 为了进行安全的访问和操作,InfluxDB 2.x 引入了 API 密钥和访问令牌。API 密钥用于身份验证,而访问令牌用于授权对数据的访问。
- Flux 查询语言: Flux 是 InfluxDB 2.x 中的查询语言,用于执行强大的数据处理和分析操作。它支持更灵活的查询、过滤、聚合和转换操作。
- 仪表板和可视化: InfluxDB 2.x 提供了仪表板和可视化工具,允许用户创建、定制和共享图表和仪表板,以直观地展示时间序列数据。
Bucket类似于关系型数据库中的库的概念。Measurement类似于数据库中表。
应用场景
广泛应用于物联网、监控系统、日志分析、金融数据分析等。以下是一些时序数据库的常见应用场景:
- 监控和运维: 时序数据库常用于监控系统的性能和运行状态。通过记录服务器、网络设备、应用程序等的性能指标,管理员可以实时监测系统健康状况,快速识别并解决问题。
- 物联网(IoT): 物联网设备产生大量的时间序列数据,包括传感器读数、设备状态和事件记录。时序数据库能够高效地存储和查询这些数据,为物联网应用提供实时性和高吞吐量。
- 能源管理: 时序数据库可用于跟踪能源使用情况,记录电表读数、太阳能发电量、温度等信息。这对于能源消耗分析、可再生能源监测和节能优化非常有用。
- 金融数据分析: 金融行业需要处理大量的交易数据、股价变动和其他市场指标。时序数据库能够提供快速的查询和分析功能,以便实时监控市场情况和执行高频交易策略。
- 日志分析: 服务器日志、应用程序日志等产生的日志数据通常是按时间戳记录的。时序数据库可用于存储和分析大量的日志数据,以便追踪系统事件、调查问题和进行故障排除。
- 工业自动化: 在制造业和工业自动化中,时序数据库用于记录生产过程中的传感器数据、生产线状态和设备运行情况,以实现实时监控和质量控制。
- 环境监测: 时序数据库可用于存储气象数据、空气质量指标、水质数据等环境监测信息。这有助于研究气候变化、提高自然灾害预警系统的效能等。
服务端配置
安装环境
Windows系统,Docker安装最新influxdb镜像,设置宿主机映射默认端口:8086,运行成功后,Dashboad访问路径127.0.0.1:8086,初次打开需要设置用户名,密码和组织Organization名称。
创建数据库
点击Bluckets选项,点击 “CREATE BLUCKET” 按钮添加Blucket
为bucket添加TELEGRAF配置
为添加的bucket添加数据来源配置,点击”CREATE CONFIGURATION”按钮
选择新加的Blucket为其添加System数据源
只需要填写配置名称/描述,配置内容不需要自己填写,保存自动生成。
保存后出现该界面代表创建完成,会返回给配置信息:export INFLUX_TOKEN 和 telegraf --config
点击后关闭界面
TELEGRAF配置参数说明
这四个key的值就对应的是JAVA应用程序中yaml中的配置的四个属性值,分别是url、token、org、bucket
注意:2.x版本是通过这四个属性来访问的,不再是账号和密码。
其中token的值就是第二步创建完配置文件后返回的两个配置文件中的 export INFLUX_TOKEN
经过测试发现了问题,注意这个TOKEN是数据库配置的TOKEN虽然可以连接到数据库并成功插入数据,但是并不具备访问的权限的,也就是说只能保存不能进行其他操作。查询报错:HTTP status code: 404; Message: failed to initialize execute state: could not find bucket “XX”
配置数据库的访问权限API TOKENS
选择“API TOKENS”选项后,点击”GENERATE API TOKEN”按钮中”Customer API Token”选项配置
为新建的bucket创建读写权限
点击“Generate”按钮后,会弹窗生成API TOKEN值,该TOKENS直接替换掉yaml配置文件中的token即可
客户端使用
应用环境
以spring-boot 项目,maven构建为例,maven文件引入客户端包,源码Github地址:https://github.com/influxdata/influxdb-client-java,maven配置如下所示:
<dependency>
<groupId>com.influxdb</groupId>
<artifactId>influxdb-client-java</artifactId>
<version>6.10.0</version>
</dependency>
<dependency>
<groupId>com.influxdb</groupId>
<artifactId>influxdb-client-core</artifactId>
<version>6.10.0</version>
</dependency>
配置参数
在yaml中配置或者新增配置文件。本项目以配置文件为例,在resouces下新增influx2.properties配置文件。配置文件内容如下所示:
influx2.url=http://127.0.0.1:8086
influx2.org=21
influx2.token=n1laoPJafR9NkPJ_eIulO8fURL2I1-s_K582HByY1PGMtqVXmP1z2zHtwz-hmxS6aO336MfIF4H7EqpXOKZ1Fg==
influx2.bucket=chj-influx
添加配置类InfluxConfigProperties.java
package guru.springframework.config;
import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.InfluxDBClientFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class InfluxConfigProperties {
@Bean
public InfluxDBClient influxDBClient() {
return InfluxDBClientFactory.create();
}
}
FLUX查询语言
Flux 是一种功能强大的查询语言,特别适用于处理时间序列数据。Flux 具有丰富的操作和函数,使用户能够执行复杂的数据处理、过滤、聚合和计算任务。其中bucket 和 range 是必填的。
以下是一些 Flux 查询和计算的基本示例:
查询数据:
-
选择数据点:
from(bucket: "your_bucket") |> range(start: -1h) |> filter(fn: (r) => r._measurement == "your_measurement")
-
选择特定字段:
from(bucket: "your_bucket") |> range(start: -1h) |> filter(fn: (r) => r._measurement == "your_measurement") |> filter(fn: (r) => r._field == "your_field")
数据处理和计算:
-
计算平均值:
from(bucket: "your_bucket") |> range(start: -1h) |> filter(fn: (r) => r._measurement == "your_measurement") |> filter(fn: (r) => r._field == "your_field") |> mean()
-
计算总和:
from(bucket: "your_bucket") |> range(start: -1h) |> filter(fn: (r) => r._measurement == "your_measurement") |> filter(fn: (r) => r._field == "your_field") |> sum()
-
计算差异:
from(bucket: "your_bucket") |> range(start: -1h) |> filter(fn: (r) => r._measurement == "your_measurement") |> difference()
-
计算百分比:
from(bucket: "your_bucket") |> range(start: -1h) |> filter(fn: (r) => r._measurement == "your_measurement") |> map(fn: (r) => ({ r with _value: r._value * 100 }))
数据过滤和筛选:
-
按标签过滤:
from(bucket: "your_bucket") |> range(start: -1h) |> filter(fn: (r) => r._measurement == "your_measurement" and r.your_tag == "desired_value")
-
按条件过滤:
from(bucket: "your_bucket") |> range(start: -1h) |> filter(fn: (r) => r._measurement == "your_measurement" and r._value > 10.0)
-
按时间范围过滤:
from(bucket: "your_bucket") |> range(start: -1h, stop: now()) |> filter(fn: (r) => r._measurement == "your_measurement")
这些示例演示了 Flux 查询语言中的一些常见操作,但 Flux 提供了更多的功能,可以通过组合和嵌套这些操作来执行更复杂的查询和计算任务。
数据集结构体说明
FluxTable.java
//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by FernFlower decompiler)
//
package com.influxdb.query;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.StringJoiner;
import java.util.stream.Collectors;
import javax.annotation.Nonnull;
public final class FluxTable implements Serializable {
private List<FluxColumn> columns = new ArrayList();
private List<FluxRecord> records = new ArrayList();
public FluxTable() {
}
@Nonnull
public List<FluxColumn> getColumns() {
return this.columns;
}
@Nonnull
public List<FluxColumn> getGroupKey() {
return (List)this.columns.stream().filter(FluxColumn::isGroup).collect(Collectors.toList());
}
@Nonnull
public List<FluxRecord> getRecords() {
return this.records;
}
public String toString() {
return (new StringJoiner(", ", FluxTable.class.getSimpleName() + "[", "]")).add("columns=" + this.columns.size()).add("records=" + this.records.size()).toString();
}
}
FluxRecord.java
//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by FernFlower decompiler)
//
package com.influxdb.query;
import com.influxdb.utils.Arguments;
import java.io.Serializable;
import java.time.Instant;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.StringJoiner;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
public final class FluxRecord implements Serializable {
private final Integer table;
private LinkedHashMap<String, Object> values = new LinkedHashMap();
private List<Object> row = new ArrayList();
public FluxRecord(@Nonnull Integer table) {
Arguments.checkNotNull(table, "Table index");
this.table = table;
}
@Nullable
public Instant getStart() {
return (Instant)this.getValueByKey("_start");
}
@Nullable
public Instant getStop() {
return (Instant)this.getValueByKey("_stop");
}
@Nullable
public Instant getTime() {
return (Instant)this.getValueByKey("_time");
}
@Nullable
public Object getValue() {
return this.getValueByKey("_value");
}
@Nullable
public String getField() {
return (String)this.getValueByKey("_field");
}
@Nullable
public String getMeasurement() {
return (String)this.getValueByKey("_measurement");
}
@Nonnull
public Integer getTable() {
return this.table;
}
@Nonnull
public Map<String, Object> getValues() {
return this.values;
}
@Nonnull
public List<Object> getRow() {
return this.row;
}
@Nullable
public Object getValueByIndex(int index) {
return this.values.values().toArray()[index];
}
@Nullable
public Object getValueByKey(@Nonnull String key) {
Arguments.checkNonEmpty(key, "key");
return this.values.get(key);
}
public String toString() {
return (new StringJoiner(", ", FluxRecord.class.getSimpleName() + "[", "]")).add("table=" + this.table).add("values=" + this.values.size()).toString();
}
public boolean equals(Object o) {
if (this == o) {
return true;
} else if (o != null && this.getClass() == o.getClass()) {
FluxRecord that = (FluxRecord)o;
return Objects.equals(this.table, that.table) && Objects.equals(this.values, that.values);
} else {
return false;
}
}
public int hashCode() {
return Objects.hash(new Object[]{this.table, this.values});
}
}
返回结果如下图所示:
说明:cloumns前8个是固定,从第9个为设置的Tag。records里面的值对应的列cloumn的取值。桶一个tag下每个字段会展示一条数据,_field为字段名称,_value为该字段的值,按tag为索引,按字段为维度存取数据,侧重于某个字段(指标)的展示,在DashBoads里面展示如下图所示:
部分功能代码演示
新增数据
- 通过Influx 行协议的方法写入。writeApiBlocking.writeRecord
- 通过写入 Point点。writeApiBlocking.writePoint
- POJO的模式写入。writeApi.writeMeasurement
@Autowired
private InfluxDBClient influxDBClient;
public void addMyMeasurement() {
WriteApiBlocking writeApiBlocking = influxDBClient.getWriteApiBlocking();
// 1. 通过Influx 行协议的方法写入 writeRecord
// 选择时间戳模式
writeApiBlocking.writeRecord(WritePrecision.NS, "MyMeasurement,location=河南 temperature=34.2,humidity=99");
// 2. 通过写入 Point点API
Point point = Point.measurement("MyMeasurement")
.addTag("location", "厦门")
.addField("temperature", 32.12)
.addField("humidity", 88.0);
writeApiBlocking.writePoint(point);
// 3. POJO的模式写入
WriteApi writeApi = influxDBClient.makeWriteApi();
MyMeasurement myMeasurement = new MyMeasurement();
myMeasurement.setLocation("广州");
myMeasurement.setTemperature(88.0);
myMeasurement.setHumidity(77);
writeApi.writeMeasurement(WritePrecision.NS,myMeasurement);
}
如果使用第三种方式,则其中POJO类代码如下所示:
package guru.springframework.domain;
import com.influxdb.annotations.Column;
import com.influxdb.annotations.Measurement;
import lombok.Data;
import java.time.Instant;
@Data
@Measurement(name = "MyMeasurement")
public class MyMeasurement {
@Column(tag = true)
private String location;
/**
* 温度
*/
@Column
private double temperature;
/**
* 湿度
*/
@Column
private double humidity;
@Column(timestamp = true)
Instant time;
}
查询数据
public List<MyMeasurement> getMyMeasurements() {
String flux = "from(bucket: \"chj-influx\")\n" +
" |> range(start: -38h)\n" ;
List<MyMeasurement> myMeasurements = new ArrayList<>();
List<FluxTable> results = influxDBClient.getQueryApi().query(flux);
for (FluxTable table : results) {
for (FluxRecord record : table.getRecords()) {
MyMeasurement myMeasurement = new MyMeasurement();
if (record.getField().equals("temperature")) {
myMeasurement.setTemperature(Double.parseDouble(record.getValue().toString()));
}else{
continue;
}
// if (record.getField().equals("humidity")) {
// myMeasurement.setHumidity(Double.parseDouble(record.getValue().toString()));
// }
myMeasurement.setLocation(record.getValueByKey("location").toString());
myMeasurement.setTime(record.getTime());
myMeasurements.add(myMeasurement);
}
}
地区温度字段(指标)的结果展示如下所示
{
"code": 200,
"message": "操作成功",
"data": [
{
"location": "厦门",
"temperature": 32.12,
"humidity": 0,
"time": "2024-03-09T03:43:11.162747749Z"
},
{
"location": "河南",
"temperature": 34.2,
"humidity": 0,
"time": "2024-03-09T03:43:11.101985536Z"
}
]
}
删除数据
只能按tag删除,调用deleteApi.delete方法。
/*
* 只能按tag删除
*/
public void deleteMyMeasurement() {
String predictSql = "location=\"河南\"";
DeleteApi deleteApi = influxDBClient.getDeleteApi();
deleteApi.delete(OffsetDateTime.now().minusHours(24),
OffsetDateTime.now(),
predictSql,
bucket,
org
);
}
示例项目地址下载
[https://github.com/caohuajin/influxDB2.x]
标签:return,bucket,InfluxDB,your,import,public From: https://www.cnblogs.com/caohuajin/p/18063025