TDengine 无代码接入 Pulsar-Tuya通过 taosExplorer 构建涂鸦云数据同步任务【免费下载链接】tdengineTDengine is an open source, high-performance, cloud native time-series database optimized for Internet of Things (IoT), Connected Cars, Industrial IoT and DevOps.项目地址: https://gitcode.com/taosdata/tdengine导读本文介绍如何通过 TDengine 的 taosExplorer 可视化界面零代码创建从涂鸦云定制版 Apache PulsarPulsar-Tuya集群到 TDengine 的数据迁移任务覆盖历史数据回填与实时数据接入两大场景。读完本文你将掌握添加 Pulsar-Tuya 数据源、配置连接与涂鸦平台认证、设置采集与 Payload 解析字段拆分、数据过滤、表映射以及异常处理策略的完整实操流程。企业版特性说明本文涉及的 Pulsar-Tuya 数据接入功能属于TDengine TSDB-Enterprise企业版专属能力TDengine TSDB-OSS 开源版本不包含该功能。相关声明可参见 关联文档头部。功能概述Pulsar-Tuya 与 TDengine 的对接价值Apache Pulsar 是云原生、开源、分布式的消息与流处理平台具备多租户、持久化消息存储、跨地域复制等能力。Pulsar-Tuya 是涂鸦智能Tuya基于开源 Apache Pulsar 定制而成的集群承载涂鸦物联网平台海量设备的上行数据设备状态、告警、遥测等。TDengine 作为时序数据库可以高效地从 Pulsar-Tuya 中读取消息并写入 TDengine 集群从而实现两类典型目标历史数据迁移将涂鸦平台中沉淀的历史设备数据批量回填到 TDengine用于后续离线分析、AI 建模或平台切换实时数据接入持续消费 Pulsar-Tuya 中的实时消息让 TDengine 中的时序数据与设备侧保持同步支撑实时监控大屏与告警系统。整个对接过程完全通过 taosExplorer 的图形界面完成无需编写任何消费代码属于 TDengine 无代码数据接入No-Code Data Ingestion体系的一部分。同一体系还覆盖 MQTT、Kafka、InfluxDB、OpenTSDB、MySQL、Oracle、OPC-UA 等数十种数据源可参见 无代码接入目录。操作前置条件在创建任务前请确认以下环境准备就绪已部署 TDengine 企业版集群且 taosExplorer 服务可用。taosExplorer 的 Web 界面默认运行在 TDengine 所在主机或 IP 的6060 端口。已具备涂鸦开放平台提供的Access Id与Access Key下文认证环节使用并确认目标执行环境如生产环境 prod。已确认 Pulsar-Tuya 集群的 Broker 地址与端口示例格式mqe.tuyaus.com:6650。如需在远端服务器执行采集数据源与 TDengine 不在同一网络可先部署 taosX-Agent安装方式参见 Install taosX-Agent。操作流程创建 Pulsar-Tuya 数据接入任务整体流程分为添加数据源 → 配置连接信息 → 配置认证机制 → 配置采集信息 → 配置 Payload 解析解析 / 字段拆分 / 数据过滤 / 表映射→ 配置高级选项 → 配置异常处理 → 提交任务。添加数据源按以下步骤在 taosExplorer 中新建任务在浏览器中打开 taosExplorerTDengine 所在主机或 IP 的 6060 端口。在左侧主菜单点击Data In数据接入然后点击Add Task添加任务。在Name字段中为数据接入任务填写唯一名称。在Type下拉列表中选择Pulsar-Tuya。可选如果该数据接入任务需要远端 Agent 执行从Agent下拉列表中选择 Agent也可以点击Create New Agent现场创建。从Target DB下拉列表中选择本任务写入数据的目标数据库也可以点击Create Database现场创建数据库。说明以上步骤来自 taosExplorer 的通用数据源接入模板源文件见 resources/_resources.mdx。配置连接信息在Connection Configuration区域填写Broker ServerBroker 服务器地址例如mqe.tuyaus.com:6650。此处只需填写一个有效的 Broker 服务器地址即可无需填写完整集群拓扑。taosX 会通过该地址完成 Pulsar 协议握手并自动发现集群。认证机制Pulsar-Tuya 使用涂鸦平台特有的认证体系与标准 Apache Pulsar 的 Basic Auth / JWT / mTLS 等认证方式不同。你需要在Authentication区域填写涂鸦平台提供的认证信息Access Id涂鸦开放平台颁发的访问 ID必填。Access Key涂鸦开放平台颁发的访问密钥必填。Execute Environment选择对应的执行环境例如生产环境prod不同环境对应不同的涂鸦服务端点。实现层面的便利点填写 Access Id / Access Key 后taosX 会自动推导 Pulsar 消费所需的Topic主题、Consumer Name消费者名和 Subscription Name订阅名无需用户手工指定涂鸦侧的 Pulsar 内部主题结构降低了对接门槛。配置采集信息在Collection Configuration区域填写采集任务相关参数。如前所述Pulsar 所需的 topics、consumer names、subscription names 会根据涂鸦平台提供的 Access Id 和 Access Key 自动生成你只需关注以下三项配置项说明默认值Timeout超时时间当超过该时长仍无法从 Pulsar 消费到数据时采集任务将退出设置为 0 时无限期等待直到有数据到达或发生错误0 msInitial Position起始位置选择开始消费数据的位置EarliestChar Encoding字符编码消息体编码格式。taosX 收到消息后按该编码解码消息体还原原始数据UTF_8Initial Position 两个选项的含义Earliest请求从最早位置开始消费适合历史数据回填场景Latest请求从最新位置开始消费适合只关心新数据的实时接入场景。Char Encoding 支持以下取值UTF_8、GBK、GB18030、BIG5。当消息体为中文 GBK 编码例如部分涂鸦设备厂商上报的旧协议数据时应选择对应编码避免解码乱码导致数据解析失败。配置完成后点击Check Connection连通性检查按钮验证数据源是否可用。配置 Payload 解析在Payload Parsing区域配置消息体Payload的解析逻辑这是将涂鸦消息转换为 TDengine 数据行的核心环节包含解析、字段拆分、数据过滤、表映射四个子步骤。解析获取样例数据有三种方式点击Retrieve from Server从服务器获取按钮直接从 Pulsar-Tuya 拉取样例数据点击File Upload文件上传按钮上传 CSV 文件获取样例数据在Message Body消息体输入框中手工粘贴 Pulsar 消息体样例。JSON 数据支持JSONObject或JSONArray两种形态以下数据均可使用 JSON 解析器解析单条 JSON 对象JSONObject{id: 1, message: hello-world} {id: 2, message: hello-world}JSON 数组JSONArray[{id: 1, message: hello-world},{id: 2, message: hello-world}]解析结果界面可参考 Payload 解析结果示意图。点击放大镜图标可查看解析结果预览预览界面见 解析结果预览示意图。字段拆分在Extract or Split from Columns从列中提取或拆分中填写要从消息体提取或拆分的字段。例如将message字段按分隔符-拆分成 2 段得到message_0和message_1操作方式为选择 split 提取器拆分器分隔符separator填-数量number填 2。点击Add可添加更多提取规则点击Delete可删除当前提取规则。界面参考 提取/拆分配置示意图点击放大镜图标可预览提取/拆分结果见 拆分结果预览示意图。数据过滤在Filter过滤条件中填写过滤规则。例如输入id ! 1则只有 id 不等于 1 的数据才会写入 TDengine。这意味着你可以在数据进入 TDengine 之前丢弃无效或噪声数据既节省存储又避免污染时序数据。点击Add可添加更多过滤规则点击Delete可删除当前过滤规则。界面参考 过滤条件配置示意图点击放大镜图标可预览过滤结果见 过滤结果预览示意图。表映射在Target Supertable目标超级表下拉框中选择目标超级表或点击右侧的Create Supertable创建超级表按钮现场创建。在Mapping映射部分填写目标超级表下的子表名例如t_{id}——{id}是模板变量运行时会被消息中id字段的实际值替换实现一台设备或一个数据实体一张子表的经典时序建模按需填写映射规则映射支持设置默认值当字段缺失时使用默认值填充。界面参考 表映射配置示意图。点击Preview可查看映射结果见 映射结果预览示意图。配置高级选项Advanced Options高级选项区域默认折叠点击右侧的展开后可以看到更多与采集吞吐、批处理、连接相关的调优参数。界面参考 折叠状态示意图 与 展开状态示意图。实际参数项以 taosExplorer 当前版本界面为准一般包括批量大小、刷新间隔、并发度、重试等项用于在吞吐量与延迟之间取得平衡。配置异常处理Exception Handling Strategy异常处理策略区域默认折叠点击展开。taosX 针对写入目标端可能出现的各类异常提供了精细化的处置策略核心策略如下策略行为说明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主时间戳为空Archive / Discard / Error /使用当前时间复合主键为空Archive / Discard / Error表名超过 192 个字符Archive / Discard / Error /截断/ 截断并归档表名含非法字符如.Archive / Discard / Error /将非法字符替换为指定字符串表名模板变量为空Discard / 置空变量 /替换为指定字符串列不存在Archive / Discard / Error /自动加列并重试列名超过 64 个字符Archive / Discard / Error列值超出定义长度Archive / Discard / Error / 截断 / 截断并归档也可启用Automatic Column Expansion自动扩列修改表结构后重试其他数据错误Archive / Discard / Error除上述逐类策略外还有以下全局设置项Connection Timeout目标连接超时时间单位秒取值范围1600Temporary Storage Location临时存储位置为相对于${data_dir}/tasks/id/的路径Archive Retention Days归档保留天数非负整数0表示不限制Archive Available Space归档可用空间取值范围0655350表示不限制Archive Location归档位置为相对于${data_dir}/tasks/id/的路径Archive Write Failure Strategy归档写入失败时的策略可选删除旧文件 / 丢弃数据 / 报错并停止任务。上述异常处理策略明细来自 resources/_03-exception-handling-strategy.mdx为 Pulsar-Tuya、Kafka、MQTT 等所有无代码接入任务共用的通用异常处理模型。完成任务创建点击Submit提交按钮即完成 Pulsar-Tuya 到 TDengine 数据同步任务的创建。返回Data Source List数据源列表页面可查看任务执行状态、消费进度与异常记录并据此调整超时时间、起始位置、过滤规则等配置。实践要点与常见问题历史迁移与实时接入的起始位置选择做历史回填时选Earliest并合理设置 Timeout 为 0 以等待数据做实时接入时可选Latest跳过存量数据两者的本质差异只是 Pulsar 消费起点不同其余解析、映射流程完全一致。编码一致性是关键消息体编码设置UTF_8/GBK/GB18030/BIG5必须与实际数据一致否则中文等非 ASCII 内容在解析阶段就可能出现乱码或解析失败进而触发异常处理策略。过滤前置优于入库后清理利用 Filter 在写入前丢弃噪声数据可显著降低 TDengine 存储膨胀配合表映射的默认值机制可保证模板变量缺失时任务仍能稳定建表写入。异常处理兜底生产环境建议对“表不存在”与“列不存在”配置自动建表/自动加列并重试对“主时间戳为空”配置使用当前时间同时将 Archive 保留天数设为非零值便于事后追溯被丢弃的数据。延伸阅读标准 Apache Pulsar 接入支持 Basic Auth / JWT / mTLS / 自定义认证需手工指定 Topic参见 Pulsar 接入指南无代码数据接入的整体概念与全部数据源列表见 No-Code Data Ingestion 索引远端采集 Agent 的安装与部署见 Install taosX-Agent异常处理策略的通用定义见 resources/_03-exception-handling-strategy.mdx。【免费下载链接】tdengineTDengine is an open source, high-performance, cloud native time-series database optimized for Internet of Things (IoT), Connected Cars, Industrial IoT and DevOps.项目地址: https://gitcode.com/taosdata/tdengine创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
