TDengine 零代码接入 SparkplugB：基于 taosExplorer 的 IIoT 数据同步任务配置指南-北京尧图网络

TDengine 零代码接入 SparkplugB基于 taosExplorer 的 IIoT 数据同步任务配置指南【免费下载链接】TDengineHigh-performance, scalable time-series database designed for Industrial IoT (IIoT) scenarios项目地址: https://gitcode.com/GitHub_Trending/tde/TDengine本文基于 TDengine 开源仓库 docs/en/08-data-ingest-and-delivery/01-no-code-ingestion/18-sparkplugb.md 编写。SparkplugB 是专为工业物联网IIoT设计、构建于 MQTT 之上的开放消息规范。借助 TDengine 的零代码数据接入平台你可以在 taosExplorer 图形界面中直接创建数据源任务让 taosX 连接器从 MQTT Broker 订阅 SparkplugB 消息并实时写入 TDengine 集群——无需编写任何代码。读完本文你将掌握从新增数据源、配置连接认证、订阅过滤、Payload 转换解析/拆分/过滤/表映射到高级选项与异常处理策略的完整任务创建流程。背景为什么需要 SparkplugB 数据接入SparkplugB 是一种开放的消息规范专为工业物联网IIoT应用设计底层基于 MQTT 协议。它定义了 IIoT 场景下设备与 MQTT Broker 之间的标准消息格式如 NBIRTH、NDATA、DDATA 等被广泛用于工厂、产线、设备监控等工业数据采集场景。TDengine 通过内置的 SparkplugB 连接器可以从 MQTT Broker 订阅 SparkplugB 消息并将数据实时写入 TDengine实现工业数据的实时入库。整个流程在 taosExplorer 的数据写入Data In页面通过零代码配置完成无需额外部署 ETL 工具。从零代码接入平台总览可知taosExplorer 是 TDengine 的可视化数据管理工具支持在浏览器中通过简单配置向 TDengine 提交任务实现多种数据源到 TDengine 的零代码导入并在导入过程中自动完成数据的抽取、过滤与转换。SparkplugB 正是该平台支持的众多数据源之一。说明SparkplugB 连接器在消息断点续传方面有明确限制——从任务断点续传一节可知与 MQTT、Kafka 等数据源不同SparkplugB 当前不支持消息持久化与恢复任务重启后无法从上次断点续传。创建数据写入任务新增数据源登录 taosExplorer 后进入数据写入页面点击新增数据源Add Data Source按钮进入任务创建页面。配置基本信息在任务创建页面中配置以下基本信息名称输入任务名称例如test_spb类型在下拉列表中选择SparkplugB代理可选选择一个已创建的 Agent 代理或点击右侧创建新的代理按钮新建目标数据库在下拉列表中选择一个目标数据库或点击右侧创建数据库按钮新建。配置连接与认证信息在连接配置区域需要填写 MQTT Broker 的接入参数配置项说明BrokersMQTT Broker 地址例如localhost:1883。可以填写多个用逗号,分隔用于连接多个 BrokerMQTT 协议使用的 MQTT 协议版本默认5.0可选 3.1、3.1.1、5.0客户端 ID连接到每个 Broker 时使用的客户端标识符Keep Alive保持活动间隔。如果 Broker 在该间隔内没有收到来自客户端的任何消息会假定客户端已断开并关闭连接。该间隔是客户端与 Broker 之间协商的、用于检测客户端活跃状态的时长用户 / 密码MQTT Broker 认证所需的用户名与密码若 Broker 开启了认证则必填实操提示同一 MQTT Broker 下如果创建多个同步任务各任务的客户端 ID 必须互不相同否则会造成冲突导致任务无法正常运行。TLS 校验模式TLS 校验TLS Verification支持三种模式不开启Disabled不进行 TLS 证书认证。连接 MQTT 时会先尝试 TCP 连接若失败则改为无证书认证模式的 TLS 连接。单向认证One-way开启 TLS 连接并验证服务端证书此时需要上传CA 证书。双向认证Mutual开启 TLS 连接并与服务端进行双向认证此时需要上传CA 证书、客户端证书以及客户端私钥。配置完成后点击检查连通性Check Connectivity按钮验证数据源是否可用若检查失败请根据页面返回的具体错误提示修改配置。订阅配置订阅配置决定连接器从哪些主题、哪些设备、哪些消息类型中消费数据。Group ID填写 SparkplugB 规范定义的 group id。通常一个 group id 代表一个集团/公司/工厂/流水线等概念。节点/设备列表Node/Device List填写需要订阅的节点和设备列表以逗号分隔。其中节点直接填写 ID 即可设备需要按照节点ID/设备ID的格式填写。消息类型Message Types填写需要订阅的 SparkplugB 消息类型以逗号分隔。支持的类型包括NBIRTH、NDEATH、NDATA、NCMD、DBIRTH、DDEATH、DDATA、DCMD、STATE。其中NBIRTH、NDEATH、NDATA、NCMD类型的消息只会匹配节点/设备列表中的节点而DBIRTH、DDEATH、DDATA、DCMD只会匹配节点/设备列表中的设备。下发 REBIRTH 命令Send REBIRTH Command开启后taosX 会自动下发 NCMD 中的Node Control/Rebirth命令从而获取节点和设备的所有 metric 信息包括 metric name 与 metric alias 的对应关系。如果节点/设备在上报数据时不使用 alias 别名机制可以不开启此选项。配置 Payload 转换Payload 转换是 SparkplugB 数据接入的核心环节包含解析、字段拆分、数据过滤、表映射四步是 taosX 内置 ETL 能力的具体体现参见数据抽取、过滤与转换。解析 PayloadPayload 解析区域提供三种获取示例数据的方式点击从服务器检索从已配置的 MQTT Broker 获取示例数据点击文件上传上传文件获取示例数据在消息体中手动填写 MQTT 消息体的示例数据。由于 SparkplugB 消息使用Protocol Buffersprotobuf编码从服务器检索到的数据会先被解码为 JSON 格式。JSON 数据支持 JSONObject 或 JSONArray 两种形态可以用于解析 SparkplugB 中的 metadata、properties 等 JSON 格式字段。点击放大镜图标可预览解析结果从列中提取或拆分字段解析后的数据可能仍不满足目标表的要求此时可以在从列中提取或拆分Extract or Split区域填写提取/拆分规则。典型的场景是将datatype_str字段的值转换为 TDengine 数据类型。选择映射mapping提取器在rule输入框中填写如下 JSON在name中填写td_datatype{ Int8: TINYINT, UInt8: TINYINT UNSIGNED, Int16: SMALLINT, UInt16: SMALLINT UNSIGNED, Int32: INT, UInt32: INT UNSIGNED, Int64: BIGINT, UInt64: BIGINT UNSIGNED, Float: FLOAT, DOUBLE: DOUBLE, Boolean: BOOL, String: VARCHAR(128), DateTime: TIMESTAMP }该规则会将datatype_str列的值例如字符串Int8转换为对应的 TDengine 类型例如TINYINT并生成新的列td_datatype。你可以点击新增添加更多提取规则点击删除移除当前规则点击放大镜图标预览提取/拆分结果。数据过滤在过滤Filter区域填写过滤表达式只有满足条件的数据行才会被写入 TDengine。例如填写datatype_str ! Int8则只有datatype_str值不为Int8的数据才会被写入。过滤表达式的结果必须为布尔类型支持基于字段类型的判断函数与比较运算符、、、、、!多个条件可通过逻辑运算符、||、!组合。例如location.starts_with(beijing) voltage 200表示只同步北京地区电压大于 200 的智能电表数据。相关过滤语法细节可参考零代码接入平台的过滤章节。点击删除可移除当前过滤规则点击放大镜图标可预览过滤结果。表映射表映射将解析、提取、拆分后的源字段映射到 TDengine 目标表。目标超级表在下拉列表中选择一个目标超级表或点击右侧创建超级表按钮新建。创建模板当超级表需要根据消息动态生成时选择创建模板。此时超级表名称、列名、列类型等均可以使用模板变量。接收到数据后程序会自动计算模板变量并生成对应的超级表模板当数据库中该超级表不存在时使用模板创建超级表对于已创建的超级表如果缺少通过模板变量计算得到的列也会自动创建对应列。映射填写目标超级表中的子表名称例如t_{id}根据需求填写映射规则其中 mapping 支持设置缺省值默认值。点击预览可查看映射结果确认子表名称、列与标签的映射是否符合预期。配置高级选项高级选项Advanced Options区域默认折叠点击展开。MQTT 与 SparkplugB 数据源常用的选项如下字段名可能因连接器而异参见高级选项详解选项说明Message Queue Size消息队列大小接收缓冲区大小。队列满且未开启缓存实时数据时新到达的数据会被丢弃设为0表示禁用缓冲Maximum In-Process Batches最大进行中批次可并发处理的批次数上限。达到上限后连接器停止从接收队列取消息消息会在队列中累积最小值为1Batch Size批量大小每次送入处理管道的消息条数。与批量延迟配合使用即使延迟未到批量已满也会立即发送最小值为1Batch Delay批量延迟每批次的超时时间毫秒从该批次第一条消息到达开始计时。超时后即使未达到 Batch Size 也会发送该批次最小值为1Write Concurrency写入并发并发写入 TDengine 的任务数Cache Realtime Data缓存实时数据开启后消费到的数据先写入本地文件由后台任务转发下游当下游处理跟不上时起到流量整形作用积压消费完毕后缓存文件会被清除。默认关闭。详见 Store and ForwardCache Storage Directory缓存存储目录覆盖缓存文件的存储目录仅在开启缓存实时数据时生效否则默认使用 taosX 启动时配置的数据目录Save Raw Data保存原始数据开启后可进一步配置最大保留天数与原始数据存储目录此外高级选项中还包含健康监控设置Health Check Duration、Busy State Threshold、Max Write Queue Length、Write Error Threshold用于任务列表页的健康状态展示具体说明参见 Health Status。配置异常处理策略异常处理策略Exception Handling Strategy区域默认折叠点击展开。taosX 为各类写入异常提供了统一的分流策略参见异常处理策略详解归档Archive将无效数据写入归档文件默认位于${data_dir}/tasks/id/datetime下不写入目标数据库丢弃Discard忽略无效数据报错Error报告错误缓存Cache目标连接失败或资源不足时将数据写入缓存文件待目标恢复后再行入库。可针对以下场景分别配置策略目标连接超时归档 / 丢弃 / 报错 / 缓存目标数据库不存在归档 / 丢弃 / 报错表不存在归档 / 丢弃 / 报错 / 自动建表并重试主时间戳超出范围now - keep1至now 100y归档 / 丢弃 / 报错主时间戳为空归档 / 丢弃 / 报错 / 使用当前时间复合主键为空归档 / 丢弃 / 报错表名超过 192 字符归档 / 丢弃 / 报错 / 截断 / 截断并归档表名含非法字符如.归档 / 丢弃 / 报错 / 用配置的字符串替换非法字符表名模板变量为空丢弃 / 变量留空 / 用配置的字符串替换列不存在归档 / 丢弃 / 报错 / 自动补列并重试列名超过 64 字符归档 / 丢弃 / 报错列值超出定义长度归档 / 丢弃 / 报错 / 截断 / 截断并归档也可通过自动扩列修改表结构后重试其他数据错误归档 / 丢弃 / 报错。附加设置项连接超时Connection Timeout目标连接超时时间秒取值范围1~600临时存储位置相对${data_dir}/tasks/id/的路径归档保留天数Archive Retention Days非负整数0表示不限归档可用空间Archive Available Space取值范围0~655350表示不限归档位置Archive Location相对${data_dir}/tasks/id/的路径归档写入失败策略删除旧文件 / 丢弃数据 / 报错并停止任务。提交任务完成上述所有配置后点击提交Submit按钮即完成 SparkplugB 到 TDengine 的数据同步任务创建自动回到数据源列表Data Source List页面。提交成功后可在任务列表页查看任务执行情况包括写入记录数、流量等运行指标任务状态会切换为 Running。你也可以在任务列表页对任务进行启动、停止、查看、删除、复制等管理操作并查看每个任务的健康状态Ready、Idle、Active、Pending、Busy、Bounce、SourceError、SinkError、Fatal 等详见任务管理。小结SparkplugB 数据接入任务的核心链路可概括为taosX 连接器订阅 MQTT Broker → 解码 protobuf 为 JSON → 解析/拆分/过滤 → 映射到超级表与子表 → 实时写入 TDengine。整个过程完全通过 taosExplorer 的零代码界面完成涵盖连接认证含 TLS 单向/双向认证、订阅配置Group ID、节点/设备、消息类型、REBIRTH、Payload 转换四种 ETL 步骤以及高级选项与异常处理兜底策略。配置时需特别注意SparkplugB 当前不支持消息持久化与断点续传对于需要高可靠连续采集的工业场景建议结合网络稳定性保障与异常归档策略共同使用。相关参考文档SparkplugB 接入指南英文原档SparkplugB 接入指南中文原档零代码数据接入平台总览taosX Agent 存储转发Store and Forward【免费下载链接】TDengineHigh-performance, scalable time-series database designed for Industrial IoT (IIoT) scenarios项目地址: https://gitcode.com/GitHub_Trending/tde/TDengine创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

TDengine 零代码接入 SparkplugB：基于 taosExplorer 的 IIoT 数据同步任务配置指南

相关推荐

mdput评测：一款免费轻量级Markdown编辑器，Typora的替代之选

使用 AI SDK 接入 GMI Cloud：OpenAI 兼容协议下的开源权重模型推理实战

考虑需求响应的微网优化调度模型与粒子群算法实现详解

更多资讯

Repomix 入门指南：将整个代码库打包成 AI 友好的单文件

猫抓浏览器资源嗅探扩展：从网页视频到本地 MP4 完整指南

qwen-code 输出 Token 上限自适应升级（Adaptive Output Token Escalation）机制解析

kkFileView JDK8/JDK11 版本选型：国产化部署的三问实用指南

Lucide 图标字体（Icon Font）使用指南：通过 CSS 类在项目中渲染全部 Lucide 图标

如何为自托管 sim 配置事务邮件提供商让工作区邀请与验证邮件真正发出

今日更新

本周精选

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？