★ FEATURED ARTICLE 1. 先把同步这件事想清楚时序库的同步为什么比关系库更麻烦这篇文章是这个 IoTDB 实操系列里专门讲数据同步的一篇主题是 Apache IoTDB 的 Pipe 机制以及基于插件体系做自定义同步的完整指南。先从我自己的一段经历说起之前帮一个新能源平台做集群迁移源端累积了三年多的测点数据好几千亿个点如果按老办法导出再导入光校验就得折腾一周。后来用 IoTDB 1.2 的 Pipe 一条 SQL 建管道history 拉存量、realtime 接增量半天时间目标端数据就已经可以对外服务了。这种体验在数据库同步场景里其实很少见也因此值得专门写一篇。内容会覆盖时序数据库同步的基本思路、Pipe 的 Source/Processor/Sink 三段架构、从零创建一个同步管道、以及自定义插件同步的完整步骤适合正在做时序数据容灾、双活、边云汇聚或跨集群迁移的架构师、DBA 和平台开发。为了有个参照系我也会拿 SQL Server 的 CDC/CT 和 MongoDB 生态里的 MongoShake 做对比毕竟很多人对“数据同步”的第一印象是从关系库那边来的。1.1 关系库与 NoSQL 的同步方案CDC、CT、oplog 重放先给几个参照物。SQL Server 的数据同步有非常经典的两种方式CDCChange Data Capture变更数据捕获和 CTChange Tracking变更跟踪。CDC 的思路是后台进程持续解析事务日志把每一次插入、更新、删除解析成结构化变更记录。好处是你拿到的不仅是“谁变了”还有变更前后的完整数据非常适合数据仓库增量抽取、审计和跨库同步。代价是解析日志有额外开销链路会带来延迟。CT 就轻量得多它只维护一个版本号标记记录“哪些行在某个版本之后发生过变化”不保存变化的内容。业务端需要定时查询变更版本拿到变更行号之后再自己回表捞数据。省资源但拿不到最终数据适合“只想知道哪些数据变了再自己处理”的场景。再看 MongoDB 生态里常用的 MongoShake。它的原理是订阅 MongoDB 的主节点 oplog把变更事件解析后投递给目标端可以是另一个 MongoDB 集群也可以是其他异构存储属于典型的日志消费型同步工具。这三者的共同点一句话能概括源的变更先进日志工具消费日志、解析事件、按序投递。记住这个模型后面理解 IoTDB 的 Pipe 会非常顺。顺带说一句市面上时序数据库还有不少InfluxDB、TDengine、TimescaleDB、Prometheus、QuestDB、KDB 等每家的同步方案差异很大有的是商业版功能有的依赖外部工具真正把同步做成数据库内建能力、还开放插件接口的并不多IoTDB 的 Pipe 算是其中做得比较完整的一套。1.2 时序数据的写入形态决定了同步不能照搬关系库思路既然参照物有了接下来要回答的问题是为什么不直接把 SQL Server CDC 那套搬过来用关键在于时序数据库的写入形态和关系库完全不一样。关系库的最小单元是“一行数据”有主键、有约束、有事务时序库的最小单元是一个又一个数据点按时间序列时间线组织同一条测点下面不断追加不同时间戳的值。这意味着几件事底层存储是 TsFile 文件加 WAL不是传统事务日志。想把“某个测点新写了一个值”从存储层单独解析出来必须重新设计一套事件模型无法直接套用 SQL 日志解析。写入体量不在一个量级。一个大型风电场一天可能写入上千亿个点如果按“每条变更一条事件”的方式同步事件本身的序列化和投递开销就会压垮链路。乱序写入是常态。设备侧网络抖动、边缘缓存补发都会导致晚到的数据点时间戳更早。同步机制如果要求严格按提交序回放反而会制造大量等待和阻塞。所以 IoTDB 在设计 Pipe 时必须同时处理三件事从 WAL 实时捕获新事件、从历史文件回溯存量数据、以及容忍目标端乱序批量写入。它不是把数据库日志复制一下而是构建了一条完整的、面向时序数据形态的同步管道。1.3 Pipe 机制解决的四个核心问题正式给个定位Pipe 是 Apache IoTDB 从 1.0 版本开始重构并持续演进的内置数据同步框架。它在数据库内部解决了四个典型问题存量迁移旧集群到新集群不需要导 dump 再 load建一条带 history 的管道历史数据就能按时间片流式搬过去。增量实时同步创建管道之后新写入的数据持续投递到目标端中间无需人工干预。跨网络形态适配既能通过 Thrift RPC 直接写对端 IoTDB也能落地成文件应对物理隔离环境。可编程扩展通过 pipe-api 插件接口自定义处理逻辑与投递目标不用改内核代码。下面我把 Pipe 拆成三段逐一说这也是理解插件同步的地基。2. Pipe 三段式管道拆解Source、Processor、Sink 各自承担什么Pipe 的命名很形象它就是一段一段的管道拼接。你可以在 create pipe 语句里显式声明三段也可以省略让系统用默认值。默认三段分别是 iotdb-source、do-nothing-processor、iotdb-thrift-sink这个组合覆盖了绝大多数“IoTDB 到 IoTDB”的同步需求。2.1 Source存量数据与增量数据的双通道捕获Source 是整个管道的进水口决定数据从哪里来。我习惯把它的工作拆成两个通道来看增量通道和存量通道。增量通道监听 WAL 中的新写入事件解析后转成管道内部的事件对象。你可以把它理解成“数据库自己的 CDC 读取端”——只不过它读的不是 SQL 事务日志而是 IoTDB 的写入日志。存量通道则是直接扫描本地已有的 TsFile 数据文件按时间范围切片读取并把读取进度持久化下来。source 段常用的参数就几个source插件类型默认 iotdb-source一般不用换realtime是否同步新写入的数据true/falsehistory是否回溯同步历史数据true/falsestart-time/end-time历史数据的时间范围ISO 8601 格式注意带时区including/excluding部分版本支持的测点路径过滤规则按需使用。这里有个新手常踩的误区以为“得等 history 跑完再手动开 realtime”。实际不用。history 和 realtime 可以同时开启IoTDB 会一边把存量按时间片搬过去一边把新数据接上两边各自维护进度。实测大存量迁移时这种并行追平比串行“先迁历史再等增量”快得多。2.2 Processor在数据到达对端之前拦一道干点活Processor 是管道中间的处理段也是插件体系里最常被扩展的地方。默认的 do-nothing-processor 就是原样放行不做任何加工零额外开销。很多人第一反应是“那我永远用默认不就行了”没问题但如果同步的数据在途中需要被改造Processor 就是那个唯一的改造点。可以做的事情包括但不限于按测点路径过滤比如只同步wind_*前缀的时间序列其他一概丢弃字段级别的裁剪或格式转换比如去掉 tag、把摄氏温度换算成华氏路由改写比如给目标路径统一加前缀避免边云汇聚时多个边缘端的设备 ID 互相冲突敏感信息脱敏比如对设备 ID 做哈希处理再写入目标端。用过流处理框架比如 Kafka Streams的人会觉得很熟悉无非就是 filter、map、transform 的组合。区别在于 Processor 操作的对象是批量时序事件不是单条消息所以性能模型上更接近批处理。2.3 Sink数据出口决定你能把数据运到哪里Sink 是管道末端负责把数据投递出去。IoTDB 1.2 内置了两种典型的 Sinkiotdb-thrift-sink通过 Thrift RPC 直接写入另一个 IoTDB 集群。这是默认选择核心参数是node-urls对端 DataNode 地址多个地址用分号分隔、username、password。传输采用批量 Tablet 形式网络利用率比较高。iotdb-air-gap-sink专门应对不能开网络端口的隔离场景。它把同步数据滚动写成本地文件物理隔离环境下由运维用摆渡方式把文件带到目标环境再导入。适合等保要求高、专网隔离这类生产环境。选 Sink 的本质是在“实时性”和“可达性”之间做取舍。网络能通走 thrift-sink 最简单网络不通就老老实实走 air-gap 文件摆渡不要硬开端口安全部门不会同意的。2.4 一条数据从写入到目标端落地的完整旅程把三段拼起来一条数据被同步的完整过程是这样的业务执行 insert 写入源端 IoTDB数据先落 WALSource 的增量通道消费 WAL 事件解析成管道内部事件Processor 按配置处理事件默认透传Sink 把事件批量投递给目标端目标端写入成功之后返回确认管道收到确认才把这一批事件的进度 checkpoint 往前走一格。注意第 5 步这是整个机制的灵魂。管道坚持“只有对端确认才推进进度”保证了断点续传的语义。管道中途异常中断重启后从最后一个已确认位置继续不会把已经同步过的数据整段重发。提示排查“数据对不上”问题时先回想这个五步模型问题一定出在某一段源端没抓到、中间处理挂了、目标端没写进或者进度没推进。3. 最快上手一条 SQL 把两个 IoTDB 实例接起来理论讲完直接上手。这一节我用一套可以在本地复现的配置把两个 IoTDB 实例用一条 SQL 接起来。3.1 动手前的环境检查假设你已经有源端和目标端两套 IoTDB 实例版本以 1.2.x 为例。同步本身是“写入即转发”目标端不需要预先建好时间序列写入时会自动创建。动手前确认三件事两端大版本尽量一致。跨大版本不是百分百不能同步但事件格式可能存在不兼容生产环境我不会冒险。目标端需要给同步账号开写权限。最省事先用 root 把链路跑通后续再收敛成专用账号。集群环境注意管道是节点级别运行的。多 DataNode 场景下需要确认每个 DataNode 上的管道都已经建好可以在任意节点执行 show pipes 检查整体情况。3.2 创建同步管道下面是我在 1.2.x 上验证过的创建语句create pipe p_to_backup with source ( source iotdb-source, realtime true, history true, start-time 2024-01-01T00:00:0008:00 ), processor ( processor do-nothing-processor ), sink ( sink iotdb-thrift-sink, node-urls 10.0.1.10:6667, username root, password root );各参数的含义拆开看historytrue且不写 end-time表示把从 start-time 到当前的所有历史数据按时间顺序同步过去一直扫到最新。整库迁移不要 start-time系统会从最早数据开始。realtimetrue创建完成之后的新写入持续同步。我建议存量加增量同时开一步到位省得后面再补一条管道。start-time注意带时区。写成2024-01-01T00:00:0008:00和2023-12-31T16:00:00Z是同一个时间搞混时区会漏数据。processor先用默认不要一上来就加自定义插件基线跑通最重要。node-urls填目标端 DataNode 的地址和 6667 端口。如果目标端是集群可以写多个地址10.0.1.10:6667;10.0.1.11:6667。3.3 管道的生命周期管理创建完成之后第一件事是确认状态show pipes; show pipe p_to_backup;show pipe 会输出管道的运行状态、任务数、延迟等关键信息。如果状态不是 RUNNING先看日志再往下走不要急着删管道重建。日常管理命令很简单stop pipe p_to_backup; -- 暂停定义保留 start pipe p_to_backup; -- 恢复执行 drop pipe p_to_backup; -- 删除管道定义这里有个值得注意的细节drop 之前尽量先 stop。在同步还在大规模灌数据时直接 drop目标端可能出现半截写入的中间态虽然不影响数据正确性但会给后续检查制造噪音。3.4 三种常见部署拓扑怎么配通道建好之后落地到具体业务无非就三种拓扑主备容灾一主一备源端建一条管道指向备库备库只读。主库故障时切到备库管道可以 stop 掉避免浪费资源。这种最简单不需要额外配置。边云汇聚多对一多个边缘 IoTDB 各建管道指向中心集群难点在路径冲突。多个边缘端可能都有同名的设备测点直接在中心端会互相覆盖。我的做法是用 Processor 给每条管道写入的目标路径加不同的前缀从管道上做隔离比到中心之后再想办法区分要省事。双活 / 读写分离A-B 双向A 到 B 一条B 到 A 一条。最大的坑是回环A 的数据同步到 BB 端把它当成新写入再被 B 到 A 的管道同步回 A形成无限循环。我的经验是从架构层面解决而不是指望数据库去重——让两端只允许本地写入不同的设备或路径段从源头切断交叉路径。4. 插件同步指南内置插件不够用时自己写一个前面讲的是“开箱即用”的部分。这一节进入标题里另一个关键词插件同步。当你发现内置的 Source、Processor、Sink 不够用需要自己定义同步逻辑时就进入这个环节了。4.1 什么场景需要进入插件开发模式内置三段在“IoTDB 到 IoTDB”的同步场景里确实够用但下面三类情况就必须写插件目标端不是 IoTDB。数据要同步到 Kafka、RocketMQ、MySQL、对象存储或者自研平台这时候最干净的做法是写一个自定义 Sink。数据需要在途中做业务转换。单位换算、按区域分流、敏感字段脱敏、路径前缀改写这些都是 Processor 的职责。硬要用外部 ETL 工具绕一圈延迟和复杂度都会上来。同步链路上需要观测能力。比如每条事件打一条日志、输出指标、触发告警同样可以做成一个轻量 Processor 插件挂在管道里。4.2 手写一个自定义 Processor测点过滤示例先讲骨架。一个 Pipe 插件本质上是依赖 pipe-api 模块的 Java 工程实现对应接口打成 jar 放到 DataNode 扩展目录在 create pipe 时用全类名引用。Maven 依赖长这样版本换成你实际目标版本的坐标dependency groupIdorg.apache.iotdb/groupId artifactIdiotdb-pipe-api/artifactId version1.2.2/version scopeprovided/scope /dependency以“只同步特定前缀测点”的 Processor 为例核心逻辑如下。接口方法为了讲清思路做了简化实际签名以你拉取的 pipe-api 源码为准package com.example.pipe; import org.apache.iotdb.pipe.api.PipeProcessor; public class PathFilterProcessor implements PipeProcessor { private String allowPrefix; Override public void validate(PipeParameterValidator validator) { String v validator.getAttributes().getString(allow-prefix); if (v null || v.isBlank()) { throw new IllegalArgumentException(allow-prefix must be set); } } Override public void customize(PipeProcessorRuntimeConfiguration config) { this.allowPrefix config.getEvaluator().getString(allow-prefix); } Override public void process(PipeData data) { if (data.getPath().startsWith(allowPrefix)) { data.forward(); } else { data.drop(); } } }接口虽然简化了但你要记住的关键点不变validate 校验参数、customize 读取 create pipe 里传入的配置、process 决定每批事件是放行、丢弃还是改写。特别提醒事件在自定义 Processor 里必须明确调用放行方法否则后面的 Sink 什么都收不到而且不会报错排查时很容易懵。4.3 手写一个自定义 Sink转发到消息队列再举一个更贴合“插件同步”诉求的例子目标端不是 IoTDB而是 Kafka。package com.example.pipe; import org.apache.iotdb.pipe.api.PipeSink; public class KafkaSink implements PipeSink { private KafkaProducerString, byte[] producer; private String topic; Override public void customize(PipeSinkRuntimeConfiguration config) { this.topic config.getEvaluator().getString(topic); // 用 config 提供的 bootstrap.servers 等参数初始化 producer } Override public void transfer(PipeData data) { producer.send(new ProducerRecord(topic, data.toBytes())); } Override public void close() { producer.close(); } }这个例子里最关键的也是最有教训的一点资源生命周期。KafkaProducer 这类连接型资源必须在 customize 或 start 阶段初始化、close 阶段释放。如果写进 transfer 里每批事件都 new 一个 Producer同步吞吐会肉眼可见地崩掉而且大概率把目标端连接数打满。4.4 插件的部署与使用完整流程插件写好之后部署流程是固定的mvn clean package打出 jar把 jar 复制到每个 DataNode 的ext/pipe目录不同版本扩展目录名略有差异以官方文档为准重启 DataNode确保插件类被加载。有些版本支持动态加载但重启是最稳妥的做法在 create pipe 语句的对应段里把插件名换成你的全类名并传入插件需要的参数create pipe p_kafka with source ( source iotdb-source, realtime true ), processor ( processor do-nothing-processor ), sink ( sink com.example.pipe.KafkaSink, bootstrap.servers 10.0.2.10:9092, topic iotdb-raw );用 show pipes 确认状态后往源端写一条数据去 Kafka 消费验证。注意自定义插件的加载依赖 DataNode 的 classpath。遇到“插件放了却没生效”第一件事去 DataNode 日志搜 ClassNotFoundException第二件事确认 ext/pipe 路径下没有因为手动解压多套一层目录。5. 生产环境实战笔记从踩坑到监控优化管道跑起来只是开始真正考验人的是长期稳定运行。这一节把我实际踩过的、以及帮别人排查过的典型问题整理成笔记。5.1 最容易翻车的几个典型瞬间先列高频问题清单都是真实发生过的目标端磁盘写满Sink 写失败不断重试管道进入 ERROR 状态。磁盘恢复以后需要手动 start pipe不要指望它会自动恢复。history 同步的 start-time 时区配错导致漏了前一个小时的数据。排查时先看管道进度和源端最早数据的时间戳对不上就基本锁定是时区问题。双向同步没做路径隔离A 同步给 B 的数据在 B 端被当成新写入又被 B 到 A 的管道同步回 A形成回环风暴。这个从架构上隔离最靠谱。自定义 Processor 里抛异常没处理整个事件流被阻塞。症状是源端业务正常、目标端迟迟没数据定位时看 DataNode 日志里的 pipe-plugin 异常栈。5.2 故障排查的完整链路我自己排查询题的顺序是固定的按这个链路走90% 的问题能定位到具体一段show pipe p1看总体状态和最近延迟先判断管道是不是活的看源端 DataNode 的 pipe 相关日志logs 目录下按天滚动直接搜 pipe 关键字看目标端 DataNode 日志确认有没有收到连接、写入有没有失败如果两端日志都正常再回到数据本身源端查最新写入目标端查对应测点看时间差是否合理最后才怀疑进度问题确认管道的 checkpoint 是否长期不推进。这个顺序的核心逻辑是从外围到核心逐步缩小范围。最忌讳的是还没看日志就 drop 管道重建那等于销毁了现场。管道定义重建很容易但日志和进度信息一丢下次再出问题还是抓瞎。5.3 延迟、吞吐与参数调优同步管道的性能调优方向上无非四个维度批量大小与批量时间Sink 端攒批发送吞吐高但延迟会稍微增加。追求低延迟就调小批量时间追求吞吐就调大。并发通道数单管道吞吐不够时优先看 Source 侧读历史文件的并发通道数。多测点大批量场景加并发通常是立竿见影的。目标端写入能力同步的终点还是写目标库。目标端如果有 TTL 清理、压缩、乱序合并策略都会成为隐形瓶颈规格上不要比源端差。网络参数跨机房同步要配置连接超时和重试次数避免长链路抖动把管道打成 ERROR。具体参数名以你所用版本官方手册为准方向就是“别让一次性抖动变成持久故障”。5.4 我现在默认使用的稳定性配置清单最后给一份基线建议都是我经历过多次之后沉淀下来的默认做法架构上默认只做单向业务同步双活必须做路径隔离不赌数据库的去重能力监控上把 show pipe 的延迟和状态接入告警IoTDB 的 metrics 可以对接 Prometheus或者脚本轮询 show pipe 输出至少保证异常半小时内能被发现容量上目标端预留源端写入峰值 1.2 到 1.5 倍的磁盘和内存余量同步是有放大效应的演练上每个月做一次 stop/start pipe 加目标端断网恢复演练确认断点续传在你自己环境里真的可用别到灾备切换那天才第一次验证文档上每条管道建完在平台描述里写清楚源地址、目标地址、过滤规则、创建人和时间。半年之后没人记得这条管道是谁建来干什么的是运维事故的开始。最后说点个人体会。我最早用 IoTDB 做同步时总想着“同步不就该靠独立同步工具吗”后来把 Pipe 的 Source、Processor、Sink 三段拆开看明白之后发现它其实是把“日志捕获、流式处理、连接器”这一整套思路内聚到了数据库自身和关系库的 CDC、MongoDB 的 oplog 同步在精神上是同一件事。插件接口的存在让这套机制不至于变成黑盒遇到内置能力不够的同步场景先写一个最小的 Processor 或 Sink 验证事件流再逐步加逻辑比一开始就设计一个全功能插件要稳妥得多。如果你也在生产环境跑 Pipe建议按上面的排查链路把问题过一遍尤其是那些“看起来没报错但数据就是没到”的情况多半都出在进度推进和插件事件处理这两个容易被忽略的环节上。
阅读完成 · 觉得有帮助？
)
