國產時序資料庫IotDB安裝、與SpringBoot集成
一.簡介：
本文將完成一個真實業務中的設備上報數據的一個例子，完整的展示後台服務接收到設備上報的數據後，將數據添加到時序資料庫，並且將數據查詢出來的一個例子。本文所有程式碼已經上傳GitHub：//github.com/Tom-shushu/work-study 下的 iotdb-demo 下。
IoTDB 是針對時間序列數據收集、存儲與分析一體化的數據管理引擎。它具有體量輕、性能高、易使用的特點，完美對接 Hadoop 與 Spark 生態，適用於工業物聯網應用中海量時間序列數據高速寫入和複雜分析查詢的需求。
我的理解：它就是一個樹形結構的資料庫可以很靈活的查詢各個級下面的數據，因為它特殊的數據結構也使得它的查詢效率會更高一些。
二.Docker安裝IotDB:
1.拉取鏡像（使用0.13,在使用的過程中0.14在查詢時出現了問題）
docker pull apache/iotdb:0.13.1-node
2.創建數據文件和日誌的 docker 掛載目錄 (docker volume)
docker volume create mydata
docker volume create mylogs
3.直接運行鏡像
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
docker exec -it iotdb /bin/bash
/iotdb/sbin/start-cli.sh -h localhost -p 6667 -u root -pw root
這樣就算安裝完成，然後打開伺服器6667安全組
三.IotDB與SpringBoot集成
1.引入必要的依賴
<dependency>
<groupId>org.apache.iotdb</groupId>
<artifactId>iotdb-session</artifactId>
<version>0.14.0-preview1</version>
</dependency>
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-all</artifactId>
<version>5.6.3</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.83</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>org.junit.vintage</groupId>
<artifactId>junit-vintage-engine</artifactId>
</exclusion>
</exclusions>
</dependency>
2.編寫配置類並且封裝對應的方法 IotDBSessionConfig
package com.zhouhong.iotdbdemo.config;
import lombok.extern.log4j.Log4j2;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.Session;
import org.apache.iotdb.session.SessionDataSet;
import org.apache.iotdb.session.util.Version;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.write.record.Tablet;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;
import java.rmi.ServerException;
import java.util.ArrayList;
import java.util.List;
/**
* description: iotdb 配置工具類（常用部分，如需要可以自行擴展）
* 注意：可以不需要創建分組，插入時默認前兩個節點名稱為分組名稱 比如： root.a1eaKSRpRty.CA3013A303A25467 或者
* root.a1eaKSRpRty.CA3013A303A25467.heart 他們的分組都為 root.a1eaKSRpRty
* author: zhouhong
*/
@Log4j2
@Component
@Configuration
public class IotDBSessionConfig {
private static Session session;
private static final String LOCAL_HOST = "XXX.XX.XXX.XX";
@Bean
public Session getSession() throws IoTDBConnectionException, StatementExecutionException {
if (session == null) {
log.info("正在連接iotdb.......");
session = new Session.Builder().host(LOCAL_HOST).port(6667).username("root").password("root").version(Version.V_0_13).build();
session.open(false);
session.setFetchSize(100);
log.info("iotdb連接成功~");
// 設置時區
session.setTimeZone("+08:00");
}
return session;
}
/**
* description: 帶有數據類型的添加操作 - insertRecord沒有指定類型
* author: zhouhong
* @param * @param deviceId:節點路徑如：root.a1eaKSRpRty.CA3013A303A25467
* time:時間戳
* measurementsList：物理量 即：屬性
* type：數據類型： BOOLEAN((byte)0), INT32((byte)1),INT64((byte)2),FLOAT((byte)3),DOUBLE((byte)4),TEXT((byte)5),VECTOR((byte)6);
* valuesList：屬性值 --- 屬性必須與屬性值一一對應
* @return
*/
public void insertRecordType(String deviceId, Long time,List<String> measurementsList, TSDataType type,List<Object> valuesList) throws StatementExecutionException, IoTDBConnectionException, ServerException {
if (measurementsList.size() != valuesList.size()) {
throw new ServerException("measurementsList 與 valuesList 值不對應");
}
List<TSDataType> types = new ArrayList<>();
measurementsList.forEach(item -> {
types.add(type);
});
session.insertRecord(deviceId, time, measurementsList, types, valuesList);
}
/**
* description: 帶有數據類型的添加操作 - insertRecord沒有指定類型
* author: zhouhong
* @param deviceId:節點路徑如：root.a1eaKSRpRty.CA3013A303A25467
* @param time:時間戳
* @param measurementsList：物理量 即：屬性
* @param valuesList：屬性值 --- 屬性必須與屬性值一一對應
* @return
*/
public void insertRecord(String deviceId, Long time,List<String> measurementsList, List<String> valuesList) throws StatementExecutionException, IoTDBConnectionException, ServerException {
if (measurementsList.size() == valuesList.size()) {
session.insertRecord(deviceId, time, measurementsList, valuesList);
} else {
log.error("measurementsList 與 valuesList 值不對應");
}
}
/**
* description: 批量插入
* author: zhouhong
*/
public void insertRecords(List<String> deviceIdList, List<Long> timeList, List<List<String>> measurementsList, List<List<String>> valuesList) throws StatementExecutionException, IoTDBConnectionException, ServerException {
if (measurementsList.size() == valuesList.size()) {
session.insertRecords(deviceIdList, timeList, measurementsList, valuesList);
} else {
log.error("measurementsList 與 valuesList 值不對應");
}
}
/**
* description: 插入操作
* author: zhouhong
* @param deviceId:節點路徑如：root.a1eaKSRpRty.CA3013A303A25467
* @param time:時間戳
* @param schemaList: 屬性值 + 數據類型 例子： List<MeasurementSchema> schemaList = new ArrayList<>(); schemaList.add(new MeasurementSchema("breath", TSDataType.INT64));
* @param maxRowNumber：
* @return
*/
public void insertTablet(String deviceId, Long time,List<MeasurementSchema> schemaList, List<Object> valueList,int maxRowNumber) throws StatementExecutionException, IoTDBConnectionException {
Tablet tablet = new Tablet(deviceId, schemaList, maxRowNumber);
// 向iotdb裡面添加數據
int rowIndex = tablet.rowSize++;
tablet.addTimestamp(rowIndex, time);
for (int i = 0; i < valueList.size(); i++) {
tablet.addValue(schemaList.get(i).getMeasurementId(), rowIndex, valueList.get(i));
}
if (tablet.rowSize == tablet.getMaxRowNumber()) {
session.insertTablet(tablet, true);
tablet.reset();
}
if (tablet.rowSize != 0) {
session.insertTablet(tablet);
tablet.reset();
}
}
/**
* description: 根據SQL查詢
* author: zhouhong
*/
public SessionDataSet query(String sql) throws StatementExecutionException, IoTDBConnectionException {
return session.executeQueryStatement(sql);
}
/**
* description: 刪除分組 如 root.a1eaKSRpRty
* author: zhouhong
* @param groupName：分組名稱
* @return
*/
public void deleteStorageGroup(String groupName) throws StatementExecutionException, IoTDBConnectionException {
session.deleteStorageGroup(groupName);
}
/**
* description: 根據Timeseries刪除 如：root.a1eaKSRpRty.CA3013A303A25467.breath （個人理解：為具體的物理量）
* author: zhouhong
*/
public void deleteTimeseries(String timeseries) throws StatementExecutionException, IoTDBConnectionException {
session.deleteTimeseries(timeseries);
}
/**
* description: 根據Timeseries批量刪除
* author: zhouhong
*/
public void deleteTimeserieList(List<String> timeseriesList) throws StatementExecutionException, IoTDBConnectionException {
session.deleteTimeseries(timeseriesList);
}
/**
* description: 根據分組批量刪除
* author: zhouhong
*/
public void deleteStorageGroupList(List<String> storageGroupList) throws StatementExecutionException, IoTDBConnectionException {
session.deleteStorageGroups(storageGroupList);
}
/**
* description: 根據路徑和結束時間刪除 結束時間之前的所有數據
* author: zhouhong
*/
public void deleteDataByPathAndEndTime(String path, Long endTime) throws StatementExecutionException, IoTDBConnectionException {
session.deleteData(path, endTime);
}
/**
* description: 根據路徑集合和結束時間批量刪除 結束時間之前的所有數據
* author: zhouhong
*/
public void deleteDataByPathListAndEndTime(List<String> pathList, Long endTime) throws StatementExecutionException, IoTDBConnectionException {
session.deleteData(pathList, endTime);
}
/**
* description: 根據路徑集合和時間段批量刪除
* author: zhouhong
*/
public void deleteDataByPathListAndTime(List<String> pathList, Long startTime,Long endTime) throws StatementExecutionException, IoTDBConnectionException {
session.deleteData(pathList, startTime, endTime);
}
}
3.入參
package com.zhouhong.iotdbdemo.model.param;
import lombok.Data;
/**
* description: 入參
* date: 2022/8/15 21:53
* author: zhouhong
*/
@Data
public class IotDbParam {
/***
* 產品PK
*/
private String pk;
/***
* 設備號
*/
private String sn;
/***
* 時間
*/
private Long time;
/***
* 實時呼吸
*/
private String breath;
/***
* 實時心率
*/
private String heart;
/***
* 查詢開始時間
*/
private String startTime;
/***
* 查詢結束時間
*/
private String endTime;
}
4.返回參數
package com.zhouhong.iotdbdemo.model.result;
import lombok.Data;
/**
* description: 返回結果
* date: 2022/8/15 21:56
* author: zhouhong
*/
@Data
public class IotDbResult {
/***
* 時間
*/
private String time;
/***
* 產品PK
*/
private String pk;
/***
* 設備號
*/
private String sn;
/***
* 實時呼吸
*/
private String breath;
/***
* 實時心率
*/
private String heart;
}
5.使用
package com.zhouhong.iotdbdemo.server.impl;
import com.zhouhong.iotdbdemo.config.IotDBSessionConfig;
import com.zhouhong.iotdbdemo.model.param.IotDbParam;
import com.zhouhong.iotdbdemo.model.result.IotDbResult;
import com.zhouhong.iotdbdemo.server.IotDbServer;
import lombok.extern.log4j.Log4j2;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.SessionDataSet;
import org.apache.iotdb.tsfile.read.common.Field;
import org.apache.iotdb.tsfile.read.common.RowRecord;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.rmi.ServerException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* description: iot服務實現類
* date: 2022/8/15 9:43
* author: zhouhong
*/
@Log4j2
@Service
public class IotDbServerImpl implements IotDbServer {
@Resource
private IotDBSessionConfig iotDBSessionConfig;
@Override
public void insertData(IotDbParam iotDbParam) throws StatementExecutionException, ServerException, IoTDBConnectionException {
// iotDbParam: 模擬設備上報消息
// bizkey: 業務唯一key PK :產品唯一編碼 SN:設備唯一編碼
String deviceId = "root.bizkey."+ iotDbParam.getPk() + "." + iotDbParam.getSn();
// 將設備上報的數據存入資料庫(時序資料庫)
List<String> measurementsList = new ArrayList<>();
measurementsList.add("heart");
measurementsList.add("breath");
List<String> valuesList = new ArrayList<>();
valuesList.add(String.valueOf(iotDbParam.getHeart()));
valuesList.add(String.valueOf(iotDbParam.getBreath()));
iotDBSessionConfig.insertRecord(deviceId, iotDbParam.getTime(), measurementsList, valuesList);
}
@Override
public List<IotDbResult> queryDataFromIotDb(IotDbParam iotDbParam) throws Exception {
List<IotDbResult> iotDbResultList = new ArrayList<>();
if (null != iotDbParam.getPk() && null != iotDbParam.getSn()) {
String sql = "select * from root.bizkey."+ iotDbParam.getPk() +"." + iotDbParam.getSn() + " where time >= "
+ iotDbParam.getStartTime() + " and time < " + iotDbParam.getEndTime();
SessionDataSet sessionDataSet = iotDBSessionConfig.query(sql);
List<String> columnNames = sessionDataSet.getColumnNames();
List<String> titleList = new ArrayList<>();
// 排除Time欄位 -- 方便後面後面拼裝數據
for (int i = 1; i < columnNames.size(); i++) {
String[] temp = columnNames.get(i).split("\\.");
titleList.add(temp[temp.length - 1]);
}
// 封裝處理數據
packagingData(iotDbParam, iotDbResultList, sessionDataSet, titleList);
} else {
log.info("PK或者SN不能為空！！");
}
return iotDbResultList;
}
/**
* 封裝處理數據
* @param iotDbParam
* @param iotDbResultList
* @param sessionDataSet
* @param titleList
* @throws StatementExecutionException
* @throws IoTDBConnectionException
*/
private void packagingData(IotDbParam iotDbParam, List<IotDbResult> iotDbResultList, SessionDataSet sessionDataSet, List<String> titleList)
throws StatementExecutionException, IoTDBConnectionException {
int fetchSize = sessionDataSet.getFetchSize();
if (fetchSize > 0) {
while (sessionDataSet.hasNext()) {
IotDbResult iotDbResult = new IotDbResult();
RowRecord next = sessionDataSet.next();
List<Field> fields = next.getFields();
String timeString = new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(next.getTimestamp());
iotDbResult.setTime(timeString);
Map<String, String> map = new HashMap<>();
for (int i = 0; i < fields.size(); i++) {
Field field = fields.get(i);
// 這裡的需要按照類型獲取
map.put(titleList.get(i), field.getObjectValue(field.getDataType()).toString());
}
iotDbResult.setTime(timeString);
iotDbResult.setPk(iotDbParam.getPk());
iotDbResult.setSn(iotDbParam.getSn());
iotDbResult.setHeart(map.get("heart"));
iotDbResult.setBreath(map.get("breath"));
iotDbResultList.add(iotDbResult);
}
}
}
}
6.控制層
package com.zhouhong.iotdbdemo.controller;
import com.zhouhong.iotdbdemo.config.IotDBSessionConfig;
import com.zhouhong.iotdbdemo.model.param.IotDbParam;
import com.zhouhong.iotdbdemo.response.ResponseData;
import com.zhouhong.iotdbdemo.server.IotDbServer;
import lombok.extern.log4j.Log4j2;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.springframework.web.bind.annotation.*;
import javax.annotation.Resource;
import java.rmi.ServerException;
/**
* description: iotdb 控制層
* date: 2022/8/15 21:50
* author: zhouhong
*/
@Log4j2
@RestController
public class IotDbController {
@Resource
private IotDbServer iotDbServer;
@Resource
private IotDBSessionConfig iotDBSessionConfig;
/**
* 插入數據
* @param iotDbParam
*/
@PostMapping("/api/device/insert")
public ResponseData insert(@RequestBody IotDbParam iotDbParam) throws StatementExecutionException, ServerException, IoTDBConnectionException {
iotDbServer.insertData(iotDbParam);
return ResponseData.success();
}
/**
* 插入數據
* @param iotDbParam
*/
@PostMapping("/api/device/queryData")
public ResponseData queryDataFromIotDb(@RequestBody IotDbParam iotDbParam) throws Exception {
return ResponseData.success(iotDbServer.queryDataFromIotDb(iotDbParam));
}
/**
* 刪除分組
* @return
*/
@PostMapping("/api/device/deleteGroup")
public ResponseData deleteGroup() throws StatementExecutionException, IoTDBConnectionException {
iotDBSessionConfig.deleteStorageGroup("root.a1eaKSRpRty");
iotDBSessionConfig.deleteStorageGroup("root.smartretirement");
return ResponseData.success();
}
}
四.PostMan測試
1.添加一條記錄
介面：localhost:8080/api/device/insert
入參：
{
"time":1660573444672,
"pk":"a1TTQK9TbKT",
"sn":"SN202208120945QGJLD",
"breath":"17",
"heart":"68"
}
查看IotDB數據
2.根據SQL查詢時間區間記錄(其他查詢以此類推)
介面：localhost:8080/api/device/queryData
入參：
{
"pk":"a1TTQK9TbKT",
"sn":"SN202208120945QGJLD",
"startTime":"2022-08-14 00:00:00",
"endTime":"2022-08-16 00:00:00"
}
結果：
{
"success": true,
"code": 200,
"message": "請求成功",
"localizedMsg": "請求成功",
"data": [
{
"time": "2022-08-15 22:24:04",
"pk": "a1TTQK9TbKT",
"sn": "SN202208120945QGJLD",
"breath": "19.0",
"heart": "75.0"
},
{
"time": "2022-08-15 22:24:04",
"pk": "a1TTQK9TbKT",
"sn": "SN202208120945QGJLD",
"breath": "20.0",
"heart": "78.0"
},
{
"time": "2022-08-15 22:24:04",
"pk": "a1TTQK9TbKT",
"sn": "SN202208120945QGJLD",
"breath": "17.0",
"heart": "68.0"
}
]
}
