用 SeaTunnel 同步 MySQL 到 Doris：全量增量 + SQL 过滤
Apache SeaTunnel 能够实现 MySQL 到 Doris 的全量和增量数据同步，同时也支持 SQL 级别的数据过滤。以下是具体实现方式及功能特点：
全量与增量同步支持
1. 全量同步
实现方式：通过 SeaTunnel 的批处理模式（job.mode = "BATCH"），将 MySQL 的历史数据一次性导入 Doris。支持分片读取（如按主键分片）以提升效率，并可通过配置参数优化并行度、批量写入大小等。
source {Jdbc {query = "SELECT * FROM orders" -- 全量数据读取partition_column = "id" -- 分片字段split.size = 5000 -- 每分片读取行数}
}
2. 增量同步
- 基于时间戳字段：通过
WHERE update_time >= '${last_update_time}'动态参数筛选增量数据，需外部系统记录时间点并触发定期任务。 - 基于 CDC（变更数据捕获）：使用 MySQL CDC 连接器实时捕获 Binlog 变更（如插入、更新、删除），同步至 Doris。支持精确一次语义（Exactly-Once）和流式处理。
source {MySQL-CDC {startup.mode = "latest" -- 从最新位点开始同步table-names = ["db.table"]}
}
SQL 级数据过滤支持
1. 数据源层过滤
- 在源端 SQL 中定义过滤条件：通过 source 模块的 query 参数直接筛选数据，例如仅同步特定状态或时间范围的数据。
source {Jdbc {query = """SELECT * FROM orders WHERE status = 1 AND create_time > '2025-01-01'"""}
}
- 转换层过滤
Transform 模块的 SQL 转换：在 transform 阶段通过自定义 SQL 对数据进行清洗、过滤或字段映射。
transform {Sql {query = "SELECT id, name FROM source WHERE amount > 1000" -- 过滤金额小于1000的数据}
}
关键配置与注意事项
- Doris Sink 参数
需配置 Doris FE 节点地址、批量写入参数（batch_size）、数据合并策略（merge_type）等。
sink {Doris {fenodes = "doris_fe:8030"batch_size = 10000stream_load_properties = { "merge_type" = "MERGE" }}
}
- CDC 同步依赖条件
MySQL 需开启 Binlog 并配置ROW模式，用户需具备SELECT, REPLICATION SLAVE权限。
- 性能调优
全量同步建议分片读取避免单节点压力；增量同步可调整 Flink 或 Zeta 引擎的并行度以提升吞吐量。
与其他工具的对比
- SeaTunnel 优势：支持批流一体、SQL 级灵活性、精确一次语义，适合复杂场景；相比之下，DataX 仅支持离线批量，Sqoop 局限于 Hadoop 生态 123。
- 与 CloudCanal 对比：虽然 CloudCanal 提供可视化界面，但 SeaTunnel 的插件化架构和开源特性更适合自定义开发。
原文链接：https://blog.csdn.net/a772304419/article/details/146341445