InfluxDB数据库>时序数据库的使用

InfluxDB
- InfluxDB概述
- 相关概念
安装InfluxDB
InfluxDB的操作
- 用户管理
- 启用认证
- 数据库操作
- 数据库表操作
- 查询操作
Spring Boot集成InfluxDB
- 添加依赖
- 创建数据库
- 配置InfluxDB
- 编码
- 执行测试

InfluxDB

InfluxDB概述

InfluxDB 是一个从头开始构建的时间序列数据库，用于处理高写入和查询负载。InfluxDB 旨在用作任何涉及大量时间戳数据的用例的后备存储，包括 DevOps 监控、应用程序指标、物联网传感器数据和实时分析。

什么是时间序列的数据？从定义上来说，就是一串按时间维度索引的数据。

时间序列数据库(TSDB)特点：

持续高并发写入、不用更新
数据压缩存储
低延时查询

常见的时间序列数据库除了InfluxDB之外还有：opentsdb、timeScaladb、Druid等。

官网： https://www.influxdata.com/

GitHub：https://github.com/influxdata/influxdb

中文文档：https://jasper-zhang1.gitbooks.io/influxdb/content/Introduction/installation.html

nfuxDB	传统数据库
database	数据库
measurement	数据库中的表
point	表中的行

point属性	含义
time	数据记录的时间，主索引，默认自动生成，相当于每行数据都具备的列
tags	相当于有索引的列。tag中存储的值的类型是字符串类型
fields	value值，没有索引的列。field中存储的值得类型：字符串、浮点数(Double)、整数、布尔型。一个field value总是和一个timestamp相关联

安装InfluxDB

拉取influxdb

java">docker pull influxdb:1.8

启动influxdb

java">docker run -d --name influxdb -p 8086:8086  influxdb:1.8

进入influxdb容器

java">docker exec -it influxdb /bin/bash

在命令行里输入influx，进入InfluxDB数据库环境

java">root@70783fdf1b06:/# influx
Connected to http://localhost:8086 version 1.8.10
InfluxDB shell version: 1.8.10
>

InfluxDB的操作

用户管理

创建用户

java">> create user "root" with password 'root' with all privileges
> show users
user admin
---- -----
root true
>

删除用户

java">> drop user root
> 
> show users
user admin
---- -----
>

启用认证

配置influxdb.conf文件启用认证

java">docker cp influxdb:/etc/influxdb/influxdb.conf ./

docker cp ./influxdb.conf influxdb:/etc/influxdb/

docker restart influxdb

在influxdb.conf配置文件中添加如下内容

java[meta]">[http]
auth-enabled= true

使用账号密码登录

java">root@70783fdf1b06:/#  influx -username 'root' -password 'root'
Connected to http://localhost:8086 version 1.8.10
InfluxDB shell version: 1.8.10
> show users
user admin
---- -----
root true
>

错误密码认证

java">root@70783fdf1b06:/#  influx -username 'root' -password 'root123'
Connected to http://localhost:8086 version 1.8.10
InfluxDB shell version: 1.8.10
> show users
ERR: authorization failed
Warning: It is possible this error is due to not setting a database.
Please set a database with the command "use <database>".
>

数据库操作

显示数据库：show databases

java">> show databases
name: databases
name
----
_internal
>

新建数据库：create database test

java">> create database test
> show databases
name: databases
name
----
_internal
test
>

删除数据库命令：drop database test

java">> drop database test
> show databases
name: databases
name
----
_internal
>

使用某个数据库，必须先创建create database test ，然后再使用use test

java">> create database test
> use test
Using database test
>

数据库表操作

显示所有表：SHOW MEASUREMENTS

java">> SHOW MEASUREMENTS
>

新建表：insert

InfluxDB中没有显式的新建表的语句，只能通过insert数据的方式来建立新表，或者在客户端程序代码中插入数据的时候就会自动建表。

以下命令表示：一个measurement为cpu，tag是host和region，value值为0.64的数据点被写入到InfluxDB中

java">> INSERT cpu,host=serverA,region=us_west value=0.64
> SHOW MEASUREMENTS
name: measurements
name
----
cpu
>

删除表：drop measurement cpu

java">> drop measurement cpu
> SHOW MEASUREMENTS
>

查询操作

InfluxQL是一种类似SQL的查询语言，用于与InfluxDB中的数据进行交互。

表数据查询：SELECT * FROM tb

java">>  SELECT "host", "region", "value" FROM "cpu"
name: cpu
time                host    region  value
----                ----    ------  -----
1642915045925737934 serverA us_west 0.64
>

InfluxDB中的模糊搜索和mysql中的模糊搜索一样

查询包含给定字段数据

java">select fieldName from measurementName where fieldName=~/条件值/

查询以给定字段开始的数据

java">select fieldName from measurementName where fieldName=~/^条件值/

查询以给定字段结束的数据

java">select fieldName from measurementName where fieldName=~/条件值$/

统计记录个数

java">select count(value) from measurementName where time>='2021-01-01' and time<='2022-12-31'

分页

java">select * from measurementName where time>='2021-01-01' and time<='2022-12-31' order by desc limit 10 offset 0

Spring Boot集成InfluxDB

添加依赖

java"><dependency>
    <groupId>org.influxdb</groupId>
    <artifactId>influxdb-java</artifactId>
    <version>2.21</version>
</dependency>

创建数据库

java">create database demo

配置InfluxDB

java">spring: 
  influx:
    user: root
    password: root
    url: http://IP:8086
    db: demo

编码

java">/**
 * 封装统计记录数
 **/
@Data
@Measurement(name = "bbu")
public class BbuCount {
    @Column(name = "count")
    private Long count;
}

java">/**
 * 在类上添加@Measurement注解用来标注最终存储的表名，这样在插入数据时会自动创建该Measurement，名字为bbu
 * <p>
 * 对应InfluxDB中tag的映射，在定义字段上加`@Column`注解，且用`name`来设置对应的tag名，并且设置`tag`为`true`证明是tag
 */
@Data
@Measurement(name = "bbu")
public class BbuInfo {

    /**
     * 设备编号
     */
    @Column(name = "deviceId", tag = true)
    private String deviceId;
    /**
     * 指标名称
     */
    @Column(name = "indicatorName", tag = true)
    private String indicatorName;
    /**
     * 是否告警  0：不告警  1：告警
     */
    @Column(name = "alarm", tag = true)
    private String alarm;
    /**
     * 告警级别
     */
    @Column(name = "level", tag = true)
    private String level;
    /**
     * 指标数值
     */
    @Column(name = "value")
    private Double value;
    /**
     * 时间序列
     */
    @Column(name = "time")
    private String time;
}

java">import cn.ybzy.demo.model.BbuCount;
import cn.ybzy.demo.model.BbuInfo;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.influxdb.InfluxDB;
import org.influxdb.dto.Point;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;
import org.influxdb.impl.InfluxDBResultMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Component
@Slf4j
public class BbuInflux {

    @Autowired
    private InfluxDB influxDB;

    @Value("${spring.influx.db}")
    private String dbName;

    /**
     * 添加数据
     *
     * @param object
     */
    public void add(Object object) {

        Point.Builder pointBuilder = Point.measurementByPOJO(object.getClass());

        // 将对象中的所有属性转换为tag添加到point中
        Point point = pointBuilder.addFieldsFromPOJO(object)
                // 调用time方法设置当前时间,InfluxDB自动生成的time时间列为UTC时间，所以在存储时单独设置
                .time(LocalDateTime.now().plusHours(8).toInstant(ZoneOffset.of("+8")).toEpochMilli(), TimeUnit.MILLISECONDS)
                .build();

        // 设置要存储的数据库名称
        influxDB.setDatabase(dbName);
        // 将数据插入到表(Measurement)中
        influxDB.write(point);
        influxDB.close();
    }

    /**
     * 通用查询封装
     *
     * @param sql
     * @param clazz
     * @param <T>
     * @return
     */
    public <T> List<T> query(String sql, Class<T> clazz) {
        QueryResult queryResult = influxDB.query(new Query(sql, dbName));
        influxDB.close();
        InfluxDBResultMapper resultMapper = new InfluxDBResultMapper();
        return resultMapper.toPOJO(queryResult, clazz);
    }

    /**
     * 分页查询
     *
     * @param page
     * @param pageSize
     * @param start
     * @param end
     * @param deviceId
     * @return
     */
    public HashMap<String, Object> queryBbu(Long page, Long pageSize, String start, String end, String deviceId) {
        StringBuilder sql = new StringBuilder("select * from bbu  ");

        // 构建where条件
        sql.append(" where ");
        if (StringUtils.isNoneBlank(start)) {
            sql.append(" time>='" + start + "' ");
        }
        if (StringUtils.isNoneBlank(end)) {
            sql.append("and time<='" + end + "' ");
        }
        if (StringUtils.isNoneBlank(deviceId)) {
            sql.append("and deviceId=~/" + deviceId + "/ ");
        }

        // 分页
        sql.append("order by desc limit " + pageSize + " offset " + (page - 1) * pageSize);

        // 查询记录语句
        List<BbuInfo> quotaList = this.query(sql.toString(), BbuInfo.class);

        // 统计语句
        String countSql = "select count(value) from bbu ";
        List<BbuCount> bbuCount = this.query(countSql, BbuCount.class);

        if (bbuCount == null || bbuCount.size() <= 0) {
            return null;
        }

        Long totalCount = bbuCount.get(0).getCount();
        HashMap<String, Object> map = new HashMap<>();
        map.put("total", totalCount);
        map.put("data", quotaList);
        return map;
    }

    /**
     * 查询最新数据
     *
     * @param deviceId
     * @return
     */
    public HashMap<String, Object> queryLast(String deviceId) {
        String sql = " select last(value),* from bbu where deviceId='" + deviceId + "' group by deviceId";
        List<BbuInfo> quotaList = this.query(sql, BbuInfo.class);

        HashMap<String, Object> map = new HashMap<>();
        map.put("data", quotaList);
        return map;
    }
}

执行测试

java">@SpringBootTest
class SpringbootApplicationTests {

    @Autowired
    private BbuInflux bbuInflux;

    @Test
    public void testSave() {
        BbuInfo bbuInfo = new BbuInfo();
        bbuInfo.setDeviceId("0001");
        bbuInfo.setIndicatorName("温度");
        bbuInfo.setAlarm("0");
        bbuInfo.setLevel("正常");
        bbuInfo.setValue(65.55D);
        bbuInflux.add(bbuInfo);
    }

    @Test
    public void testQuery() {
        HashMap<String, Object> map = bbuInflux.queryBbu(1L, 10L, "2020-01-01", "2022-12-31", "");
        System.out.println("Total:" + map.get("total"));
        String toJSONString = JSON.toJSONString(map.get("data"));
        List<BbuInfo> bbuInfoList = JSON.parseArray(toJSONString, BbuInfo.class);
        for (BbuInfo bbuInfo : bbuInfoList) {
            // 时间格式处理
            LocalDateTime dateTime = LocalDateTime.parse(bbuInfo.getTime(), DateTimeFormatter.ISO_OFFSET_DATE_TIME);
            String time = dateTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd hh:mm:ss"));
            bbuInfo.setTime(time);
            System.out.println("bbuInfo:" + bbuInfo);
        }
    }

    @Test
    public void testQueryLast() {
        HashMap<String, Object> map = bbuInflux.queryLast("0001");
        String toJSONString = JSON.toJSONString(map.get("data"));
        List<BbuInfo> bbuInfoList = JSON.parseArray(toJSONString, BbuInfo.class);
        for (BbuInfo bbuInfo : bbuInfoList) {
            // 时间格式处理
            LocalDateTime dateTime = LocalDateTime.parse(bbuInfo.getTime(), DateTimeFormatter.ISO_OFFSET_DATE_TIME);
            String time = dateTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd hh:mm:ss"));
            bbuInfo.setTime(time);
            System.out.println("bbuInfo:" + bbuInfo);
        }
    }
}

java">> use demo
Using database demo
> select * from bbu
name: bbu
time                alarm deviceId indicatorName level  value
----                ----- -------- ------------- ----- ------
1642948114035000000 0     0001     温度           正常   56.5
>

java">> select * from bbu   where  time>='2020-01-01' and time<='2022-12-31' order by desc limit 10 offset 0
name: bbu
time                alarm deviceId indicatorName level  value
----                ----- -------- ------------- -----  -----
1642950703148000000 0     0001     温度            正常    65.55
1642949227228000000 0     0001     温度            正常    89.56
1642948114035000000 0     0002     温度            正常    56.5

java">> select last(value),* from bbu where deviceId='0001' group by deviceId
name: bbu
tags: deviceId=0001
time                last  alarm indicatorName level  value
----                ----  ----- ------------- -----  -----
1642950703148000000 65.55 0     温度            正常   65.55
>