零代码数据接入指南基于 taosExplorer 与 taosX 的数据源集成、ETL 与任务管理【免费下载链接】TDengineHigh-performance, scalable time-series database designed for Industrial IoT (IIoT) scenarios项目地址: https://gitcode.com/GitHub_Trending/tde/TDengine本文档基于 TDengine 官方指南《Data Connectors》docs/en/08-data-ingest-and-delivery/01-no-code-ingestion/index.md编写系统讲解如何通过 taosExplorer 与 taosX 以零代码方式将第三方数据源持续写入 TDengine。引言TDengine 内置了可视化数据管理工具taosExplorer即 TDengine TSDB Explorer。借助 taosExplorer用户只需在浏览器中通过简单配置即可向 TDengine 提交任务实现从多种数据源到 TDengine 的零代码数据导入。导入过程中TDengine 自动完成数据的解析、过滤和转换保证导入数据质量。通过这种零代码数据源集成方式TDengine 已成为时序大数据聚合平台用户无需额外部署 ETL 工具从而极大简化整体架构设计、提升数据处理效率。从架构上看零代码集成平台由两个核心组件构成taosExplorer提供可视化 Web 界面用于创建数据接入任务、配置解析规则、预览结果、监控任务状态。taosX负责连接数据源、执行解析/过滤/转换逻辑并写入 TDengine 的服务端组件。当 taosX 无法直连数据源时例如数据源位于隔离的 OT 网络、工业控制网段或受限内网可在数据源所在网络部署taosX-Agent作为代理由 Agent 采集数据并转发给 taosX详见安装 taosX-Agent与taosX-Agent 参考。图零代码集成平台的系统架构。支持的数据源TDengine 当前支持的数据源如下数据源支持版本说明Aveva PI SystemPI AF Server 2.10.9.593 及以上工业数据管理与分析平台前身为 OSIsoft PI System可实时采集、集成、分析与可视化工业数据帮助企业实现智能决策与精细化管理Aveva HistorianAVEVA Historian 2020 RS SP1工业大数据分析软件前身为 Wonderware Historian面向工业环境存储、管理和分析来自各类工业设备与传感器的实时及历史数据OPC DAMatrikon OPC 1.7.2.7433Open Platform Communications 缩写一种开放、标准化的通信协议用于不同制造商自动化设备之间的数据交换。最初由微软开发旨在解决工业控制领域的互操作性问题OPC 协议于 1996 年首次发布当时称为 OPC DAData Access主要用于实时数据采集与控制OPC UAKeepWare KEPServerEx 6.52006 年 OPC 基金会发布 OPC UAUnified Architecture标准面向服务、面向对象灵活性和可扩展性更高已成为 OPC 协议的主流版本MQTTemqx: 3.0.0 ~ 5.7.1hivemq: 4.0.0 ~ 4.31.0mosquitto: 1.4.4 ~ 2.0.18Message Queuing Telemetry Transport 缩写基于发布/订阅模式的轻量级通信协议低开销、低带宽占用广泛适用于物联网、小型设备、移动应用等领域Kafka2.11 ~ 3.8.0Apache 软件基金会开发的开源流处理平台主要用于处理实时数据提供统一、高吞吐、低延迟的消息系统。具有高速、可扩展、持久化与分布式设计可每秒处理数十万次读写、支持数千客户端同时保持数据可靠性与可用性InfluxDB1.7、1.8、2.0-2.7广受欢迎的开源时序数据库针对海量时序数据的处理进行了优化OpenTSDB2.4.1基于 HBase 的分布式、可扩展时序数据库主要用于存储、索引和访问从大规模集群包括网络设备、操作系统、应用等采集的指标数据使数据更易于访问和图形化展示MySQL5.6、5.7、8.0最流行的关系型数据库管理系统之一以体积小、速度快、总体拥有成本低著称特别是其开源特性使其成为中大型网站数据库开发的选择Oracle11G/12c/19c世界流行的关系型数据库管理系统之一可移植性好、使用方便、功能强大适用于各种大中小型计算机环境是高效、可靠、高吞吐的数据库解决方案PostgreSQLv15.0非常强大的开源客户端/服务器关系型数据库管理系统具备大型商业 RDBMS 的众多特性包括事务、子查询、触发器、视图、外键引用完整性和复杂锁定能力SQL Server2012/2022微软开发的关系型数据库管理系统易用性好、可扩展性强、与相关软件集成度高MongoDB3.6介于关系型与非关系型数据库之间的产品广泛用于内容管理系统、移动应用和物联网等众多领域CSV-Comma Separated Values 缩写以逗号分隔的纯文本文件格式常用于电子表格或数据库软件TDengine Query2.4、3.0从旧版本 TDengine 查询并写入新集群TDengine Data Subscription3.0使用 TMQ 订阅 TDengine 中指定的数据库或超级表数据抽取、过滤与转换内置 ETL由于数据源众多各数据源可能具有不同的物理单位、命名约定和时区。为应对这一问题TDengine 内置了 ETL 能力可以从数据源的数据包中解析并提取所需数据并执行过滤与转换以保证写入数据的质量并提供统一的命名空间。具体功能如下解析使用 JSON Path 或正则表达式从原始消息中解析字段。从列中抽取或拆分使用 split 或正则表达式从一个原始字段中提取多个字段。过滤仅当表达式值为真时消息才写入 TDengine。转换在解析字段与 TDengine 超级表字段之间建立转换与映射关系。解析Parsing仅非结构化数据源需要此步骤。目前 MQTT 和 Kafka 数据源使用本步骤提供的规则解析非结构化数据以初步获得结构化数据即可用字段描述的行列数据。在 Explorer 中需要提供示例数据和解析规则并预览解析后的结构化数据表格。示例数据Sample Data图示例数据输入区。文本区域中存放示例数据可通过三种方式获取直接在文本区域中输入示例数据点击右侧Retrieve from Server从服务器获取按钮从已配置的服务器获取示例数据并追加到示例数据文本区域上传文件将文件内容追加到示例数据文本区域。每条示例数据以回车符结尾。解析规则解析是将非结构化字符串解析为结构化数据的过程。消息体的解析规则目前支持JSON、Regex正则表达式和UDT自定义解析脚本三种。1. JSON 解析JSON 解析支持 JSONObject 或 JSONArray。以下 JSON 示例数据可自动解析出字段groupid、voltage、current、ts、inuse、location。{groupid: 170001, voltage: 221V, current: 12.3, ts: 2023-12-18T22:12:00, inuse: true, location: beijing.chaoyang.datun} {groupid: 170001, voltage: 220V, current: 12.2, ts: 2023-12-18T22:12:02, inuse: true, location: beijing.chaoyang.datun} {groupid: 170001, voltage: 216V, current: 12.5, ts: 2023-12-18T22:12:04, inuse: false, location: beijing.chaoyang.datun}或[{groupid: 170001, voltage: 221V, current: 12.3, ts: 2023-12-18T22:12:00, inuse: true, location: beijing.chaoyang.datun}, {groupid: 170001, voltage: 220V, current: 12.2, ts: 2023-12-18T22:12:02, inuse: true, location: beijing.chaoyang.datun}, {groupid: 170001, voltage: 216V, current: 12.5, ts: 2023-12-18T22:12:04, inuse: false, location: beijing.chaoyang.datun}]后续示例仅以 JSONObject 说明。以下嵌套 JSON 数据可自动解析出字段groupid、data_voltage、data_current、ts、inuse、location_0_province、location_0_city、location_0_datun同时可以选择要解析的字段并为解析出的字段设置别名。{groupid: 170001, data: { voltage: 221V, current: 12.3 }, ts: 2023-12-18T22:12:00, inuse: true, location: [{province: beijing, city:chaoyang, street: datun}]}图嵌套 JSON 的解析与字段别名设置。2. Regex 正则表达式可以使用正则表达式中的命名捕获组named capture groups从任意字符串文本字段中提取多个字段。例如下图从 nginx 日志中提取访问 IP、时间戳、访问 URL 等字段(?ip\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b)\s-\s-\s\[(?ts\d{2}/\w{3}/\d{4}:\d{2}:\d{2}:\d{2}\s\\d{4})\]\s(?method[A-Z])\s(?url[^\s]).*(?status\d{3})\s(?length\d)图使用命名捕获组解析 nginx 日志。3. UDT 自定义解析脚本支持编写自定义 rhai 语法脚本解析输入数据语法参考 rhai 官方文档脚本目前仅支持 json 格式的原始数据。输入脚本中可使用参数data它是原始数据经 json 解析后的 Object Map输出输出数据必须是一个数组。例如某设备上报三相电压值需要分别写入三张子表待解析的数据如下{ ts: 2024-06-27 18:00:00, voltage: 220.1,220.3,221.1, dev_id: 8208891 }可使用如下脚本提取三路电压数据let v3 data[voltage].split(,); [ #{ts: data[ts], val: v3[0], dev_id: data[dev_id]}, #{ts: data[ts], val: v3[1], dev_id: data[dev_id]}, #{ts: data[ts], val: v3[2], dev_id: data[dev_id]} ]最终解析结果如下图所示图UDT 脚本解析三相电压数据的结果。抽取或拆分Extraction or Splitting解析后的数据可能仍不满足目标表的数据要求。例如智能电表采集的原始数据如下json 格式{groupid: 170001, voltage: 221V, current: 12.3, ts: 2023-12-18T22:12:00, inuse: true, location: beijing.chaoyang.datun} {groupid: 170001, voltage: 220V, current: 12.2, ts: 2023-12-18T22:12:02, inuse: true, location: beijing.chaoyang.datun} {groupid: 170001, voltage: 216V, current: 12.5, ts: 2023-12-18T22:12:04, inuse: false, location: beijing.chaoyang.datun}使用 json 规则解析后voltage 被解析为带单位的字符串而用户希望以 int 类型记录电压和电流值用于统计分析因此需要进一步拆分 voltage此外还希望将日期拆分为日期和时间分别存储。可对源字段ts使用 split 规则拆分为日期和时间并对字段voltage使用 regex 提取电压值与单位。split 规则需要设置分隔符delimiter和拆分次数number of splits拆分字段的命名规则为{原字段名}_{序号}。Regex 规则与解析过程相同使用命名捕获组命名提取出的字段。过滤Filtering过滤功能可设置过滤条件只有满足条件的数据行才会写入目标表。过滤条件表达式的结果必须为布尔类型。编写过滤条件前需要确定解析字段的类型并根据字段类型使用判断函数和比较运算符、、、、、!进行判断。字段类型与转换只有明确解析出的每个字段的类型才能使用正确的语法进行数据过滤。使用 json 规则解析出的字段会根据属性值自动设置类型bool 类型inuse: trueint 类型voltage: 220float 类型current : 12.2string 类型location: MX001使用 regex 规则解析出的数据均为 string 类型使用 split 和 regex 抽取或拆分出的数据为 string 类型。如果提取出的数据类型不是预期类型可以进行数据类型转换。常见的数据类型转换是将字符串转换为数值类型。支持的转换函数如下函数从类型到类型示例parse_intstringintparse_int(56) // 结果为整数 56parse_floatstringfloatparse_float(12.3) // 结果为浮点数 12.3条件表达式不同数据类型有其各自的条件表达式写法。1. BOOL 类型可以使用变量本身或!运算符。例如对于字段 inuse: true可写如下表达式inuse!inuse2. 数值类型int/float数值类型支持比较运算符、!、、、、。3. 字符串类型使用比较运算符比较字符串。字符串函数如下函数说明示例is_empty字符串为空时返回 trues.is_empty()contains检查字符串中是否包含某字符或子串s.contains(substring)starts_with字符串以某字符串开头时返回 trues.starts_with(prefix)ends_with字符串以某字符串结尾时返回 trues.ends_with(suffix)len返回字符串的字符数非字节数必须与比较运算符配合使用s.len 5 检查字符串长度是否为 5len 作为属性返回 int与前四个直接返回 bool 的函数不同4. 复合表达式多个条件表达式可使用逻辑运算符、||、!组合。例如以下表达式表示获取安装在北京且电压值大于 200 的智能电表数据location.starts_with(beijing) voltage 200映射Mapping映射是将解析、抽取或拆分出的源字段映射到目标表字段。可以直接映射也可以经过一定规则计算后再映射到目标表。选择目标超级表选择目标超级表后将加载该超级表的所有标签tag和列column。源字段会根据名称按映射规则自动映射到目标超级表的标签和列。映射规则支持的映射规则如下表规则说明mapping直接映射需要选择映射源字段value常量可输入字符串常量或数值常量输入的常量值直接存储generator生成器目前仅支持时间戳生成器 now存储时写入当前时间join字符串连接器可指定连接字符拼接选中的多个源字段format字符串格式化工具填写格式化字符串例如有三个源字段 year、month、day 分别表示年、月、日希望以 yyyy-MM-dd 日期格式存储可提供格式化字符串${year}-${month}-${day}。其中${}作为占位符占位符可以是源字段或字符串类型字段函数处理sum选择多个数值字段进行加法计算expr数值运算表达式可对数值字段进行更复杂的函数处理和数学运算1.format中支持的字符串处理函数函数说明示例pad(len, pad_chars)用字符或字符串将字符串填充到至少指定长度1.2.pad(5, 0) // 结果为 1.200trim去除字符串开头和结尾的空白 abc ee .trim() // 结果为 abc eesub_string(start_pos, len)提取子字符串两个参数1. 起始位置 0 时从末尾计数2.可选要提取的字符数≤ 0 时不提取省略时提取到末尾012345678.sub_string(5) // 5678012345678.sub_string(5, 2) // 56012345678.sub_string(-2) // 78replace(substring, replacement)用另一字符串替换子串012345678.replace(012, abc) // abc3456782.expr中的数学表达式基本数学运算支持加法、减法-、乘法*、除法/。例如数据源采集的温度值为摄氏度而目标数据库存储值为华氏度则需要对采集的温度数据进行换算。若源字段为temperature则使用表达式temperature * 1.8 32。数学表达式还支持使用数学函数如下表函数说明示例sin, cos, tan, sinh, cosh三角函数a.sin()asin, acos, atan, asinh, acosh反三角函数a.asin()sqrt平方根a.sqrt() // 4.sqrt() 2exp指数a.exp()ln, log对数a.ln() // e.ln() 1a.log() // 10.log() 1floor, ceiling, round, int, fraction取整a.floor() // (4.2).floor() 4a.ceiling() // (4.2).ceiling() 5a.round() // (4.2).round() 4a.int() // (4.2).int() 4a.fraction() // (4.2).fraction() 0.2子表名映射子表名是字符串可以使用映射规则中的字符串格式化format表达式定义。创建任务以 MQTT 为例下面以 MQTT 数据源为例说明如何创建 MQTT 类型的任务从 MQTT Broker 消费数据并写入 TDengine。登录 taosExplorer 后点击左侧导航栏中的 Data Writing数据写入进入任务列表页。在任务列表页点击 Add Data Source添加数据源进入任务创建页。输入任务名称后选择类型为 MQTT然后可以新建代理或选择已创建的代理。输入 MQTT broker 的 IP 地址和端口号例如192.168.1.100:1883。配置认证与 SSL 加密如果 MQTT broker 已启用用户认证在认证区域输入 MQTT broker 的用户名和密码如果 MQTT broker 已启用 SSL 加密可打开页面上的 SSL 证书开关上传 CA 证书以及客户端的证书和私钥文件。在 Collection Configuration采集配置区域可选择 MQTT 协议版本目前支持 3.1、3.1.1、5.0配置 Client ID 时注意如果对同一 MQTT broker 创建多个任务Client ID 应各不相同以避免冲突否则可能导致任务无法正常运行配置 topic 和 QoS 时使用topic name::QoS格式QoS 取值范围为 0、1、2分别表示至多一次、至少一次、恰好一次配置完上述信息后可点击 Check Connectivity检查连通性按钮检查配置若连通性检查失败请根据页面返回的具体错误提示进行修改。在从 MQTT broker 同步数据的过程中taosX 还支持对消息体中的字段进行提取、过滤和映射操作。在 Payload Transformation消息体转换下的文本框中可直接输入消息体样本或通过上传文件导入未来还将支持直接从已配置的服务器获取样本消息。对于消息体字段的提取目前支持 JSON 和正则表达式两种方式。对于简单的 key/value 格式 JSON 数据可直接点击提取按钮展示解析出的字段名对于复杂 JSON 数据可使用 JSON Path 提取感兴趣的字段使用正则表达式提取字段时需确保正则表达式的正确性。消息体字段解析完成后可根据解析出的字段名设置过滤规则只有符合过滤规则的数据才会写入 TDengine否则消息将被忽略。例如可配置过滤规则为 voltage 200表示只有电压大于 200V 的数据才会同步到 TDengine。最后配置消息体字段与超级表字段之间的映射规则后即可提交任务。除了基本映射外这里还可以对消息体字段的值进行转换例如使用表达式expr从原始消息体的电压和电流计算出功率后再写入 TDengine。提交任务后将自动返回任务列表页如果提交成功任务状态将切换为 Running如果提交失败可查看任务的运行日志activity log寻找错误原因。对于正在运行的任务点击指标metrics的查看按钮可查看任务的详细运行指标弹窗分为 2 个标签页分别显示任务多次运行的累计指标和本次运行的指标这些指标每 2 秒自动刷新一次。MQTT 连接的更多细节结合 MQTT 数据源完整配置指南连接与采集配置还包含以下要点TLS 校验有三种模式Disabled不校验 TLS 证书连接器先尝试 TCP失败后尝试不带证书校验的 TLS、单向认证使用 TLS 并校验服务器证书上传 CA 证书、双向认证使用双向 TLS上传 CA 证书、客户端证书和客户端私钥。Client ID输入标识符后将生成带taosx前缀的 client id例如输入foo生成taosxfoo。若开启末尾开关则会在taosx与输入标识符之间拼接当前任务的任务 id生成形如taosx100foo的 client id。连接同一 MQTT 地址的所有 client id 必须唯一。Keep Alive若 broker 在 keep alive 时间内未收到客户端消息将认为客户端已断开并关闭连接。Clean Session选择是否清除会话默认值为 true。Topics QoS Config格式为{topic_name}::{qos}如my_topic::0。MQTT 5.0 支持共享订阅多个客户端可订阅同一 topic 实现负载均衡格式为$share/{group_name}/{topic_name}::{qos}其中$share是固定前缀group_name是客户端组名类似 Kafka 的 consumer group。Topic Analysis主题解析解析 MQTT Topic 的每一级为对应变量名_表示解析时忽略当前级。例如 MQTT Topica//c对应解析规则v1/v2/_表示将第一级a赋给变量v1第二级值通配符代表任意值赋给变量v2第三级c被忽略。在后续的消息体解析中主题解析得到的变量也可参与各种转换和计算。Compression压缩配置消息体压缩算法taosX 接收消息后按对应算法解压。选项包括 none不压缩、gzip、snappy、lz4、zstd默认 none。Char Encoding字符编码配置消息体编码格式选项包括 UTF_8、GBK、GB18030、BIG5默认 UTF_8。高级选项Advanced OptionsAdvanced Options区域默认折叠点击展开。MQTT 和 SparkplugB 数据源通常提供以下选项字段名可能因连接器而异Message Queue Size消息队列大小指定接收缓冲区大小。如果队列已满且Cache Realtime Data未启用新到达的数据将被丢弃。设置为0可禁用缓冲。Maximum In-Process Batches最大在途批次指定可并发处理的批次数。达到该限制后连接器停止从接收队列取消息导致消息积压。最小值为1。Batch Size批次大小指定一次送入处理管道的消息数。与Batch Delay配合即使延迟未到批次满也会立即发送。最小值为1。Batch Delay批次延迟指定每批的毫秒级超时从该批第一条消息起算。超时后即使未达到Batch Size也会发送。最小值为1。Write Concurrency写入并发指定可并发写入 TDengine 的任务数。Cache Realtime Data缓存实时数据启用后消费的数据先写入本地文件由后台任务转发给下游。当下游处理跟不上时提供流量整形traffic shaping。积压数据消费完后缓存文件被删除。默认关闭。详见 Store and Forward。Cache Storage Directory缓存存储目录覆盖缓存文件目录。仅在启用Cache Realtime Data时生效否则默认使用 taosX 启动时配置的数据目录。启用Save Raw Data保存原始数据后还可配置Maximum Retention Days最大保留天数和Raw Data Storage Directory原始数据存储目录。异常处理策略Exception HandlingException Handling Strategy区域默认折叠点击展开。常用策略包括Archive归档将无效数据写入归档文件默认位于${data_dir}/tasks/id/datetime下不写入目标数据库。Discard丢弃忽略无效数据。Error报错报告错误。Cache缓存针对目标连接失败或资源不足将数据写入缓存文件待目标恢复后再摄入。可为以下场景配置策略场景可选策略目标连接超时Archive、Discard、Error、Cache目标数据库不存在Archive、Discard、Error表不存在Archive、Discard、Error、自动建表并重试主时间戳超出范围now - keep1到now 100yArchive、Discard、Error主时间戳为 nullArchive、Discard、Error、使用当前时间复合主键为 nullArchive、Discard、Error表名超过 192 字符Archive、Discard、Error、截断、截断并归档表名包含非法字符如.Archive、Discard、Error、用配置的字符串替换非法字符表名模板变量为 nullDiscard、变量留空、用配置的字符串替换列不存在Archive、Discard、Error、自动补列并重试列名超过 64 字符Archive、Discard、Error列值超过定义长度Archive、Discard、Error、截断、截断并归档Automatic Column Expansion可改为修改表结构后重试其他数据错误Archive、Discard、Error其他附加设置Connection Timeout连接超时目标连接超时秒数范围1~600。Temporary Storage Location临时存储位置相对${data_dir}/tasks/id/的路径。Archive Retention Days归档保留天数非负整数0表示不限。Archive Available Space归档可用空间0~655350表示不限。Archive Location归档位置相对${data_dir}/tasks/id/的路径。Archive Write Failure Strategy归档写入失败策略删除旧文件、丢弃数据、或报错并停止任务。任务管理在任务列表页可以对任务执行启动、停止、查看、删除、复制等操作还可以查看每个任务的运行状态包括写入记录数、流量等。健康状态Health Status从 v3.3.5.0 开始任务列表为每个运行中的任务显示健康状态。Advanced Options区域提供以下健康监控设置Health Check Duration健康检查时长计算任务状态的最近时间段。Busy State Threshold繁忙状态阈值排队项与写入队列容量的比值默认100%。Max Write Queue Length最大写入队列长度写入队列的最大容量。Write Error Threshold写入错误阈值健康检查期间允许的写入错误数超过即报错。任务列表可显示以下状态Ready就绪源与目标健康检查均通过可正常读写。Idle空闲监控周期内没有数据进入处理管道。Active活跃任务正常运行并处理数据。Pending挂起源健康但写入端正在等待且没有消息可写。Busy繁忙写入队列超过配置阈值。这可能表明存在性能瓶颈需要调整参数或资源但本身并不代表错误。Bounce抖动源与目标健康但写入错误超过阈值。这可能表明存在大量无效数据或数据丢失。SourceError源错误源无法读取工作负载会尝试重连。SinkError目标错误目标无法写入工作负载会尝试重连恢复后回到Ready。Fatal致命发生严重或不可恢复的错误。健康状态为空表示任务尚无数据进入。从检查点恢复任务Checkpoint 断点续传大多数 taosX 数据源可以从最后持久化的检查点恢复TDengine Query、MySQL、PostgreSQL、Oracle、Microsoft SQL Server、MongoDB持久化上次查询的时间戳并从该时间点恢复。TDengine Data Subscription依赖 TDengine 订阅进度。PI、InfluxDB、OpenTSDB持久化消息时间戳并从该时间点恢复。OPC UA、OPC DA、MQTT启用Cache Realtime Data时将消息持久化到磁盘。Kafka依赖 Kafka 的 consumer-offset 管理。CSV持久化文件名与已消费行数从下一行继续。AVEVA Historian查询历史数据时持久化最后一条消息的时间戳。SparkplugB目前不支持消息持久化与恢复。进阶实践完整的零代码接入流程如果你希望从一个端到端的视角快速跑通零代码接入可以参考快速入门章节 Zero-Code Data Ingestion 中的完整流程以 MQTT 为例准备环境确认 TDengine 服务运行、taosExplorer 可在浏览器访问、taosX 相关服务运行且 Explorer 左侧菜单包含Data In入口并能访问外部 MQTT broker该示例使用 EMQX 提供的公共 brokerbroker.emqx.io:1883。创建任务在 Explorer 中进入Data In点击 Add Source新建任务设置名称如quick_mqtt_meter、类型为MQTT、目标数据库如test_mqtt。配置连接填写 MQTT Host 与端口、协议版本3.1 / 3.1.1 / 5.0、Client ID可留空使用自动生成、Keep Alive、Clean Session、认证信息与 TLS 选项并将 Topics QoS Config 配置为tdengine/quickstart/meter::0格式为topic::QoS多个 topic 用逗号分隔。随后点击Check Connectivity验证连通性。配置消息体转换在 Payload Transformation 中输入 JSON 样例如智能电表读数点击识别/预览按钮确认 JSON 字段被正确提取。配置映射选择或创建目标超级表meters配置TIMESTAMP类型的ts列、DOUBLE类型的current/phase列、INT类型的voltage列以及groupid、location标签将 SubTableName 设置为t_{id}以按消息中的id生成子表名并完成其余字段的映射后提交任务。发送测试数据并验证任务状态变为Running后使用mosquitto_pub向同一 topic 发布消息mosquitto_pub -h broker.emqx.io -p 1883 -t tdengine/quickstart/meter -q 0 -m { ts: 2026-07-27T14:31:0008:00, id: 1, current: 10.58, phase: 1.41, voltage: 221, groupid: 7, location: beijing }参数说明-h/-p指定 broker 地址与端口-t指定发布 topic须与 Topics QoS Config 中的 topic 一致不带::QoS后缀-q指定 QoS-m指定消息体字段须与样例和映射规则一致。重复测试时请修改ts为当前时间以避免覆盖同一时间戳。然后在 Explorer 的数据浏览器或 shell 中查询验证SELECT tbname, ts, current, voltage, phase, groupid, location FROM test_mqtt.meters ORDER BY ts DESC LIMIT 5;若返回类似t_1 | 2026-07-27 14:31:00.000 | 10.5800 | 221 | 1.410 | 7 | beijing的行说明 MQTT 消息已通过 taosX 写入 TDengine。故障排查若连通性检查失败检查 taosX 所在主机能否访问 broker 端口、Host/Port/TLS/认证配置是否正确、企业网络或云安全组是否拦截出站 MQTT若任务运行但查询无数据检查发布 topic 是否与订阅完全一致、样例能否被识别且字段名与映射匹配、SubTableName 是否设置、目标库与超级表是否创建成功、查询的库名是否正确。延伸阅读本主题相关的更多资料可继续阅读当前仓库中的以下文档No-Code Data Ingestion 目录支持的数据源、ETL 规则、健康状态与断点续传本文主体。Zero-Code Data Ingestion 快速入门MQTT 端到端零代码接入示例。MQTT 完整配置指南MQTT 连接、TLS、采集、解析、映射与高级选项的完整说明。安装 taosX-Agent在数据源侧网络部署 Agent 代理采集。taosX-Agent 参考Agent 的部署、配置、运行与排障。Data Ingest and Delivery 章节总览零代码数据接入、零代码数据投递、数据流管理与迁移指南。其他数据源专项指南Kafka、CSV、OPC UA、InfluxDB、OpenTSDB、PI 等均位于 01-no-code-ingestion 目录 下。 /output文章【免费下载链接】TDengineHigh-performance, scalable time-series database designed for Industrial IoT (IIoT) scenarios项目地址: https://gitcode.com/GitHub_Trending/tde/TDengine创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考