时序数据库 InfluxDB

目录

一、介绍

二、安装

三、inflxudb保留字

四、基本语法

1、客户端操作

1. 数据库操作

2. 数据表和数据操作

3. series 操作

4.Shard 

5. 用户操作

2、API操作

状态码

3、Java操作

五、常用函数 

六、存储策略

1.查看策略

2.创建策略

3、修改策略

4. 删除

七、目录与文件结构

八、数据备份

1、备份和恢复DB数据

一、DB备份

二、DB恢复

二、备份和恢复元数据

1、备份元数据

2、恢复元数据

---

一、介绍

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

特点：

• 专门为时间序列数据编写的自定义高性能数据存储。TSM 引擎允许高速摄取和数据压缩
• 完全用 Go 编写。它编译成一个没有外部依赖性的二进制文件。
• 简单、高性能的写入和查询 HTTP API。
• 插件支持其他数据摄取协议，例如 Graphite、collectd 和 OpenTSDB。
• 专为轻松查询聚合数据而定制的类似 SQL 的表达查询语言。
• 标签允许为系列建立索引以实现快速高效的查询。
• 保留策略有效地自动使陈旧数据过期。
• 连续查询自动计算聚合数据，使频繁查询更高效。

InfluxDB有三大特性： 
    Time Series （时间序列）：你可以使用与时间有关的相关函数（如最大，最小，求和等）        
    Metrics（度量）：你可以实时对大量数据进行计算 
    Eevents（事件）：它支持任意的事件数据

InfluxDB 提供三种操作方式：

1. 客户端命令行方式
2. HTTP API 接口
3. 各语言 API 库

InfluxDB 和传统数据库（如：MySQL）区别：

注意：​​point​​​由3部分组成​​time+fields+tags​​。

说明：

• 在插入新数据时，tag、field 和 time之间用空格分隔 。
• fields和tags key名称严格区分大小写。

​fields数据类型：

注意

        在influxdb中，字段必须存在。因为字段是没有索引的。如果使用字段作为查询条件，会扫描符合查询条件的所有字段值，性能不及tag。tags是可选的，但是强烈建议你用上它，因为tag是有索引的，tags相当于SQL中的有索引的列。tag value只能是string类型。

端口

• 8083：Web admin 管理服务的端口, http://localhost:8083
• 8086：HTTP API 的端口
• 8088：集群端口 (目前还不是很清楚, 配置在全局的 bind-address，默认不配置就是开启的)

二、安装

docker pull influxdb:1.8.3
docker run -d -p 18083:8083 -p 18088:8088 -p 18086:8086 --name my_influxdb influxdb:1.8.3
docker container update --restart=always my_influxdb

可视化工具：

1、web界面 

2、influxDB Studio 

客户端为绿色版，下载解压打开即可。

下载地址：Releases · CymaticLabs/InfluxDBStudio

案例数据：

curl https://s3.amazonaws.com/noaa.water-database/NOAA_data.txt -o NOAA_data.txt 

导入数据： 

influx -import -path=NOAA_data.txt -precision=s -database=NOAA_water_database 

三、inflxudb保留字

 

四、基本语法

docker exec -it my_influxdb  bash
cd /etc/influxdb/
influx 

1、客户端操作

1. 数据库操作

# 查看所有数据库

show databases;

# 建库

create database dbname;

# 删库

drop database daname

# 切换使用数据库

use dbname

2. 数据表和数据操作

查看所有表

show measurements

数据中的时间字段默认显示的是一个纳秒时间戳，改成可读格式。时间默认是UTC时间

precision rfc3339; # 之后再查询，时间就是rfc3339标准格式 
# 或可以在连接数据库的时候，直接带该参数
influx -precision rfc3339

查看一个measurement中所有的tag key

show tag keys

 查看一个measurement中所有的field key

show field keys

查看一个measurement中所有的保存策略(可以有多个，一个标识为default)

 show retention policies;

插入数据

        标准格式，注意在写数据的时候如果不添加时间戳，系统会默认添加一个时间。InfluxDB 中没有显式的新建表的语句，只能通过 insert 数据的方式来建立新表。

insert [,=...] =[,=...] [unix-nano-timestamp]

insert 表名,tags fileds;

#插入fields，和tags.

insert student,a=1,b=2,c=3 name="sa",age=12,emial="xxx@163.com";
#只插入​​fields​​​，可以没有​​tags
insert student name="sa",age=12,emial="xxx@163.com";

删表

drop measurement student

 查询 

SELECT field keys [time | tag kyes | * ] FROM measurements WHERE conditions GROUP BY [tag keys | time] ORDER BY time [asc | desc] LIMIT number [OFFSET number]

1. SELECT​​后面查询显示字段​必须至少有一个​​field key​​​，否则会抛异常​​ERR: at least 1 non-time field must be queried​​​。还可显示​​tag keys​​​、​​time​​​，或者​​*​​显示所有字段。
2. ​​FROM​​​查询数据来源一个或者多个​​measurement​​。
3. ​​WHERE​​​查询条件可为​​tag keys​​​和​​time​​​，​​field keys​​​ 也可作为查询条件但是不常用，因为不是索引，查询效率比较低。​​tag keys​​​条件操作符支持​​=、!=、<>、正则​​​，​​field keys​​​支持​​=、!=、<>、>、>=、<、<=、正则​​。
4. ​​GROUP BY​​​只能对​​tag keys​​​和​​time​​​进行合分组，可以多字段排序，如​​group by tag1,tag2,time​​​，也可一个​​*​​​对所有的​​tag​​​进行分组聚合（不包括​​time​​）。
5. ​​ORDER BY​​​只能对​​time​​​进行排序，​​asc​​​升序，​​desc​​降序。
6. ​​LIMIT​​​分页，​​LIMIT​​​后面的数字是查询显示多少条，​​OFFSET​​​后面数字代表偏移量（从0开始代表第一条），如​​limit 10 offset 10​​意思为从第11条开始往后10条。

正则 

#每个表输出一行（支持 Go 语言的正则表达式、支持类似于 MySQL 中的 limit 语句）

SELECT * FROM /.*/ LIMIT 1

limit

limit​​​可单独使用，也可配合​​offset​​​使用，​​offset​​偏移量的意思

select * from student limit 2;

排序

​order by​​​ 只能对​​time​​​进行排序，​​asc​​​升序，​​desc​​降序。

select * from student  order by time desc;

分组

SELECT () FROM_clause WHERE GROUP BY time(time_interval,[)] 

（1）group by tag

select * from h2o_pH group by location limit 2

（2）group by time(1m) 

        对​​time​​​分组时并不是简单的​​group by time​​​，​​time​​​后面还需要加一个分组聚合的持续时间，如​​group by time(1m)​​。支持的持续时间单位有：

（3）fill(0)填充null
        可以看到​​values​​​结果集中有​​null​​​的情况，可在查询语句中加​​fill(0)​​​，遇到​​null​​​用​​0​​​来填充。​​fill()​​中只能填数字。

条件查询

select * from student where a='1' and age=12; # a字段是tag ：字符串

select sum(age) as ageNums from student where time < now() group by a order by time desc；

格式化显示查询数据

format json
select * from student where a='1' and age=12;

模糊查询（正则表达式）

# location名称中包含coyote的数据
select * from h2o_quality where location=~/.*coyote.*/ limit 10

没有in还有or

select * from h2o_pH where location='coyote_creek' or location='santa_monica' limit 4

from多表查询 

        多表查询，以时间进行连接，不存在的值用null填充。一般情况连接的两张表tag和field上有一定的联系和比较，毫不相干的两表连接查询没什么价值。

select * from h2o_feet,h2o_pH limit 1

注意：

1.field tag是关键字 如果字段名是这两个，则需要加上单引号。

2.tags​​​之间用逗号分隔，​​fields​​之间用逗号分隔，​​​tags​​和​​fields​​之间用空格分隔​。

3.tags​​​都是字符串类型，但是不用双引号括起来；​​fields​​中有字符串类型需要用​英文双引号​括起来，如果不用英文双引号，会报语法错误​​invalid boolean​​，会认为是无效的布尔值，因为布尔类型无需加双引号。

4.tags​​​中设置布尔值就是字符串，​​fields​​​中有布尔值，可用 ​​t , T , true , True , TRUE，f , F , false , False​​表示。

3. series 操作

        series 相当于是 InfluxDB 中一些数据的集合，在同一个 database 中，retention policy、measurement、tag sets 完全相同的数据同属于一个 series，同一个 series 的数据在物理上会按照时间顺序排列存储在一起。

​​shard​​​主要由4部分组成​​Cache​​​、​​Wal​​​、​​Tsm file​​​、​​Compactor​​。

#series 表示这个表里面的数据，可以在图表上画成几条线，series 主要通过 tags 排列组合算出来。

show series from student;

4.Shard 

        shard 在 InfluxDB 中是一个比较重要的概念，它和 retention policy 相关联。每一个存储策略下会存在许多 shard，每一个 shard 存储一个指定时间段内的数据，并且不重复，例如 7点-8点 的数据落入 shard0 中，8点-9点的数据则落入 shard1 中。每一个 shard 都对应一个底层的 tsm 存储引擎，有独立的 cache、wal、tsm file。

        如果创建一个新的 retention policy 设置数据的保留时间为 1 天，则单个 shard 所存储数据的时间间隔为 1 小时，超过1个小时的数据会被存放到下一个 shard 中。

5. 用户操作

# 显示用户
SHOW USERS
# 创建用户
CREATE USER "username" WITH PASSWORD 'password'
# 创建管理员权限的用户 

#influxdb 的权限设置比较简单，只有读、写、ALL 几种。
CREATE USER "username" WITH PASSWORD 'password' WITH ALL PRIVILEGES
# 删除用户
DROP USER "username"

2、API操作

状态码

• 2xx：服务请求正常
• 4xx：代表请求语法有问题
• 5xx：服务端出问题，导致超时等故障

官方地址

3、Java操作

    org.influxdb
    influxdb-java
    2.23

spring:
  application:
    name: spring-lean-influxdb
  influx:
    url: http://192.168.56.22:18086
    user:
    password:
    database: test

import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
@Data
@Configuration
public class InfluxDBConfig {
    @Value("${spring.influx.user}")
    public String userName;
 
    @Value("${spring.influx.password}")
    public String password;
 
    @Value("${spring.influx.url}")
    public String url;
 
    //数据库
    @Value("${spring.influx.database}")
    public String database;
}

import lombok.Builder;
import lombok.Data;
import org.influxdb.annotation.Column;
import org.influxdb.annotation.Measurement;
import org.influxdb.annotation.TimeColumn;
 
import java.util.Date;
 
@Builder
@Data
@Measurement(name = "t_log")
public class LogInfo {
    // Column中的name为measurement中的列名
    // 此外,需要注意InfluxDB中时间戳均是以UTC时保存,在保存以及提取过程中需要注意时区转换
    @Column(name = "time")
    private String time;
    // 注解中添加tag = true,表示当前字段内容为tag内容
    @Column(name = "module", tag = true)
    private String module;
    @Column(name = "level", tag = true)
    private String level;
    @Column(name = "device_id", tag = true)
    private String deviceId;
    @Column(name = "msg")
    private String msg;
    @TimeColumn
    private Date createTime;
}

import com.lean.influxdb.config.InfluxDBConfig;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
 
import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
 
@Service
public class InfluxDBService {
 
    @Autowired
    private InfluxDBConfig influxDBConfig;
 
    @PostConstruct
    public void initInfluxDb() {
        this.retentionPolicy = retentionPolicy == null || "".equals(retentionPolicy) ? "autogen" : retentionPolicy;
        this.influxDB = influxDbBuild();
    }
 
    //保留策略
    private String retentionPolicy;
 
    private InfluxDB influxDB;
 
 
    /**
     * 连接时序数据库；获得InfluxDB
     **/
    private InfluxDB influxDbBuild() {
        if (influxDB == null) {
            if(StringUtils.isEmpty(influxDBConfig.userName)){
                influxDB = InfluxDBFactory.connect(influxDBConfig.url);
            }else {
                influxDB = InfluxDBFactory.connect(influxDBConfig.url, influxDBConfig.userName, influxDBConfig.password);
            }
            influxDB.setDatabase(influxDBConfig.database);
        }
        return influxDB;
    }
 
    /**
     * 插入
     */
    public void insertPoint(Point point) {
        influxDbBuild();
        influxDB.write(point);
    }
 
    /**
     * 插入
     *
     * @param measurement 表
     * @param tags        标签
     * @param fields      字段
     */
    public void insert(String measurement, Map tags, Map fields) {
        influxDbBuild();
        Point.Builder builder = Point.measurement(measurement);
        builder.time(System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        builder.tag(tags);
        builder.fields(fields);
        influxDB.write(influxDBConfig.database, "", builder.build());
    }
 
    /**
     * @param measurement
     * @param time
     * @param tags
     * @param fields
     * @return void
     * @desc 插入, 带时间time
     */
    public void insert(String measurement, long time, Map tags, Map fields) {
        influxDbBuild();
        Point.Builder builder = Point.measurement(measurement);
        builder.time(time, TimeUnit.MILLISECONDS);
        builder.tag(tags);
        builder.fields(fields);
        influxDB.write(influxDBConfig.database, "", builder.build());
    }
    /**
     * 查询
     *
     * @param command 查询语句
     * @return
     */
    public QueryResult query(String command) {
        influxDbBuild();
        return influxDB.query(new Query(command, influxDBConfig.database));
    }
 
    /**
     * @param queryResult
     * @desc 查询结果处理
     */
    public List> queryResultProcess(QueryResult queryResult) {
        List> mapList = new ArrayList<>();
        List resultList = queryResult.getResults();
        //把查询出的结果集转换成对应的实体对象，聚合成list
        for (QueryResult.Result query : resultList) {
            List seriesList = query.getSeries();
            if (seriesList != null && seriesList.size() != 0) {
                for (QueryResult.Series series : seriesList) {
                    List columns = series.getColumns();
                    String[] keys = columns.toArray(new String[columns.size()]);
                    List> values = series.getValues();
                    if (values != null && values.size() != 0) {
                        for (List