从建模到运维:联犀如何完美融入时序数据库 TDengine 实现物联网数据流畅管理

本文涉及的产品
日志服务 SLS,月写入数据量 50GB 1个月
简介: 本篇文章是“2024,我想和 TDengine 谈谈”征文活动的三等奖作品。文章从一个具体的业务场景出发,分析了企业在面对海量时序数据时的挑战,并提出了利用 TDengine 高效处理和存储数据的方法,帮助企业解决在数据采集、存储、分析等方面的痛点。通过这篇文章,作者不仅展示了自己对数据处理技术的理解,还进一步阐释了时序数据库在行业中的潜力与应用价值,为读者提供了很多实际的操作思路和技术选型的参考。

小T导读:本篇文章是“2024,我想和 TDengine 谈谈”征文活动的三等奖作品。文章从一个具体的业务场景出发,分析了企业在面对海量时序数据时的挑战,并提出了利用 TDengine 高效处理和存储数据的方法,帮助企业解决在数据采集、存储、分析等方面的痛点。通过这篇文章,作者不仅展示了自己对数据处理技术的理解,还进一步阐释了时序数据库在行业中的潜力与应用价值,为读者提供了很多实际的操作思路和技术选型的参考。

在当前的物联网行业,TDengine 已成为国内占有率最高的开源时序数据库,并在开源物联网平台中扮演着至关重要的角色。接下来,我们将分几个模块详细探讨联犀物联网平台如何与 TDengine 深度结合,并进一步提升 TDengine 的扩展性,推动其在物联网场景中的应用潜力。

物模型如何结合 TDengine?

物模型介绍
现实世界由众多真实存在的物理设备组成,我们可以将这些设备称之为“物”。物联网的目标是通过网络将这些“物”连接在一起,并将其数字化为云端服务或资源,从而实现智能化应用。因此,在物联网构建的数字世界中,首先需要对“物”进行清晰、统一的定义,明确其功能和能提供的服务资源。
ICA 联盟从产品层面对“物”进行了功能建模,提出了统一的“物的抽象模型”和“物的描述语言”(TSL,Things Specification Language)。物的抽象模型描述了“设备是什么”以及“设备能做什么”,包括物的状态、档案信息和功能定义等方面。
1.PNG

在物模型中,最为关键的是属性(property)。以智能电灯为例,其状态具有二元性:开启或关闭。用户可以通过控制操作轻松在这两种状态之间切换。此外,一些智能电灯还具备更多高级功能,允许用户根据个人需求调整亮度、颜色和色温等参数。
智能设备的属性通常具备读写能力,这意味着应用程序不仅可以读取设备的当前状态,还可以修改属性来调整设备行为。例如,在环境监测设备中,应用程序可以读取温度和湿度数据,并根据需要调整参数,以适应不同的环境条件。

物模型如何结合 TDengine?

物模型超级表与产品的关系
在联犀物联网平台中,存在两种物模型:公共物模型和产品物模型。公共物模型是由多个产品共同定义的共享物模型,而产品物模型则为每个特定产品设备单独定义。定义物模型时,通常会有以下两种操作:

  • 产品物模型:每类特定产品设备使用一个超级表来定义。
  • 通用物模型:只创建一个超级表,后续的产品可以按需引入该物模型,产品下的设备则直接使用这个超级表。
    TDengine 支持灵活的数据模型设计,既可以使用多列模型,也可以选择单列模型。多列模型相当于将多个字段存储在同一张超级表中,通常在写入和存储效率上表现较优。而在某些情况下,例如数据采集点的种类和数量经常变化时,单列模型则可能更为适用,因为它简化了应用程序的设计和管理,允许独立管理和扩展每个物理量的超级表。
    综上所述,TDengine 提供了灵活的数据模型选项,用户可以根据具体需求和应用场景选择最适合的模型。无论是采用窄表设计、单列模型还是多列模型,最终目的是为了优化性能和简化管理复杂性。

物模型和表分为三种对应关系

  1. 普通类型: 根据物模型的数据类型,映射单列模式的超级表。物模型定义示例如下图所示。

2.PNG
示例 SQL 如下:

CREATE STABLE IF NOT EXISTS model_custom_property_00b_int (ts timestamp,param BIGINT) 
  TAGS (product_id BINARY(50),device_name BINARY(50),property_type BINARY(50));

相关参数说明:

3.PNG

param 物模型和 TDengine 表结构的定义对照如下:
4.png
下面是完整的建表语句:

CREATE STABLE IF NOT EXISTS model_custom_property_00b_int (ts timestamp,param BIGINT) TAGS (product_id BINARY(50),device_name BINARY(50),property_type BINARY(50));
CREATE STABLE IF NOT EXISTS model_custom_property_00b_float (ts timestamp,param DOUBLE) TAGS (product_id BINARY(50),device_name BINARY(50),property_type BINARY(50));
CREATE STABLE IF NOT EXISTS model_custom_property_00b_enum (ts timestamp,param SMALLINT) TAGS (product_id BINARY(50),device_name BINARY(50),property_type BINARY(50));
CREATE STABLE IF NOT EXISTS model_custom_property_00b_timestamp (ts timestamp,param TIMESTAMP) TAGS (product_id BINARY(50),device_name BINARY(50),property_type BINARY(50));
CREATE STABLE IF NOT EXISTS model_custom_property_00b_bool (ts timestamp,param BOOL) TAGS (product_id BINARY(50),device_name BINARY(50),property_type BINARY(50));
CREATE STABLE IF NOT EXISTS model_custom_property_00b_string (ts timestamp,param BINARY(5000)) TAGS (product_id BINARY(50),device_name BINARY(50),property_type BINARY(50));

对应的每个设备创建的普通表如下:

CREATE TABLE IF NOT EXISTS device_property_00b_device1_int USING model_custom_property_00b_int  TAGS('00b','device1','int');
CREATE TABLE IF NOT EXISTS device_property_00b_device1_enum USING model_custom_property_00b_enum  TAGS('00b','device1','enum');
CREATE TABLE IF NOT EXISTS device_property_00b_device1_bool USING model_custom_property_00b_bool  TAGS('00b','device1','bool');
CREATE TABLE IF NOT EXISTS device_property_00b_device1_string USING model_custom_property_00b_string  TAGS('00b','device1','string');
CREATE TABLE IF NOT EXISTS device_property_00b_device1_float USING model_custom_property_00b_float  TAGS('00b','device1','float');
CREATE TABLE IF NOT EXISTS device_property_00b_device1_timestamp USING model_custom_property_00b_timestamp  TAGS('00b','device1','timestamp');

2.结构体类型:拥有多个字段并将物模型进行整体抽象,映射为多列模式的超级表。物模型定义示例如下图所示。

5.png
示例 SQL 如下:


CREATE STABLE IF NOT EXISTS model_custom_property_00b_struct (ts timestamp, latitude DOUBLE,longitude DOUBLE) TAGS (product_id BINARY(50),device_name BINARY(50),property_type BINARY(50));

相关参数说明:
39.jpg
对应的每个设备创建的普通表如下:


CREATE TABLE IF NOT EXISTS device_property_00b_device1_struct USING model_custom_property_00b_struct  TAGS('00b','device1','struct');

3.数组类型: 数组类型在物联网平台中较为特殊,传统平台中的数组无法单独操作某一位。例如,如果想单独修改开关10的状态,必须传递完整的数组(如:[0,1,1,0,1,0,1,1,0,1])来进行控制,这种方式并不符合现实世界的需求。联犀则对数组进行了扩展,支持下角标访问。比如,要修改开关 10 的状态,只需传递 "switch.10": 1 即可。接下来,让我们来看一下联犀是如何处理这种数据结构的。物模型定义示例如下图所示。
6.png
示例 SQL 如下:

 CREATE STABLE IF NOT EXISTS model_custom_property_00b_switchg (ts timestamp,param BOOL) TAGS (product_id BINARY(50),device_name BINARY(50),_num BIGINT,property_type BINARY(50));

相关参数说明:
40.jpg
可以看到,数组类型的定义与简单类型相似,区别在于它额外添加了一个 _num 字段,用来标识数组的下标。这种设计的好处在于,即使数组长度达到 1000,依然只需定义一个超级表。虽然每个设备的普通表需要定义 1000 个,但由于是由一个超级表进行管理,整体管理变得更加简便。接下来,我们来看看每个设备创建的普通表是什么样的。


CREATE TABLE IF NOT EXISTS device_property_00b_device1_switchg_0 USING model_custom_property_00b_switchg  TAGS('00b','device1',0,'bool');
CREATE TABLE IF NOT EXISTS device_property_00b_device1_switchg_1 USING model_custom_property_00b_switchg  TAGS('00b','device1',1,'bool');
CREATE TABLE IF NOT EXISTS device_property_00b_device1_switchg_2 USING model_custom_property_00b_switchg  TAGS('00b','device1',2,'bool');
CREATE TABLE IF NOT EXISTS device_property_00b_device1_switchg_3 USING model_custom_property_00b_switchg  TAGS('00b','device1',3,'bool');
CREATE TABLE IF NOT EXISTS device_property_00b_device1_switchg_4 USING model_custom_property_00b_switchg  TAGS('00b','device1',4,'bool');
CREATE TABLE IF NOT EXISTS device_property_00b_device1_switchg_5 USING model_custom_property_00b_switchg  TAGS('00b','device1',5,'bool');
CREATE TABLE IF NOT EXISTS device_property_00b_device1_switchg_6 USING model_custom_property_00b_switchg  TAGS('00b','device1',6,'bool');
CREATE TABLE IF NOT EXISTS device_property_00b_device1_switchg_7 USING model_custom_property_00b_switchg  TAGS('00b','device1',7,'bool');
CREATE TABLE IF NOT EXISTS device_property_00b_device1_switchg_8 USING model_custom_property_00b_switchg  TAGS('00b','device1',8,'bool');
CREATE TABLE IF NOT EXISTS device_property_00b_device1_switchg_9 USING model_custom_property_00b_switchg  TAGS('00b','device1',9,'bool');

如何发挥 TDengine 的性能?

在设备上报信息时,如果在线设备达到 120 万,每个设备每 10 分钟上报一次数据,那么每秒钟大约会有 2000 条消息需要处理。每条消息不仅需要入库一条调试日志,还需要记录一条属性历史数据。通过大量性能测试和优化,我们发现,突破 2000 QPS 是非常具有挑战性的。深入分析了 TDengine 的体系架构后,我们实施了以下优化措施:

  1. 从 HTTP 切换到 WebSocket:TDengine 从 3.x 版本开始支持 WebSocket,我们顺势升级,经过测试,性能和稳定性都有所提升,但在高并发的情况下,系统资源消耗依然较大。
  2. 改进同步操作为异步操作:通过将同步操作转为异步处理,显著提高了整体性能。
  3. SQL 批量插入优化:TDengine 支持单条 SQL 语句插入多条数据,甚至跨表插入,这大大提升了写入性能。
    官方语法如下:

、INSERT INTO
    tb_name
        [USING stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)]
        [(field1_name, ...)]
        VALUES (field1_value, ...) [(field1_value2, ...) ...] | FILE csv_file_path
    [tb2_name
        [USING stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)]
        [(field1_name, ...)]
        VALUES (field1_value, ...) [(field1_value2, ...) ...] | FILE csv_file_path
    ...];

INSERT INTO tb_name [(field1_name, ...)] subquery

官方示例如下:


INSERT INTO d21001 USING meters TAGS ('California.SanFrancisco', 2) VALUES ('2021-07-13 14:06:34.630', 10.2, 219, 0.32) ('2021-07-13 14:06:35.779', 10.15, 217, 0.33)
            d21002 USING meters (groupId) TAGS (2) VALUES ('2021-07-13 14:06:34.255', 10.15, 217, 0.33)
            d21003 USING meters (groupId) TAGS (2) (ts, current, phase) VALUES ('2021-07-13 14:06:34.255', 10.27, 0.31);

联犀利用这一语法实现了异步操作,操作流程如下:

  1. 设备插入数据时,首先生成类似 d21003 USING meters (groupId) TAGS (2) (ts, current, phase) VALUES ('2021-07-13 14:06:34.255', 10.27, 0.31) 格式的插入语句。
  2. 将生成的 SQL 语句放入 Golang 的 Channel 中。
  3. 多个异步入库协程监听 Channel,并从中取出 SQL 语句。当执行间隔超过 1 秒或 SQL 数量达到上限时,协程会将这些数据合并成完整的 SQL 批量插入数据库。

由于设备上报数据的整个流程不涉及磁盘操作,整个过程速度非常快。即使在未完全优化的情况下,经过测试,单机配置为 8 核 16GB 的服务器,能够稳定支持每秒 7000 并发请求,且保持毫秒级的低延迟,并无任何错误。
7.png
关键代码如下:


type Td struct {
    *sql.DB
}

type ExecArgs struct {
    Query string
    Args  []any
}

var (
    td         = Td{}
    once       = sync.Once{}
    insertChan = make(chan ExecArgs, 1000)
)

const (
    asyncExecMax = 200 //异步执行sql最大数量
    asyncRunMax  = 40
)

func NewTDengine(DataSource conf.TSDB) (TD *Td, err error) {
    once.Do(func() {
        if DataSource.Driver == "" {
            DataSource.Driver = "taosWS"
        }
        td.DB, err = sql.Open(DataSource.Driver, DataSource.DSN)
        if err != nil {
            return
        }
        td.DB.SetMaxIdleConns(50)
        td.DB.SetMaxOpenConns(50)
        td.DB.SetConnMaxIdleTime(time.Hour)
        td.DB.SetConnMaxLifetime(time.Hour)
        _, err = td.Exec("create database if not exists ithings;")
        if err != nil {
            return
        }
        for i := 0; i < asyncRunMax; i++ {
            utils.Go(context.Background(), func() {
                td.asyncInsertRuntime()
            })
        }
    })
    if err != nil {
        logx.Errorf("TDengine 初始化失败,err:%v", err)
    }
    return &td, err
}

func (t *Td) asyncInsertRuntime() {
    r := rand.Intn(1000)
    tick := time.Tick(time.Second/2 + time.Millisecond*time.Duration(r))
    execCache := make([]ExecArgs, 0, asyncExecMax*2)
    exec := func() {
        if len(execCache) == 0 {
            return
        }
        sql, args := t.genInsertSql(execCache...)
        var err error
        for i := 3; i > 0; i-- { //三次重试
            _, err = t.Exec(sql, args...)
            if err == nil {
                break
            }
        }
        if err != nil {
            logx.Error(err)
        }
        execCache = execCache[0:0] //清空切片
    }
    for {
        select {
        case _ = <-tick:
            exec()
        case e := <-insertChan:
            execCache = append(execCache, e)
            if len(execCache) > asyncExecMax {
                exec()
            }
        }
    }

}

func (t *Td) AsyncInsert(query string, args ...any) {
    insertChan <- ExecArgs{
        Query: query,
        Args:  args,
    }
}
func (t *Td) genInsertSql(eas ...ExecArgs) (query string, args []any) {
    qs := make([]string, 0, len(eas))
    as := make([]any, 0, len(eas))
    for _, e := range eas {
        qs = append(qs, e.Query)
        as = append(as, e.Args...)
    }
    return fmt.Sprintf("insert into %s;", strings.Join(qs, " ")), as
}

TDengine 查询

数据插入数据库后,我们开始着手进行数据查询和展示。得益于 TDengine 支持多种丰富的数据聚合方式,数据分析变得更加高效和便捷。
ORM 设计
灵活的查询方式离不开 ORM 框架的支持。然而,TDengine 的官方库并未提供 ORM 框架,开源社区也没有专门为 TDengine 开发的 ORM 框架。为了填补这一空白,联犀基于知名 ORM 框架 Squirrel(https://github.com/Masterminds/squirrel)进行了扩展,以支持 TDengine 的语法。以下是 ORM 的示例:


func (d *DeviceDataRepo) getPropertyArgFuncSelect(
    ctx context.Context,
    filter msgThing.FilterOpt) (sq.SelectBuilder, error) {
    schemaModel, err := d.getSchemaModel(ctx, filter.ProductID)
    if err != nil {
        return sq.SelectBuilder{}, err
    }
    p, ok := schemaModel.Property[filter.DataID]
    if !ok {
        return sq.SelectBuilder{}, errors.Parameter.AddMsgf("dataID:%s not find", filter.DataID)
    }
    var (
        sql sq.SelectBuilder
    )

    if p.Define.Type == schema.DataTypeStruct {
        sql = sq.Select("FIRST(ts) AS ts", d.GetSpecsColumnWithArgFunc(p.Define.Specs, filter.ArgFunc))
    } else {
        sql = sq.Select("FIRST(ts) AS ts", fmt.Sprintf("%s(param) as param", filter.ArgFunc))
    }
    if filter.Interval != 0 {
        sql = sql.Interval("?a", filter.Interval) //TDengine特有语法
    }
    if len(filter.Fill) > 0 {
        sql = sql.Fill(filter.Fill)//TDengine特有语法
    }
    return sql, nil
}

灵活的查询接口

借助上述的 ORM 底层实现,我们才得以实现灵活的查询接口,下面是查询接口示例。
请求参数:
8.png
回复参数:

9.png

TDengine 链路追踪

最后,我们还需要解决链路追踪的问题:我们需要记录每个 SQL 的执行耗时,并能够与业务链路 ID 打通,确保可以通过日志追溯到具体的 SQL 执行者和执行过程。虽然官方驱动较为底层,但它支持通过 context 进行传递。联犀的链路追踪同样基于 context 传递,因此我们只需在执行 SQL 时记录相关日志即可。无论是采用 HTTP、WebSocket 还是 CGo 连接方式,日志都统一记录在 driver-go/taosWS/connection.go 文件中。联犀打印的日志如下:
10.png

结语

通过以上一系列工作,我们成功地实现了 TDengine 在建模、数据写入、查询和运维等各个环节的无缝衔接,并与物联网业务完美融合。为了感谢 TDengine 对我们产品的支持与帮助,我们特撰此文,希望能为相关从业人士提供一些借鉴与启发。

附录

联犀开源地址: https://gitee.com/unitedrhino
TDengine ORM 定制: https://gitee.com/unitedrhino/squirrel
TDengine 官方驱动定制: https://gitee.com/unitedrhino/driver-go

相关实践学习
阿里云AIoT物联网开发实战
本课程将由物联网专家带你熟悉阿里云AIoT物联网领域全套云产品,7天轻松搭建基于Arduino的端到端物联网场景应用。 开始学习前,请先开通下方两个云产品,让学习更流畅: IoT物联网平台:https://iot.console.aliyun.com/ LinkWAN物联网络管理平台:https://linkwan.console.aliyun.com/service-open
目录
相关文章
|
2月前
|
存储 JSON 关系型数据库
【干货满满】解密 API 数据解析:从 JSON 到数据库存储的完整流程
本文详解电商API开发中JSON数据解析与数据库存储的全流程,涵盖数据提取、清洗、转换及优化策略,结合Python实战代码与主流数据库方案,助开发者构建高效、可靠的数据处理管道。
|
10天前
|
存储 数据管理 数据库
数据字典是什么?和数据库、数据仓库有什么关系?
在数据处理中,你是否常困惑于字段含义、指标计算或数据来源?数据字典正是解答这些问题的关键工具,它清晰定义数据的名称、类型、来源、计算方式等,服务于开发者、分析师和数据管理者。本文详解数据字典的定义、组成及其与数据库、数据仓库的关系,助你夯实数据基础。
数据字典是什么?和数据库、数据仓库有什么关系?
|
5月前
|
存储 缓存 数据库
数据库数据删除策略:硬删除vs软删除的最佳实践指南
在项目开发中,“删除”操作常见但方式多样,主要分为硬删除与软删除。硬删除直接从数据库移除数据,操作简单、高效,但不可恢复;适用于临时或敏感数据。软删除通过标记字段保留数据,支持恢复和审计,但增加查询复杂度与数据量;适合需追踪历史或可恢复的场景。两者各有优劣,实际开发中常结合使用以满足不同需求。
345 4
|
6月前
|
传感器 人工智能 物联网
健康监测设备的技术革命:AI+物联网如何让你随时掌握健康数据?
健康监测设备的技术革命:AI+物联网如何让你随时掌握健康数据?
732 19
|
20天前
|
存储 关系型数据库 数据库
【赵渝强老师】PostgreSQL数据库的WAL日志与数据写入的过程
PostgreSQL中的WAL(预写日志)是保证数据完整性的关键技术。在数据修改前,系统会先将日志写入WAL,确保宕机时可通过日志恢复数据。它减少了磁盘I/O,提升了性能,并支持手动切换日志文件。WAL文件默认存储在pg_wal目录下,采用16进制命名规则。此外,PostgreSQL提供pg_waldump工具解析日志内容。
|
3月前
|
存储 SQL Java
数据存储使用文件还是数据库,哪个更合适?
数据库和文件系统各有优劣:数据库读写性能较低、结构 rigid,但具备计算能力和数据一致性保障;文件系统灵活易管理、读写高效,但缺乏计算能力且无法保证一致性。针对仅需高效存储与灵活管理的场景,文件系统更优,但其计算短板可通过开源工具 SPL(Structured Process Language)弥补。SPL 提供独立计算语法及高性能文件格式(如集文件、组表),支持复杂计算与多源混合查询,甚至可替代数据仓库。此外,SPL 易集成、支持热切换,大幅提升开发运维效率,是后数据库时代文件存储的理想补充方案。
|
5月前
|
人工智能 关系型数据库 分布式数据库
让数据与AI贴得更近,阿里云瑶池数据库系列产品焕新升级
4月9日阿里云AI势能大会上,阿里云瑶池数据库发布重磅新品及一系列产品能力升级。「推理加速服务」Tair KVCache全新上线,实现KVCache动态分层存储,显著提高内存资源利用率,为大模型推理降本提速。
|
6月前
|
存储 SQL 关系型数据库
从 MySQL 到时序数据库 TDengine:Zendure 如何实现高效储能数据管理?
TDengine 助力广州疆海科技有限公司高效完成储能业务的数据分析任务,轻松应对海量功率、电能及输入输出数据的实时统计与分析,并以接近 1 : 20 的数据文件压缩率大幅降低存储成本。此外,taosX 强大的 transform 功能帮助用户完成原始数据的清洗和结构优化,而其零代码迁移能力更实现了历史数据从 TDengine OSS 与 MySQL 到 TDengine 企业版的平滑迁移,全面提升了企业的数据管理效率。本文将详细解读这一实践案例。
228 0
|
4月前
|
数据采集 机器学习/深度学习 人工智能
运维人的“福音”?AI 驱动的自动化网络监控到底香不香!
运维人的“福音”?AI 驱动的自动化网络监控到底香不香!
280 0
|
11天前
|
人工智能 运维 安全
运维老哥的救星?AI 驱动的自动化配置管理新趋势
运维老哥的救星?AI 驱动的自动化配置管理新趋势
61 11