引言
进入工业物联网（IIoT）时代，如何把海量时序数据用起来、挖出新价值，已经成为企业推进数字化转型的关键一环。以后产生的数据规模会持续激增。在这样的背景下，怎样在多集群、跨地域的场景下实现数据高效、稳定同步，也就成了必须解决的难题。 Apache IoTDB 是一款专门面向时序数据场景打造的开源数据库，它自带的 Pipe 数据同步功能，采用 “抽取 – 处理 – 发送” 三段式设计，能够以毫秒级的延迟完成数据复制和灾备，很好地应对了跨节点、跨区域的数据同步需求。
Apache IoTDB 时序数据库【系列篇章】：
| 1 | Apache IoTDB（1）：时序数据库介绍与单机版安装部署指南 |
| 2 | Apache IoTDB（2）：时序数据库 IoTDB 集群安装部署的技术优势与适用场景分析 |
| 3 | Apache IoTDB（3）：时序数据库 IoTDB Docker部署从单机到集群的全场景部署与实践指南 |
| 4 | Apache IoTDB（4）：深度解析时序数据库 IoTDB 在Kubernetes 集群中的部署与实践指南 |
| 5 | Apache IoTDB（5）：深度解析时序数据库 IoTDB 中 AINode 工具的部署与实践 |
| 6 | Apache IoTDB（6）：深入解析数据库管理操作——增删改查与异构数据库实战指南 |
| 7 | Apache IoTDB（7）：设备模板管理——工业物联网元数据标准化的破局之道 |
| 8 | Apache IoTDB（8）：时间序列管理——从创建到分析的实战指南 |
| 9 | Apache IoTDB（9）：数据库操作——数据写入从CLI到集群部署的六种实战 |
| 10 | Apache IoTDB（10）：数据库操作——从查询到优化的全链路实践指南 |
| 11 | Apache IoTDB（11）：分段聚合深度解析——从原理到实战的完整指南 |
| 12 | Apache IoTDB（12）：深度解析时序数据聚合的GROUP BY与HAVING子句 |
| 13 | Apache IoTDB（13）：数据处理的双刃剑——FILL空值填充与LIMIT/SLIMIT分页查询实战指南 |
| 14 | Apache IoTDB（14）：IoTDB结果集排序与查询对齐模式——ORDER BY与ALIGN BY DEVICE使用 |
| 15 | Apache IoTDB（15）：IoTDB查询写回（INTO子句）深度解析——从语法到实战的ETL全链路指南 |
| 16 | Apache IoTDB（16）：数据删除从单点精准清除到企业级数据生命周期管理 |
| 17 | Apache IoTDB（17）：IoTDB数据保留时间管理从TTL设置到智能数据控制 |
本文将从Pipe机制的工作原理，讲解IoTDB数据同步的全流程操作。涵盖功能概述、任务管理、插件扩展，友友们可参考并构建高可靠、低延迟的时序数据同步链路。
1. 功能概述
1.1 Pipe三阶段架构
一个数据同步任务包含 3 个阶段： 
抽取（Source）阶段：该部分用于从源 IoTDB 抽取数据，在 SQL 语句中的 source 部分定义 处理（Process）阶段：该部分用于处理从源 IoTDB 抽取出的数据，在 SQL 语句中的 processor 部分定义 发送（Sink）阶段：该部分用于向目标 IoTDB 发送数据，在 SQL 语句中的 sink 部分定义
通过 SQL 语句声明式地配置 3 个部分的具体内容，可实现灵活的数据同步能力。目前数据同步支持以下信息的同步，您可以在创建同步任务时对同步范围进行选择（默认选择 data.insert，即同步新写入的数据）
1.2 功能限制
元数据（schema）、权限（auth）同步功能存在如下限制：
使用元数据同步时，要求Schema region、ConfigNode 的共识协议必须为默认的 ratis 协议，即iotdb-system.properties配置文件中是否包含config_node_consensus_protocol_class=org.apache.iotdb.consensus.ratis.RatisConsensus、schema_region_consensus_protocol_class=org.apache.iotdb.consensus.ratis.RatisConsensus，不包含即为默认值ratis 协议。
为了防止潜在的冲突，请在开启元数据同步时关闭接收端自动创建元数据功能。可通过修改 iotdb-system.properties配置文件中的enable_auto_create_schema配置项为 false，关闭元数据自动创建功能。
开启元数据同步时，不支持使用自定义插件。
在进行数据同步任务时，请避免执行任何删除操作，防止两端状态不一致。
2. 创建任务
数据同步任务有三种状态：RUNNING、STOPPED 和 DROPPED
任务状态转换图： 
创建后任务会直接启动，同时当任务发生异常停止后，系统会自动尝试重启任务。
使用 CREATE PIPE 语句来创建一条数据同步任务，下列属性中PipeId和sink必填，source和processor为选填项，输入 SQL 时注意 SOURCE与 SINK 插件顺序不能替换。
示例：
CREATE PIPE [IF NOT EXISTS] <PipeId> — PipeId 是能够唯一标定任务的名字
— 数据抽取插件，可选插件
WITH SOURCE (
[<parameter> = <value>,],
)
— 数据处理插件，可选插件
WITH PROCESSOR (
[<parameter> = <value>,],
)
— 数据连接插件，必填插件
WITH SINK (
[<parameter> = <value>,],
)
IF NOT EXISTS 语义：用于创建操作中，确保当指定 Pipe 不存在时，执行创建命令，防止因尝试创建已存在的 Pipe 而导致报错。 注意：V2.0.8-beta 起，创建一个全量数据同步 Pipe （例如 Pipeid : alldatapipe）时，系统会自动将其拆分为两个独立的 Pipe： 历史 Pipe：PipeId 为原名称加 _history后缀（如 alldatapipe_history），source 参数默认携带 ‘realtime.enable’=‘false’, ‘inclusion’=‘data.insert’, ‘inclusion.exclusion’=‘’ 实时 Pipe：PipeId 为原名称加 _realtime后缀（如 alldatapipe_realtime），source 参数默认携带 ‘history.enable’=‘false’ ，若配置了元数据同步，则由实时 Pipe 负责发送
创建成功后，原 PipeId（如 alldatapipe）将不再作为有效标识符。在进行启动、停止、删除、查看等任务操作时，必须使用拆分后的独立 PipeId（即 *_history或 *_realtime）
3. 开始与停止任务
语法
START PIPE<PipeId>
示例
启动任务，使用START PIPE语句手动启动任务
START PIPE Beijing_To_Shanghai;
停止任务，使用STOP PIPE语句暂停任务
STOP PIPE Beijing_To_Shanghai;
状态监控
使用SHOW PIPES命令查看所有任务
SHOW PIPES;
结果
+———————+———+———————+———————+
| PipeName | Status | CreateTime | LastActiveTime |
+———————+———+———————+———————+
| Beijing_To_Shanghai | RUNNING | 2025-09-27 08:30:00 | 2025-09-27 08:35:22 |
+———————+———+———————+———————+
如果需要查看特定任务，使用SHOW PIPE语句获取信息
SHOW PIPE Beijing_To_Shanghai;
4. 删除和查看任务
4.1 删除任务
语法
DROP PIPE [IF EXISTS] <PipeId>
示例
删除任务，使用DROP PIPE语句删除任务
DROP PIPE IF EXISTS Beijing_To_Shanghai;
强制清理，使用CLEAN_EXPIRED_DATA清理过期数据
CALL CLEAN_EXPIRED_DATA();
IF EXISTS 语义：用于删除操作中，确保当指定 Pipe 存在时，执行删除命令，防止因尝试删除不存在的 Pipe 而导致报错。
删除任务不需要先停止同步任务。
4.1 删除任务
语法
查看所有任务
SHOW PIPES
查看指定任务
SHOW PIPE <PipeId>
结果示例
+——————————–+———————–+——-+———-+————-+———————————————————–+—————-+——————-+————————-+
| ID| CreationTime| State|PipeSource|PipeProcessor| PipeSink|ExceptionMessage|RemainingEventCount|EstimatedRemainingSeconds|
+——————————–+———————–+——-+———-+————-+———————————————————–+—————-+——————-+————————-+
|59abf95db892428b9d01c5fa318014ea|2024-06-17T14:03:44.189|RUNNING| {}| {}|{sink=iotdb-thrift-sink, sink.ip=127.0.0.1, sink.port=6668}| | 128| 1.03|
+——————————–+———————–+——-+———-+————-+———————————————————–+—————-+——————-+————————-+
其中每个列含义如下：
ID：同步任务的唯一标识符 CreationTime：同步任务的创建的时间 State：同步任务的状态 PipeSource：同步数据流的来源 PipeProcessor：同步数据流在传输过程中的处理逻辑 PipeSink：同步数据流的目的地 ExceptionMessage：显示同步任务的异常信息 RemainingEventCount（统计存在延迟）：剩余 event 数，当前数据同步任务中的所有 event 总数，包括数据和元数据同步的 event，以及系统和用户自定义的 event EstimatedRemainingSeconds（统计存在延迟）：剩余时间，基于当前 event 个数和 pipe 处速率，预估完成传输的剩余时间
5. 插件同步
了使得整体架构更加灵活以匹配不同的同步场景需求，支持在同步任务框架中进行插件组装。系统预置了一些常用插件可直接使用，同时也可以自定义 processor 插件 和 Sink 插件，并加载至 IoTDB 系统进行使用
查看系统中的插件（含自定义与内置插件）的语句
SHOW PIPEPLUGINS
结果
IoTDB> SHOW PIPEPLUGINS
+——————————+———-+————————————————————————————————–+—————————————————-+
| PluginName|PluginType| ClassName| PluginJar|
+——————————+———-+————————————————————————————————–+—————————————————-+
| DO-NOTHING-PROCESSOR| Builtin| org.apache.iotdb.commons.pipe.plugin.builtin.processor.donothing.DoNothingProcessor| |
| DO-NOTHING-SINK| Builtin| org.apache.iotdb.commons.pipe.plugin.builtin.connector.donothing.DoNothingConnector| |
| IOTDB-SOURCE| Builtin| org.apache.iotdb.commons.pipe.plugin.builtin.extractor.iotdb.IoTDBExtractor| |
| IOTDB-THRIFT-SINK| Builtin| org.apache.iotdb.commons.pipe.plugin.builtin.connector.iotdb.thrift.IoTDBThriftConnector| |
| IOTDB-THRIFT-SSL-SINK| Builtin| org.apache.iotdb.commons.pipe.plugin.builtin.connector.iotdb.thrift.IoTDBThriftSslConnector| |
+——————————+———-+————————————————————————————————–+—————————————————-+
预置插件详细介绍
总结
Apache IoTDB的Pipe数据同步机制通过三阶段架构实现了高效、可靠的数据复制能力。从基本配置到高级调优，从单机部署到云边协同，本文详细解析了数据同步的全流程操作。通过最佳实践帮助读者构建高可用、低延迟的时序数据同步链路。
随着物联网技术的不断发展，IoTDB将持续演进，提供更加智能、高效的数据管理能力。通过持续关注IoTDB的最新发展，企业可以始终保持在时序数据管理领域的技术领先地位，实现数字化转型的深度需求。
