1. 项目概述当数据同步不再让你加班到凌晨做数据平台的同学或多或少都经历过这样的场景老板周一早上丢过来一个需求——把业务库的表都同步到数仓做核心看板。你打开数据库一看十几套业务系统、上百张表、有全量有增量有的库还分了好几个环境。用 DataX 导全量还可以但增量 CDC 得另想办法直接用 Flink 写作业Sink 到 Hive 或者 StarRocks 的代码量不小公共集群压过来一跑性能还不一定好。这个项目标题 Apache SeaTunnel就是冲着这个痛点来的。Apache SeaTunnel 是一个开源的数据集成同步引擎上一代叫 SeaTunnel曾用名 Waterdrop进入 Apache 孵化器后改名 SeaTunnel。它在社区里的定位很简单把数据从一个地方批量或实时地挪到另一个地方不需要你从头写代码只要写一份 JSON 格式的配置文件声明 source、transform、sink 三段结构程序就会自动完成分布式读取、计算和写出。我最早关注 SeaTunnel 是在项目里要接几十张 MySQL 表到 ClickHouse当时团队的方案是 DataX 临时脚本增量还得靠人工触发。后来试了 SeaTunnel 的 JDBC 插件和 CDC 插件几分钟就能跑通一个同步作业而且它在流式同步、整库同步、多源写入这几个场景上的设计明显不是一个简单脚本能做到的。这篇内容适合正在做数据集成、数仓建设和实时同步的工程师也适合想快速搭一套数据同步链路但不想重复造轮子的同学参考。核心点我先把结论放这SeaTunnel 的三段式配置模型 内置同步引擎 插件生态是它跟纯代码方案和纯调度脚本拉开差距的关键。下面我会按设计思路、架构原理、实操步骤、故障排查四个维度拆一遍全程带的都是我自己跑过的配置和踩过的坑。2. 核心设计拆解为什么三段式配置能省掉 80% 的同步开发2.1 从用户视角理解 Source / Transform / SinkSeaTunnel 的任务配置只有一个 JSON 文件定义了三块source数据从哪来比如 MySQL 表、Kafka Topic、Hive 分区、Oracle 查询结果。transform数据要不要加工比如字段改名、类型转换、加常量、过滤脏数据。sink数据写到哪去比如 ClickHouse 表、StarRocks 表、HDFS、Elasticsearch。这三段是串行相邻的数据流从 source 一个个流到 sink中间可以有零个或多个 transform。配置结构长得像这样{ env: { parallelism: 2, job.mode: BATCH }, source: { plugin_name: Jdbc, url: jdbc:mysql://localhost:3306/source_db, user: root, password: 123456, query: SELECT id, name, created_at FROM user_table }, sink: { plugin_name: Clickhouse, host: localhost:8123, database: target_db, table: user_table, fields: [id, name, created_at] } }第一眼看到这个配置的人通常会问一个问题这跟 DataX 的 job.json 不是一样吗确实两者表达方式很像但 SeaTunnel 的核心差异不在配置长什么样而在运行时引擎。2.2 为什么 SeaTunnel 自带引擎而不是直接跑在 Flink 上老版本 SeaTunnelWaterdrop是跑到 Spark 上的通过 Spark Structured Streaming 来处理数据同步。但从 2.x 版本开始项目把底层引擎换成了自研的 SeaTunnel Engine这是一个基于微内核架构的分布式同步引擎它根据自己的数据移动场景做了大量优化而不是把通用的流批计算引擎整个拿过来用。自研引擎带来的好处非常直接部署轻不需要额外搭一套 Flink/Spark 集群只要一台机器解压即是运行环境。多任务管理简单自带 REST API 和 Web 页面可以在一个引擎实例上同时提交、停止、查看多个同步作业。资源粒度更细同步任务是并行度为单位的不用像 Flink 那样动不动申请 TM 槽位用起来省心。链路优化SeaTunnel Engine 在 Source/Sink 之间做了 receive 缓冲背压处理、checkpoint 容错都走了同步场景的专用路径比通用计算引擎的吞吐表现更稳定。当然SeaTunnel 也保留了 Flink 和 Spark 的 runner选 Flink runner 的时候任务就跑在 Flink 集群上。但对于大多数内部数据同步需求内置引擎就是最省事的选择这也是我推荐新手直接跑默认引擎的原因。2.3 插件生态设计不写代码也能接入新数据源SeaTunnel 把每种数据源封装成一个插件插件的加载机制是 SPI 包名约定。你只需要下载对应的 connector jar放到connectors/目录配置文件里把 plugin_name 写成对应名字它就能在这个任务里被加载。以官方 2.3.x 版本为准目前内置连接器覆盖了 JDBC 系的 MySQL、PostgreSQL、Oracle、SQL Server、达梦、OceanBase以及 ClickHouse、Doris、StarRocks、Elasticsearch、Kafka、Pulsar、Iceberg、Hudi、Hive、HDFS 等。社区里还有不少第三方插件基本能覆盖主流技术栈。这里想提醒一句不要贪多先掌握 Jdbc 和 CDC 这两个插件就够解决 70% 的同步问题了。剩下 30% 的场景比如对象存储、消息队列再到对应插件文档里查参数字典就行。3. 核心实操从零搭建一个 MySQL 到 ClickHouse 的同步任务3.1 环境准备与安装SeaTunnel 的部署门槛是我见过的同类工具里最低的。只要机器上有 JDK 8 或 11下载发行包解压就能跑。我习惯这样组织目录apache-seatunnel-2.3.8/ ├── bin/ │ ├── seatunnel.sh │ └── seatunnel-cluster.sh ├── config/ │ └── seatunnel.yaml ├── connectors/ │ └── connector-jdbc.jar ├── lib/ └── plugins/下载的时候要注意官方发布包只带了基础核心 jar常用连接器需要单独下载放进 connectors。别漏了这一步否则后面跑任务的时候会报 Plugin not found。配置config/seatunnel.yaml主要是设置引擎的 rest api 端口和内置 web 的监听地址单机模式基本不需要动默认就能跑。如果想让作业状态持久化到本地存储可以加一段seatunnel: engine: backup-count: 2 queue-type: blockingqueue print-execution-info-interval: 60 backup-time: 5 >{ env: { job.mode: BATCH, parallelism: 4, checkpoint.interval: 60000 }, source: { plugin_name: Jdbc, url: jdbc:mysql://127.0.0.1:3306/business, user: reader, password: read_only_pwd, query: SELECT id, name, mobile, email, status, create_time, update_time FROM user_info WHERE update_time 2025-01-01 00:00:00, result_table_name: user_source }, transform: [ { plugin_name: FieldMapper, source_column_names: [id, name, mobile, email, status, create_time], target_column_names: [user_id, user_name, phone, mail_addr, user_status, created_at] } ], sink: { plugin_name: Clickhouse, host: 127.0.0.1:8123, database: ods, table: ods_user_info, fields: [user_id, user_name, phone, mail_addr, user_status, created_at], clickhouse.config: { insert_distributed_sync: true, input_format_skip_unknown_fields: true }, save_mode: APPEND } }提交任务bin/seatunnel.sh --config jobs/mysql_to_ck_user.conf --deploy-mode client跑完之后可以在命令日志里看到同步行数、吞吐量和耗时统计。实测同步一张 20 万行的表在单机并行度 4 的情况下从读取到写 ClickHouse 大约花了 12 秒这个速度对于大多数内部同步需求已经够用。3.3 几个关键参数的选择逻辑上面的配置里有几个参数不是随便填的我解释一下为什么这么配parallelism决定并行读取和写入的并发度。这个值不是越大越好太大会给源库造成读取压力太小会浪费带宽。我通常先设成 2 到 4然后观察源库 CPU 和网络延迟再往上调。job.mode填BATCH就是跑一次退出填STREAMING就是持续监听配合 CDC 插件做增量同步。选错模式会直接影响行为不是只是概念早晚的问题。checkpoint.interval批式任务也会做 checkpoint用来做失败恢复。默认 60 秒如果你数据量小、单批次几秒就跑完可以调小到 10 秒好处是失败后重试不至于重读整张表。save_mode三种取值APPEND、OVERWRITE、ERROR_IF_EXISTS。批式全量同步一般用OVERWRITE不会因为历史数据残留导致重复记录增量同步用APPEND更安全如果目标表不存在OVERWRITE会自动建表。还有个小细节ClickHouse 的 sink 参数里我加了insert_distributed_sync这个参数非常关键它决定INSERT是否同步等待分布式表写入完成。如果不加这个参数你会在任务结束后立刻查数时发现数据量对不上其实是分布式表异步传播还没完成。真实生产环境里我因为这个参数排查了两次数据量不一致的问题加深了很多印象。3.4 Flink runner 和自带引擎的切换如果你的公司已经有一套成熟的 Flink 集群想复用 Flink 的监控和资源管理SeaTunnel 也支持把任务跑到 Flink 上。只需要把任务配置放到--flink模式下执行bin/start-seatunnel-flink-13-connector-v2.sh --config jobs/mysql_to_ck.conf但这里要注意同一个配置文件在自带引擎和 Flink runner 下行为可能有细微差别。主要体现在 sink 的提交语义、checkpoint 的触发方式还有 transform 插件里部分函数的时间精度。我建议优先选定一个运行时跑到底不要来回切。我的场景大多数不依赖 Flink 集群直接用自带引擎省了集群运维成本。4. 核心实践进阶CDC 增量、整库同步和 Transform 字段处理4.1 基于 MySQL CDC 的流式增量同步增量同步这件事早期团队的做法是轮询update_time字段每五分钟扫一次。这个方案的问题在于删除操作抓不到、时间字段不规范的表没法用、源库 binlog 被清理后数据断档。SeaTunnel 的 CDC 插件解决的就是这个问题它直接读取 MySQL binlog把 insert / update / delete 事件解析成流式数据再写到目标端。一个最小可用的 CDC 配置长这样{ env: { job.mode: STREAMING, checkpoint.interval: 10000, parallelism: 1 }, source: { plugin_name: MySQL-CDC, hostname: 127.0.0.1, port: 3306, username: cdc_user, password: cdc_password, database-name: business, table-name: user_info, startup.mode: initial }, sink: { plugin_name: StarRocks, nodeUrls: [127.0.0.1:8030], username: root, password: , database: ods, table: user_info, save_mode: APPEND, starrocks.config: { format: json, column_separator: , } } }这里有两个容易踩的坑第一startup.mode有三种取值initial表示先做一次全量快照再继续监听 binlogearliest表示从当前 binlog 位点开始读latest表示从最新位点开始读。如果目标表已经有历史数据你想做到“全量增量无缝衔接”要选initial千万别选latest否则从任务启动时间点之前的数据全部漏掉。第二MySQL 的 binlog 必须开启ROW模式而且同步账号要有REPLICATION SLAVE和REPLICATION CLIENT权限。我记得第一次跑 CDC 报了Access denied就是这两个权限没给不是 SeaTunnel 配置的问题是数据库侧权限不够。4.2 整库同步如何把几十张表一次性同步过去SeaTunnel 的整库同步方案可以方便地把一个数据库下所有表同步到目标端不需要为每张表写单独任务。你只需要配置table-list指定表名列表或者用正则匹配表名再为所有表配置相同的字段映射规则。官方推荐的做法是用dynamic_merge模式配合 JDBC source 的tableList{ source: { plugin_name: Jdbc, url: jdbc:mysql://127.0.0.1:3306/business, user: reader, password: read_only_pwd, query: [SELECT * FROM ${table_name}], tableList: [A, B, C], result_table_name: jdbc_table }, sink: { plugin_name: Clickhouse, host: 127.0.0.1:8123, database: ods, table: ${table_name}, save_mode: OVERWRITE } }这里${table_name}是 SeaTunnel 内置的模板变量它会在任务运行时自动替换成 source 表名从而动态决定 sink 的目标表名实现多表同步复用同一段配置。不过要提醒一句整库同步不等于无脑同步。生产环境里表与表之间的字段类型差异很大有的大字段表、有的带 JSON、有的还有非法字符如果不做预处理直接整库灌数据很容易在目标端报错或者产生大量空值。我的建议是先跑一遍元数据采集列出所有表的字段类型把那些需要转换的表单独写配置其余的表走整库同步模板减少任务数量。4.3 Transform 里的字段处理和清洗实践Transform 是三段式里最容易被忽略但也最好用的部分。数据同步不是简单搬运客户表里的手机号可能是 11 位明文目标系统要求脱敏后存储日志表里的状态字段可能是 1、0、2目标数仓要求转成枚举字符串。这些都可以在 SeaTunnel 的 transform 里直接完成不用先把数据落下来再开一个 Spark SQL 去洗。一个常用的 Filter Replace 案例{ transform: [ { plugin_name: Filter, condition: { status: 1 } }, { plugin_name: Replace, replace_rules: [ { source_field: mobile, target_field: mobile_encrypted, replace_type: REGEX, replace_regex: (\\d{3})\\d{4}(\\d{4}), replace_string: $1****$2 } ] } ] }这段配置先过滤掉状态不是 1 的数据再把手机号中间四位打码并生成一个新字段mobile_encrypted。如果你的需求是字段拆分把full_name拆成first_name和last_name或者字段拼接、时间格式转换Transform 里都有对应的插件不需要写 UDF。值得留个心眼的是Transform 的执行成本并不低每条数据都要经过一次计算。如果数据量特别大且同步链路本身不需要清洗尽量别加无意义的 Transform。我曾经在一个 5000 万行的同步任务里只加了一个字段重命名 Transform结果整体耗时比不加的时候慢了将近 20%后来把重命名改成在 Sink 端字段映射里做了速度立刻回来了。5. 常见问题与排查技巧实录5.1 连接器版本冲突和 NoClassDefFoundSeaTunnel 用 SPI 加载插件连接器 jar 和引擎核心 jar 在同一套 ClassLoader 下工作这就很容易出现 jar 冲突。尤其是当你在集群上同时引入了 Kafka 客户端、Hadoop 客户端、JDBC 驱动时稍不注意就会抛NoClassDefFoundError或者ClassNotFoundException。排查思路我一般按顺序来看异常堆栈里缺的类属于哪个依赖比如com.mysql.cj.jdbc.Driver属于 MySQL Connector/J。检查connectors/和lib/下是否重复放了这个驱动 jar如果有二选一。如果用了--deploy-mode cluster还要确认集群所有节点上的 jar 包是一致的只改了本地目录没同步到集群节点是常见事故原因。实在定位不了把bin/seatunnel.sh前面加JAVA_OPTS打开环境变量配置用-verbose:class 21 | grep xx去追加载来源。5.2 同步性能没达到预期的排查点有次在 K8s 上跑一个 MySQL 到 Kafka 的同步任务数据量不大但延迟高得离谱。我当时逐个排查了这几个因素Source 端query里如果带了 ORDER BY 或者 GROUP BY会阻止 JDBC 分页读取导致单线程读表。并行度CDC 任务并行度最好设为 1 或者和分片数一致设太高反而会重复读取数据。Sink 端ClickHouse/StarRocks 的 batch 大小和 flush 间隔要调。默认是 1000 条一批如果你写入目标和源之间网络延迟高可以调小批量、加多线程并发反而更稳定。背压Kafka sink 如果分区数少下游消费慢sink 端自然就会出现背压。加参数sink.properties.batch.size500和sink.properties.queue.size10000可以缓解一部分压力。5.3 数据不一致问题排查同步平台最怕的不是任务跑挂而是任务显示成功但数据不对。我遇到过一种情况MySQL 同步到 ClickHouse目标任务显示成功按主键查目标表却发现部分记录重复。排查后定位到原因是 ClickHouse 分布式表在 insert 时没有设置insert_distributed_sync导致同一批数据被分布式表异步发到了多个分片配合重试机制产生了重复记录。还有一次是 StarRocks 目标表写入时没配primary key模型导致主键重复的记录不是更新而是追加。解决方案是把目标表改成 unique key 或 primary key 模型SeaTunnel sink 端会执行 upsert 语义。如果你也在做同类同步以下几个点建议先固化到测试清单里空值和 NULL 是否会写入目标端行为一致吗主键冲突时是覆盖、追加还是报错源库删除的数据目标端有对应删除策略吗字段类型长度超限时是截断还是报错5.4 我用的几个排查命令和技巧SeaTunnel 自带的日志、REST API 和内置 Web UI 是排查问题的三条路径tail -f logs/seatunnel-engine.log通过bin/seatunnel.sh --help能看到任务提交命令。如果任务在启动阶段就失败优先看logs/seatunnel-engine.log里的ERROR行。Web UI 默认地址是http://服务器IP:8800页面里可以看到作业的状态、并行度、已处理数据量、吞吐量和错误日志摘要。我在调优性能时就是盯着 Web UI 里的Throughput指标反复调参的比每次看命令行日志直观很多。REST API 可以用来查作业列表和提交状态适合集成到自动化发布流程curl http://localhost:8800/hazelcast/rest/maps/seatunnel-engine-jobList注意8800 是需要占用一个端口的如果你的运行环境端口受限需要在config/seatunnel.yaml里改seatunnel.engine.http.port。6. 关于选型和落地的一点个人体会到底要不要用 SeaTunnel这个取决于你的团队规模和同步场景复杂度。如果你的团队只有两三个人要接的数据源不超过五个需要的只是定时跑批同步——这时候用脚本 调度平台也没什么问题引入一个引擎反而增加学习成本。判断依据是你的同步链路里“从 A 到 B”这种模式的比例有多少。如果占到一半以上SeaTunnel 值得投入如果每个需求都要写一堆自定义逻辑那用 Flink/Spark 自己做更灵活。如果已经决定要引入我建议从最小范围开始先搭一个单节点环境对接一个 MySQL 到 ClickHouse 的真实需求把全量、增量、字段映射三条链路都跑通再考虑上整库同步和 CDC。等团队用出经验再逐步铺开。一上来就想搞几十张表的大迁移遇到问题的时候定位成本会高很多。最后再分享一个我个人一直在用的小习惯所有 SeaTunnel 任务配置文件都纳入版本管理并且把目标端的建表 SQL 也一并提交到仓库里。这让我要重建环境时可以一条命令从仓库拉全量配置不用靠记忆还原。同步引擎本身在迭代但配置管理、数据校验和监控这套“人肉工程”反而才是长期稳定运行的关键。