做工業(yè)互聯(lián)網(wǎng)或物聯(lián)網(wǎng)系統(tǒng)，最基本的需求是展示數(shù)據(jù)曲線，比如功率曲線，類似于股票的分時(shí)圖，通常我們會(huì)取每分鐘內(nèi)該設(shè)備上報(bào)的最后一次功率值為這一分鐘的功率，如果某一分鐘內(nèi)，設(shè)備沒有上報(bào)，則取上一分鐘的功率值，以此類推。舉例如下：

得到的分鐘曲線：

通常我們會(huì)把設(shè)備上報(bào)的數(shù)據(jù)先寫入Apache Kafka。如果是離線計(jì)算場(chǎng)景，可能會(huì)考慮把數(shù)據(jù)寫入Hive，然后使用Spark SQL定時(shí)讀取Hive，再把計(jì)算結(jié)果寫入HBase；如果是實(shí)時(shí)計(jì)算場(chǎng)景，則會(huì)使用Apache Flink消費(fèi)Kafka數(shù)據(jù)，把結(jié)果寫入HBase，這種情況下還需要考慮數(shù)據(jù)亂序和延遲投遞計(jì)算等問題。

而且，基于傳統(tǒng)大數(shù)據(jù)Hadoop的架構(gòu)，需要搭建ZooKeeper和HDFS，然后才是Hive和HBase，整個(gè)體系維護(hù)成本很高。此外，HBase基于鍵值存儲(chǔ)時(shí)序數(shù)據(jù)，會(huì)浪費(fèi)很多空間在同一鍵值的數(shù)據(jù)設(shè)計(jì)架構(gòu)上面。

以上所舉，是物聯(lián)網(wǎng)設(shè)備屬性曲線計(jì)算場(chǎng)景的其中一個(gè)痛點(diǎn)，另外還需要考慮數(shù)據(jù)增長(zhǎng)、數(shù)據(jù)核對(duì)以及數(shù)據(jù)容災(zāi)等特點(diǎn)。

筆者所在的公司，要基于3D打印技術(shù)給客戶提供整體化解決方案，自然需要對(duì)設(shè)備的運(yùn)行狀態(tài)做持續(xù)追蹤，需要存儲(chǔ)設(shè)備的運(yùn)行數(shù)據(jù)。這時(shí)候我們找到了開源的物聯(lián)網(wǎng)大數(shù)據(jù)平臺(tái)TDengine（https://github.com/taosdata/TDengine）。

參考TDengine Database的文檔中SQL的寫法，在數(shù)據(jù)齊全的情況下，可以輕松地用一句SQL解決上面的問題：

select last(val) a from super_table_xx where ts >= '2021-06-07 18:10:00' and ts <= '2021-06-07 18:20:00' interval(60s) fill(value, 0);

為什么類似的SQL，TDengine Database的執(zhí)行效率可以如此之高呢？

這就在于它的超級(jí)表以及子表，針對(duì)單個(gè)設(shè)備的數(shù)據(jù)，TDengine設(shè)計(jì)了按照時(shí)間連續(xù)存儲(chǔ)的特性。而事實(shí)上，業(yè)務(wù)系統(tǒng)在使用物聯(lián)網(wǎng)數(shù)據(jù)的時(shí)候，無論是即時(shí)查詢還是離線分析，存在讀取單個(gè)設(shè)備的一個(gè)連續(xù)時(shí)間段數(shù)據(jù)的特點(diǎn)。

假設(shè)，我們要存儲(chǔ)設(shè)備的溫度與濕度，我們可以設(shè)計(jì)超級(jí)表如下：

create stable if not exists s_device (ts TIMESTAMP,
  temperature double,
  humidity double
) TAGS (device_sn BINARY(1000));

實(shí)際使用中，例如針對(duì)設(shè)備’d1’和’d2’的數(shù)據(jù)執(zhí)行插入的SQL如下：

insert into s_device_d1 (ts, temperature, humidity) USING s_device (device_sn) TAGS ('d1') values (1623157875000, 35.34, 80.24);
insert into s_device_d2 (ts, temperature, humidity) USING s_device (device_sn) TAGS ('d2') values (1623157891000, 29.63, 79.48);

搜索設(shè)備’d1’某個(gè)時(shí)間段的數(shù)據(jù)，其SQL如下：

select * from s_device where device_sn = 'd1' and ts > 1623157871000 and ts < 1623157890000 ;

假設(shè)統(tǒng)計(jì)過去7天的平均溫度曲線，每小時(shí)1個(gè)點(diǎn):

select avg(temperature) temperature from s_device where  device_sn = #{deviceSn} and ts >= #{startTime} and ts < #{endTime}  interval(1h)

TDengine還提供了很多聚合函數(shù)，類似上面的計(jì)算1分鐘連續(xù)曲線的last和fill，以及其他常用的sum和max等。

在和應(yīng)用程序結(jié)合的過程中，我們選用MyBatis這種靈活易上手的ORM框架，例如，針對(duì)上面的數(shù)據(jù)表’s_device’，我們先定義entity :

import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.sql.Timestamp;

/**
 * @author: DaLuo
 * @date: 2021/06/25
 * @description:
 */
@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
@TableName(value = "s_device")
public class TestSuperDeviceEntity {

    private Timestamp ts;
    private Float temperature;
    private Float humidity;
    @TableField(value = "device_sn")
    private String device_sn ;
}

再定義 mapper：

import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.hg.device.kafka.tdengine.entity.TestSuperDeviceEntity;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.ibatis.annotations.Insert;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import org.apache.ibatis.annotations.Select;

import java.sql.Timestamp;
import java.util.List;

/**
 * @author: DaLuo
 * @date: 2021/06/25
 * @description:
 */
@Mapper
public interface TestSuperDeviceMapper extends BaseMapper<TestSuperDeviceEntity> {

    /**
     * 單個(gè)插入
     * @param entity
     * @return
     */
    @Insert({
            "INSERT INTO 's_device_${entity.deviceSn}' (ts ,temperature, humidity ) ",
            "USING s_device (device_sn) TAGS (#{entity.deviceSn}) ",
            "VALUES (#{entity.ts}, #{entity.temperature}, #{entity.humidity})"
    })
    int insertOne(@Param(value = "entity") TestSuperDeviceEntity entity);


    /**
     * 批量插入
     * @param entities
     * @return
     */
    @Insert({
            "<script>",
            "INSERT INTO ",
            "<foreach collection='list' item='item' separator=' '>",
            "'s_device_${item.deviceSn}' (ts ,temperature, humidity) USING s_device (device_sn) TAGS (#{item.deviceSn}) ",
            "VALUES (#{item.ts}, #{item.temperature}, #{item.humidity})",
            "</foreach>",
            "</script>"
    })
    int batchInsert(@Param("list") List<TestSuperDeviceEntity> entities);

    /**
     *  查詢過去一段時(shí)間范圍的平均溫度，每小時(shí)1個(gè)數(shù)據(jù)點(diǎn)
     * @param deviceSn
     * @param startTime inclusive
     * @param endTime   exclusive
     * @return
     */
    @Select("select avg(temperature) temperature from s_device where  device_sn = #{deviceSn} and ts >= #{startTime} and ts < #{endTime}  interval(1h)")
    List<TempSevenDaysTemperature> selectSevenDaysTemperature(
            @Param(value = "deviceSn") String deviceSn,
            @Param(value = "startTime") long startTime,
            @Param(value = "endTime") long endTime);

    @AllArgsConstructor
    @NoArgsConstructor
    @Data
    @Builder
    class TempSevenDaysTemperature {
        private Timestamp ts;
        private float temperature;
    }
}

TDengine有一個(gè)很巧妙的設(shè)計(jì)，就是不用預(yù)先創(chuàng)建子表，所以我們可以很方便地利用’tag’標(biāo)簽作為子表名稱的一部分，即時(shí)插入數(shù)據(jù)同時(shí)創(chuàng)建子表。

注意：考慮到跨時(shí)區(qū)的國(guó)際化特性，我們所有的時(shí)間存儲(chǔ)查詢交互，都是使用的時(shí)間戳，而非”yyyy-mm-dd hh:MM:ss”格式，因?yàn)閿?shù)據(jù)存儲(chǔ)涉及到應(yīng)用程序時(shí)區(qū)，連接字符串時(shí)區(qū)，TDengine服務(wù)時(shí)區(qū)，使用”yyyy-mm-dd hh:MM:ss”格式容易導(dǎo)致時(shí)間存儲(chǔ)的不準(zhǔn)確性，而使用時(shí)間戳，長(zhǎng)整型的數(shù)據(jù)格式則可以完美地避免此類問題。

Java使用TDengine JDBC-driver目前有兩種方式：JDBC-JNI和JDBC-RESTful，前者在寫入性能上更有優(yōu)勢(shì)。但是需要在應(yīng)用程序運(yùn)行的服務(wù)器上安裝TDengine客戶端驅(qū)動(dòng)。

我們的應(yīng)用程序用到了Kubernetes集群，程序是運(yùn)行在Docker里面，為此我們制作了一個(gè)適合我們應(yīng)用程序運(yùn)行的鏡像，例如基礎(chǔ)鏡像的Dockerfile如下所示：

FROM openjdk:8-jdk-oraclelinux7
 
COPY TDengine-client-2.0.16.0-Linux-x64.tar.gz /
 
RUN tar -xzvf /TDengine-client-2.0.16.0-Linux-x64.tar.gz &&  cd /TDengine-client-2.0.16.0 &&  pwd && ls && ./install_client.sh

build：

docker build -t tdengine-openjdk-8-runtime:2.0.16.0 -f Dockerfile .

引用程序鏡像Dockerfile所示：

FROM tdengine-openjdk-8-runtime:2.0.16.0
 
ENV JAVA_OPTS="-Duser.timezone=Asia/Shanghai -Djava.security.egd=file:/dev/./urandom"
 
COPY app.jar /app.jar
 
ENTRYPOINT ["java","-jar","/app.jar"]

這樣我們的應(yīng)用程序就可以調(diào)度在任意的K8s節(jié)點(diǎn)上了。

另外，我們的程序涉及到任務(wù)自動(dòng)化調(diào)度，需要頻繁地和設(shè)備下位機(jī)進(jìn)行MQTT數(shù)據(jù)交互，比如，云端發(fā)送指令1000-“開始任務(wù)A”，下位機(jī)回復(fù)指令2000-“收到任務(wù)A”，把指令理解成設(shè)備，把指令序列號(hào)以及內(nèi)容理解成它的屬性，自然這種數(shù)據(jù)也是非常適合存儲(chǔ)在TDengine時(shí)序數(shù)據(jù)庫(kù)中的：

*************************** 1.row ***************************
       ts: 2021-06-23 16:10:30.000
      msg: {"task_id":"7b40ed4edc1149f1837179c77d8c3c1f","action":"start"}
device_sn: deviceA
     kind: 1000
*************************** 2.row ***************************
       ts: 2021-06-23 16:10:31.000
      msg: {"task_id":"7b40ed4edc1149f1837179c77d8c3c1f","action":"received"}
device_sn: deviceA
     kind: 2000

我們?cè)贫嗽诤驮O(shè)備對(duì)接的過程中，頻繁需要考究消息是否發(fā)送的問題，所以迫切需要對(duì)指令進(jìn)行保存，從而在應(yīng)用程序中新辟線程，專門訂閱指令集消息，批量寫入到TDengine數(shù)據(jù)庫(kù)。

最后，TDengine Database還有一個(gè)超級(jí)表log.dn，里面保留了內(nèi)存、CPU等使用信息，所以我們可以利用Grafana展示這些數(shù)據(jù)，為監(jiān)控提供可靠的運(yùn)營(yíng)數(shù)據(jù)參照！

作者介紹：大羅，黑格智造架構(gòu)師，主要從事云原生，大數(shù)據(jù)系統(tǒng)開發(fā)，曾參與國(guó)家示范級(jí)工業(yè)互聯(lián)網(wǎng)系統(tǒng)建設(shè)等。