国产时序数据库IotDB安装、与SpringBoot集成

一.简介：

本文将完成一个真实业务中的设备上报数据的一个例子，完整的展示后台服务接收到设备上报的数据后，将数据添加到时序数据库，并且将数据查询出来的一个例子。本文所有代码已经上传GitHub：https://github.com/Tom-shushu/work-study 下的 iotdb-demo 下。

IoTDB 是针对时间序列数据收集、存储与分析一体化的数据管理引擎。它具有体量轻、性能高、易使用的特点，完美对接 Hadoop 与 Spark 生态，适用于工业物联网应用中海量时间序列数据高速写入和复杂分析查询的需求。

我的理解：它就是一个树形结构的数据库可以很灵活的查询各个级下面的数据，因为它特殊的数据结构也使得它的查询效率会更高一些。

二.Docker安装IotDB:

1.拉取镜像（使用0.13,在使用的过程中0.14在查询时出现了问题）

1. docker pull apache/iotdb:0.13.1-node

复制代码

2.创建数据文件和日志的 docker 挂载目录 (docker volume)

1. docker volume create mydata
2. docker volume create mylogs

复制代码

3.直接运行镜像

1. docker run --name iotdb  -p 6667:6667 -v mydata:/iotdb/data -v mylogs:/iotdb/logs -d apache/iotdb:0.13.1-node /iotdb/bin/start-server.sh

复制代码

4.进入镜像并且登录IotDB

1. docker exec  -it iotdb  /bin/bash
2. /iotdb/sbin/start-cli.sh -h localhost -p 6667 -u root -pw root

复制代码

这样就算安装完成，然后打开服务器6667安全组

三.IotDB与SpringBoot集成

1.引入必要的依赖

1. 　　　　<dependency>
2.             <groupId>org.apache.iotdb</groupId>
3.             <artifactId>iotdb-session</artifactId>
4.             <version>0.14.0-preview1</version>
5.         </dependency>
7.         <dependency>
8.             <groupId>cn.hutool</groupId>
9.             <artifactId>hutool-all</artifactId>
10.             <version>5.6.3</version>
11.         </dependency>
13.         <dependency>
14.             <groupId>com.alibaba</groupId>
15.             <artifactId>fastjson</artifactId>
16.             <version>1.2.83</version>
17.         </dependency>
19.         <dependency>
20.             <groupId>org.springframework.boot</groupId>
21.             <artifactId>spring-boot-starter-web</artifactId>
22.         </dependency>
24.         <dependency>
25.             <groupId>org.projectlombok</groupId>
26.             <artifactId>lombok</artifactId>
27.             <optional>true</optional>
28.         </dependency>
29.         <dependency>
30.             <groupId>org.springframework.boot</groupId>
31.             <artifactId>spring-boot-starter-test</artifactId>
32.             <scope>test</scope>
33.             <exclusions>
34.                 <exclusion>
35.                     <groupId>org.junit.vintage</groupId>
36.                     <artifactId>junit-vintage-engine</artifactId>
37.                 </exclusion>
38.             </exclusions>
39.         </dependency>

复制代码

2.编写配置类并且封装对应的方法 IotDBSessionConfig

1. package com.zhouhong.iotdbdemo.config;
3. import lombok.extern.log4j.Log4j2;
4. import org.apache.iotdb.rpc.IoTDBConnectionException;
5. import org.apache.iotdb.rpc.StatementExecutionException;
6. import org.apache.iotdb.session.Session;
7. import org.apache.iotdb.session.SessionDataSet;
8. import org.apache.iotdb.session.util.Version;
9. import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
10. import org.apache.iotdb.tsfile.write.record.Tablet;
11. import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
12. import org.springframework.context.annotation.Bean;
13. import org.springframework.context.annotation.Configuration;
14. import org.springframework.stereotype.Component;
16. import java.rmi.ServerException;
17. import java.util.ArrayList;
18. import java.util.List;
20. /**
21. * description: iotdb 配置工具类（常用部分，如需要可以自行扩展）
22. * 注意：可以不需要创建分组，插入时默认前两个节点名称为分组名称 比如： root.a1eaKSRpRty.CA3013A303A25467 或者
23. * root.a1eaKSRpRty.CA3013A303A25467.heart  他们的分组都为 root.a1eaKSRpRty
24. * author: zhouhong
25. */
26. @Log4j2
27. @Component
28. @Configuration
29. public class IotDBSessionConfig {
31.     private static Session session;
32.     private static final String LOCAL_HOST = "XXX.XX.XXX.XX";
33.     @Bean
34.     public Session getSession() throws IoTDBConnectionException, StatementExecutionException {
35.         if (session == null) {
36.             log.info("正在连接iotdb.......");
37.             session = new Session.Builder().host(LOCAL_HOST).port(6667).username("root").password("root").version(Version.V_0_13).build();
38.             session.open(false);
39.             session.setFetchSize(100);
40.             log.info("iotdb连接成功~");
41.             // 设置时区
42.             session.setTimeZone("+08:00");
43.         }
44.         return session;
45.     }
47.     /**
48.      * description: 带有数据类型的添加操作 - insertRecord没有指定类型
49.      * author: zhouhong
50.      * @param  * @param deviceId:节点路径如：root.a1eaKSRpRty.CA3013A303A25467
51.      *                  time:时间戳
52.      *                  measurementsList：物理量 即：属性
53.      *                  type：数据类型： BOOLEAN((byte)0), INT32((byte)1),INT64((byte)2),FLOAT((byte)3),DOUBLE((byte)4),TEXT((byte)5),VECTOR((byte)6);
54.      *                  valuesList：属性值 --- 属性必须与属性值一一对应
55.      * @return
56.      */
57.     public void insertRecordType(String deviceId, Long time,List<String>  measurementsList, TSDataType type,List<Object> valuesList) throws StatementExecutionException, IoTDBConnectionException, ServerException {
58.         if (measurementsList.size() != valuesList.size()) {
59.             throw new ServerException("measurementsList 与 valuesList 值不对应");
60.         }
61.         List<TSDataType> types = new ArrayList<>();
62.         measurementsList.forEach(item -> {
63.             types.add(type);
64.         });
65.         session.insertRecord(deviceId, time, measurementsList, types, valuesList);
66.     }
67.     /**
68.      * description: 带有数据类型的添加操作 - insertRecord没有指定类型
69.      * author: zhouhong
70.      * @param  deviceId:节点路径如：root.a1eaKSRpRty.CA3013A303A25467
71.      * @param  time:时间戳
72.      * @param  measurementsList：物理量 即：属性
73.      * @param  valuesList：属性值 --- 属性必须与属性值一一对应
74.      * @return
75.      */
76.     public void insertRecord(String deviceId, Long time,List<String>  measurementsList, List<String> valuesList) throws StatementExecutionException, IoTDBConnectionException, ServerException {
77.         if (measurementsList.size() == valuesList.size()) {
78.             session.insertRecord(deviceId, time, measurementsList, valuesList);
79.         } else {
80.             log.error("measurementsList 与 valuesList 值不对应");
81.         }
82.     }
83.     /**
84.      * description: 批量插入
85.      * author: zhouhong
86.      */
87.     public void insertRecords(List<String> deviceIdList, List<Long> timeList, List<List<String>> measurementsList, List<List<String>> valuesList) throws StatementExecutionException, IoTDBConnectionException, ServerException {
88.         if (measurementsList.size() == valuesList.size()) {
89.             session.insertRecords(deviceIdList, timeList, measurementsList, valuesList);
90.         } else {
91.             log.error("measurementsList 与 valuesList 值不对应");
92.         }
93.     }
95.     /**
96.      * description: 插入操作
97.      * author: zhouhong
98.      * @param  deviceId:节点路径如：root.a1eaKSRpRty.CA3013A303A25467
99.      *  @param  time:时间戳
100.      *  @param  schemaList: 属性值 + 数据类型 例子： List<MeasurementSchema> schemaList = new ArrayList<>();  schemaList.add(new MeasurementSchema("breath", TSDataType.INT64));
101.      *  @param  maxRowNumber：
102.      * @return
103.      */
104.     public void insertTablet(String deviceId,  Long time,List<MeasurementSchema> schemaList, List<Object> valueList,int maxRowNumber) throws StatementExecutionException, IoTDBConnectionException {
106.         Tablet tablet = new Tablet(deviceId, schemaList, maxRowNumber);
107.         // 向iotdb里面添加数据
108.         int rowIndex = tablet.rowSize++;
109.         tablet.addTimestamp(rowIndex, time);
110.         for (int i = 0; i < valueList.size(); i++) {
111.             tablet.addValue(schemaList.get(i).getMeasurementId(), rowIndex, valueList.get(i));
112.         }
113.         if (tablet.rowSize == tablet.getMaxRowNumber()) {
114.             session.insertTablet(tablet, true);
115.             tablet.reset();
116.         }
117.         if (tablet.rowSize != 0) {
118.             session.insertTablet(tablet);
119.             tablet.reset();
120.         }
121.     }
123.     /**
124.      * description: 根据SQL查询
125.      * author: zhouhong
126.      */
127.     public SessionDataSet query(String sql) throws StatementExecutionException, IoTDBConnectionException {
128.         return session.executeQueryStatement(sql);
129.     }
131.     /**
132.      * description: 删除分组 如 root.a1eaKSRpRty
133.      * author: zhouhong
134.      * @param  groupName：分组名称
135.      * @return
136.      */
137.     public void deleteStorageGroup(String groupName) throws StatementExecutionException, IoTDBConnectionException {
138.         session.deleteStorageGroup(groupName);
139.     }
141.     /**
142.      * description: 根据Timeseries删除  如：root.a1eaKSRpRty.CA3013A303A25467.breath  （个人理解：为具体的物理量）
143.      * author: zhouhong
144.      */
145.     public void deleteTimeseries(String timeseries) throws StatementExecutionException, IoTDBConnectionException {
146.         session.deleteTimeseries(timeseries);
147.     }
148.     /**
149.      * description: 根据Timeseries批量删除
150.      * author: zhouhong
151.      */
152.     public void deleteTimeserieList(List<String> timeseriesList) throws StatementExecutionException, IoTDBConnectionException {
153.         session.deleteTimeseries(timeseriesList);
154.     }
156.     /**
157.      * description: 根据分组批量删除
158.      * author: zhouhong
159.      */
160.     public void deleteStorageGroupList(List<String> storageGroupList) throws StatementExecutionException, IoTDBConnectionException {
161.         session.deleteStorageGroups(storageGroupList);
162.     }
164.     /**
165.      * description: 根据路径和结束时间删除 结束时间之前的所有数据
166.      * author: zhouhong
167.      */
168.     public void deleteDataByPathAndEndTime(String path, Long endTime) throws StatementExecutionException, IoTDBConnectionException {
169.         session.deleteData(path, endTime);
170.     }
171.     /**
172.      * description: 根据路径集合和结束时间批量删除 结束时间之前的所有数据
173.      * author: zhouhong
174.      */
175.     public void deleteDataByPathListAndEndTime(List<String> pathList, Long endTime) throws StatementExecutionException, IoTDBConnectionException {
176.         session.deleteData(pathList, endTime);
177.     }
178.     /**
179.      * description: 根据路径集合和时间段批量删除
180.      * author: zhouhong
181.      */
182.     public void deleteDataByPathListAndTime(List<String> pathList, Long startTime,Long endTime) throws StatementExecutionException, IoTDBConnectionException {
183.         session.deleteData(pathList, startTime, endTime);
184.     }
186. }

复制代码

3.入参

1. package com.zhouhong.iotdbdemo.model.param;
3. import lombok.Data;
4. /**
5. * description: 入参
6. * date: 2022/8/15 21:53
7. * author: zhouhong
8. */
9. @Data
10. public class IotDbParam {
11.     /***
12.      * 产品PK
13.      */
14.     private  String  pk;
15.     /***
16.      * 设备号
17.      */
18.     private  String  sn;
19.     /***
20.      * 时间
21.      */
22.     private Long time;
23.     /***
24.      * 实时呼吸
25.      */
26.     private String breath;
27.     /***
28.      * 实时心率
29.      */
30.     private String heart;
31.     /***
32.      * 查询开始时间
33.      */
34.     private String startTime;
35.     /***
36.      * 查询结束时间
37.      */
38.     private String endTime;
40. }

复制代码

4.返回参数

1. package com.zhouhong.iotdbdemo.model.result;
3. import lombok.Data;
5. /**
6. * description: 返回结果
7. * date: 2022/8/15 21:56
8. * author: zhouhong
9. */
10. @Data
11. public class IotDbResult {
12.     /***
13.      * 时间
14.      */
15.     private String time;
16.     /***
17.      * 产品PK
18.      */
19.     private  String  pk;
20.     /***
21.      * 设备号
22.      */
23.     private  String  sn;
24.     /***
25.      * 实时呼吸
26.      */
27.     private String breath;
28.     /***
29.      * 实时心率
30.      */
31.     private String heart;
33. }

复制代码

5.使用

1. package com.zhouhong.iotdbdemo.server.impl;
3. import com.zhouhong.iotdbdemo.config.IotDBSessionConfig;
4. import com.zhouhong.iotdbdemo.model.param.IotDbParam;
5. import com.zhouhong.iotdbdemo.model.result.IotDbResult;
6. import com.zhouhong.iotdbdemo.server.IotDbServer;
7. import lombok.extern.log4j.Log4j2;
8. import org.apache.iotdb.rpc.IoTDBConnectionException;
9. import org.apache.iotdb.rpc.StatementExecutionException;
10. import org.apache.iotdb.session.SessionDataSet;
11. import org.apache.iotdb.tsfile.read.common.Field;
12. import org.apache.iotdb.tsfile.read.common.RowRecord;
13. import org.springframework.stereotype.Service;
15. import javax.annotation.Resource;
16. import java.rmi.ServerException;
17. import java.util.ArrayList;
18. import java.util.HashMap;
19. import java.util.List;
20. import java.util.Map;
22. /**
23. * description: iot服务实现类
24. * date: 2022/8/15 9:43
25. * author: zhouhong
26. */
28. @Log4j2
29. @Service
30. public class IotDbServerImpl implements IotDbServer {
32.     @Resource
33.     private IotDBSessionConfig iotDBSessionConfig;
35.     @Override
36.     public void insertData(IotDbParam iotDbParam) throws StatementExecutionException, ServerException, IoTDBConnectionException {
37.         // iotDbParam: 模拟设备上报消息
38.         // bizkey: 业务唯一key  PK :产品唯一编码   SN:设备唯一编码
39.         String deviceId = "root.bizkey."+ iotDbParam.getPk() + "." + iotDbParam.getSn();
40.         // 将设备上报的数据存入数据库(时序数据库)
41.         List<String> measurementsList = new ArrayList<>();
42.         measurementsList.add("heart");
43.         measurementsList.add("breath");
44.         List<String> valuesList = new ArrayList<>();
45.         valuesList.add(String.valueOf(iotDbParam.getHeart()));
46.         valuesList.add(String.valueOf(iotDbParam.getBreath()));
47.         iotDBSessionConfig.insertRecord(deviceId, iotDbParam.getTime(), measurementsList, valuesList);
48.     }
50.     @Override
51.     public List<IotDbResult> queryDataFromIotDb(IotDbParam iotDbParam) throws Exception {
52.         List<IotDbResult> iotDbResultList = new ArrayList<>();
54.         if (null != iotDbParam.getPk() && null != iotDbParam.getSn()) {
55.             String sql = "select * from root.bizkey."+ iotDbParam.getPk() +"." + iotDbParam.getSn() + " where time >= "
56.                     + iotDbParam.getStartTime() + " and time < " + iotDbParam.getEndTime();
57.             SessionDataSet sessionDataSet = iotDBSessionConfig.query(sql);
58.             List<String> columnNames = sessionDataSet.getColumnNames();
59.             List<String> titleList = new ArrayList<>();
60.             // 排除Time字段 -- 方便后面后面拼装数据
61.             for (int i = 1; i < columnNames.size(); i++) {
62.                 String[] temp = columnNames.get(i).split("\\.");
63.                 titleList.add(temp[temp.length - 1]);
64.             }
65.             // 封装处理数据
66.             packagingData(iotDbParam, iotDbResultList, sessionDataSet, titleList);
67.         } else {
68.             log.info("PK或者SN不能为空！！");
69.         }
70.         return iotDbResultList;
71.     }
72.     /**
73.      * 封装处理数据
74.      * @param iotDbParam
75.      * @param iotDbResultList
76.      * @param sessionDataSet
77.      * @param titleList
78.      * @throws StatementExecutionException
79.      * @throws IoTDBConnectionException
80.      */
81.     private void packagingData(IotDbParam iotDbParam, List<IotDbResult> iotDbResultList, SessionDataSet sessionDataSet, List<String> titleList)
82.             throws StatementExecutionException, IoTDBConnectionException {
83.         int fetchSize = sessionDataSet.getFetchSize();
84.         if (fetchSize > 0) {
85.             while (sessionDataSet.hasNext()) {
86.                 IotDbResult iotDbResult = new IotDbResult();
87.                 RowRecord next = sessionDataSet.next();
88.                 List<Field> fields = next.getFields();
89.                 String timeString = new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(next.getTimestamp());
90.                 iotDbResult.setTime(timeString);
91.                 Map<String, String> map = new HashMap<>();
93.                 for (int i = 0; i < fields.size(); i++) {
94.                     Field field = fields.get(i);
95.                     // 这里的需要按照类型获取
96.                     map.put(titleList.get(i), field.getObjectValue(field.getDataType()).toString());
97.                 }
98.                 iotDbResult.setTime(timeString);
99.                 iotDbResult.setPk(iotDbParam.getPk());
100.                 iotDbResult.setSn(iotDbParam.getSn());
101.                 iotDbResult.setHeart(map.get("heart"));
102.                 iotDbResult.setBreath(map.get("breath"));
103.                 iotDbResultList.add(iotDbResult);
104.             }
105.         }
106.     }
107. }

复制代码

6.控制层

1. package com.zhouhong.iotdbdemo.controller;
3. import com.zhouhong.iotdbdemo.config.IotDBSessionConfig;
4. import com.zhouhong.iotdbdemo.model.param.IotDbParam;
5. import com.zhouhong.iotdbdemo.response.ResponseData;
6. import com.zhouhong.iotdbdemo.server.IotDbServer;
7. import lombok.extern.log4j.Log4j2;
8. import org.apache.iotdb.rpc.IoTDBConnectionException;
9. import org.apache.iotdb.rpc.StatementExecutionException;
10. import org.springframework.web.bind.annotation.*;
12. import javax.annotation.Resource;
13. import java.rmi.ServerException;
15. /**
16. * description: iotdb 控制层
17. * date: 2022/8/15 21:50
18. * author: zhouhong
19. */
20. @Log4j2
21. @RestController
22. public class IotDbController {
24.     @Resource
25.     private IotDbServer iotDbServer;
26.     @Resource
27.     private IotDBSessionConfig iotDBSessionConfig;
29.     /**
30.      * 插入数据
31.      * @param iotDbParam
32.      */
33.     @PostMapping("/api/device/insert")
34.     public ResponseData insert(@RequestBody IotDbParam iotDbParam) throws StatementExecutionException, ServerException, IoTDBConnectionException {
35.         iotDbServer.insertData(iotDbParam);
36.         return ResponseData.success();
37.     }
39.     /**
40.      * 插入数据
41.      * @param iotDbParam
42.      */
43.     @PostMapping("/api/device/queryData")
44.     public ResponseData queryDataFromIotDb(@RequestBody IotDbParam iotDbParam) throws Exception {
45.         return ResponseData.success(iotDbServer.queryDataFromIotDb(iotDbParam));
46.     }
48.     /**
49.      * 删除分组
50.      * @return
51.      */
52.     @PostMapping("/api/device/deleteGroup")
53.     public ResponseData deleteGroup() throws StatementExecutionException, IoTDBConnectionException {
54.         iotDBSessionConfig.deleteStorageGroup("root.a1eaKSRpRty");
55.         iotDBSessionConfig.deleteStorageGroup("root.smartretirement");
56.         return ResponseData.success();
57.     }
59. }

复制代码

四.PostMan测试

1.添加一条记录

接口：localhost:8080/api/device/insert

入参：

1. {
2.     "time":1660573444672,
3.     "pk":"a1TTQK9TbKT",
4.     "sn":"SN202208120945QGJLD",
5.     "breath":"17",
6.     "heart":"68"
7. }

复制代码

 
 
 查看IotDB数据


 
 
 2.根据SQL查询时间区间记录(其他查询以此类推)

接口：localhost:8080/api/device/queryData

入参：

1. {
2.     "pk":"a1TTQK9TbKT",
3.     "sn":"SN202208120945QGJLD",
4.     "startTime":"2022-08-14 00:00:00",
5.     "endTime":"2022-08-16 00:00:00"
6. }

复制代码

结果：

1. {
2.     "success": true,
3.     "code": 200,
4.     "message": "请求成功",
5.     "localizedMsg": "请求成功",
6.     "data": [
7.         {
8.             "time": "2022-08-15 22:24:04",
9.             "pk": "a1TTQK9TbKT",
10.             "sn": "SN202208120945QGJLD",
11.             "breath": "19.0",
12.             "heart": "75.0"
13.         },
14.         {
15.             "time": "2022-08-15 22:24:04",
16.             "pk": "a1TTQK9TbKT",
17.             "sn": "SN202208120945QGJLD",
18.             "breath": "20.0",
19.             "heart": "78.0"
20.         },
21.         {
22.             "time": "2022-08-15 22:24:04",
23.             "pk": "a1TTQK9TbKT",
24.             "sn": "SN202208120945QGJLD",
25.             "breath": "17.0",
26.             "heart": "68.0"
27.         }
28.     ]
29. }

复制代码

IotDB还支持分页、聚合等等其他操作，详细信息可以参考 https://iotdb.apache.org/zh/UserGuide/Master/Query-Data/Overview.html


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！