國產時序資料庫IotDB安裝、與SpringBoot集成-有解無憂

一.簡介：

本文將完成一個真實業務中的設備上報資料的一個例子，完整的展示后臺服務接收到設備上報的資料后，將資料添加到時序資料庫，并且將資料查詢出來的一個例子，本文所有代碼已經上傳GitHub：https://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"
        }
    ]
}

IotDB還支持分頁、聚合等等其他操作，詳細資訊可以參考 https://iotdb.apache.org/zh/UserGuide/Master/Query-Data/Overview.html

本文來自博客園，作者：Tom-shushu，轉載請注明原文鏈接：https://www.cnblogs.com/Tom-shushu/p/16590246.html

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/501904.html

標籤：其他

上一篇：api進階Day1檔案的創建、洗掉、訪問、設定過濾器并查詢。目錄的洗掉、創建。

下一篇：聊聊客戶檔案模型的設計與管理