SeaTunnel 从零到生产分布式数据集成完整实战教程【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel本文带你用 SeaTunnel 完成分布式数据集成30 分钟跑通订单库到数仓的同步任务再搭起带告警的流式处理管道最后完成集群部署与调优。适合刚接触数据同步的新手和准备上生产的工程师。一个真实需求订单库每秒都在涨报表却总是慢半拍先讲一个很常见的场景。业务订单库sharding_order.t_order每秒都在写入BI 团队每天要拉快照做报表但数据总是滞后一天更麻烦的是订单服务一旦报错运维只能靠人肉刷日志发现。SeaTunnel 要解决的就是这两件事把关系库里的数据持续搬到数仓Doris报表准实时可见把 Kafka 里的应用日志实时过滤出 ERROR直接推送到飞书告警群全程只用配置文件和启动命令不写一行 Java 代码。跟着本文走下来你会得到两个能直接提交生产作业的.conf文件以及一套集群部署与排障方法。环境搭建两条路径都带验证步骤先确认你的机器够不够项目最低要求生产建议JDK8 或 1111内存4 GB16 GB 起磁盘10 GB独立挂载日志增长较快网络能连通目标数据库/Kafka同机房低延迟确认命令如下JDK 版本不对后面所有步骤都会白做java -version # 预期输出openjdk version 11.0.x路径一二进制包推荐新手从 Apache 官方归档页下载最新稳定版二进制包解压即用。为什么选这条路无需编译环境5 分钟拿到完整可运行目录。tar -xzf apache-seatunnel-*-bin.tar.gz cd apache-seatunnel-* sh bin/install-plugin.shinstall-plugin.sh会读取 config/plugin_config 中声明的连接器把对应 jar 下载到connectors/目录。路径二源码编译需要定制或跟社区开发进阶路径适合要改插件、发自己版本的同学。为什么自己编译可以精确锁定依赖版本也能参与上游开发。git clone https://gitcode.com/GitHub_Trending/se/seatunnel cd seatunnel sh ./mvnw clean install -DskipTests -Dskip.spotlesstrue编译产物在seatunnel-dist模块的 target 下。到这里两条路径都收敛到同一个可运行目录结构。一键验证安装跑一个不需要任何外部依赖的最小任务官方模板在 config/v2.batch.config.template./bin/seatunnel.sh --config config/v2.batch.config.template -m local预期看到Job ... is submitted successfully及 Console 打印的假数据行任务跑完进程自动退出。看到这些输出说明引擎、配置解析、插件加载三件事都正常。更多本地部署细节见 本地快速开始文档。内部机制数据在 SeaTunnel 里是怎么流动的用数据流向的视角看一条记录从进入系统到落盘会经过这样几层协作作业提交层你执行seatunnel.sh客户端解析 HOCON 配置把作业提交给 Zeta 引擎也支持翻译到 Flink/Spark 运行Source 层连接器按parallelism拆成多个读取分片各自拉取数据Transform 层中间可做字段映射、SQL 过滤、加列等无状态加工数据在管道内流转Sink 层写入目标系统批任务结束时做一致性收尾Zeta 引擎底座Master 负责调度与状态CheckpointWorker 执行计算单元Slot节点间通过 Hazelcast 集群通信任务在引擎内部的具体流转可以看这张提交与执行流程图几个对你日常操作有用的结论每个连接器是独立插件用隔离的 ClassLoader 加载依赖冲突天然被隔离集群模式靠 Hazelcast 组网默认端口5801见 config/hazelcast.yaml状态快照Checkpoint周期由引擎统一控制失败后从最近快照恢复而不是从头重跑实战任务一订单库同步到 Doris 数仓目标把订单表准实时同步进 Doris供报表查询。下面四步走改配置、启动、看结果、查监控。1. 改配置新建jobs/order2dw.conf。要点是 Source 用增量时间戳、Sink 指向 Doris 的 FEenv { parallelism 4 job.mode BATCH checkpoint.interval 30000 } source { Jdbc { plugin_output order_extract base_url jdbc:mysql://10.20.30.40:3306/sharding_order driver com.mysql.cj.jdbc.Driver user sync_reader password sync_reader_2026 query SELECT order_id, user_id, amount, status, gmt_modified FROM t_order WHERE gmt_modified ? fetch_size 2048 } } sink { Doris { plugin_input order_extract doris { fenodes 10.20.30.51:8030 database dw table ods_order_rt username dw_writer password dw_writer_2026 } batch { max_rows 50000 max_bytes 52428800 } } }fetch_size控制单次拉取行数避免大结果集撑爆内存Doris 的batch参数决定攒批提交的阈值两者直接影响吞吐。2. 启动任务先确认两个连接器都已安装config/plugin_config里应有connector-jdbc和connector-doris然后提交sh bin/install-plugin.sh ./bin/seatunnel.sh --config jobs/order2dw.conf -m local3. 看结果任务结束时日志会打印Job End及每条 pipeline 的写出行数再去 Doris 侧核对数据是否落地mysql -h10.20.30.51 -P9030 -udw_writer -e SELECT COUNT(*) FROM dw.ods_order_rt;行数与源表增量区间一致即同步成功。4. 查监控引擎自带 Web UI默认http://localhost:8080开关见 config/seatunnel.yaml。在任务详情页可以看到每条 pipeline 的接收/写出字节数、记录数与 QPS用它确认 Source 与 Sink 速率是否匹配速率长期不匹配通常意味着某端是瓶颈。实战任务二Kafka 错误日志实时推送到飞书告警目标从 Kafka topicapp-error-events实时消费应用日志过滤出 ERROR 级事件格式化后推送飞书机器人。同样是四步。1. 改配置新建jobs/log-alert.conf。这里用Sql转换做过滤与字段裁剪比多个 transform 串联更简洁env { parallelism 4 job.mode STREAMING } source { Kafka { plugin_output raw_log bootstrap.servers kafka-01:9092,kafka-02:9092 topic app-error-events properties.group.id seatunnel-alert-gateway format json } } transform { Sql { plugin_input raw_log plugin_output filtered_log query SELECT trace_id, service, level, msg FROM raw_log WHERE level ERROR } FieldMapper { plugin_input filtered_log plugin_output alert_ready field_mapper { alert_msg concat(service, | , msg, | trace: , trace_id) } } } sink { Feishu { plugin_input alert_ready webhook https://open.feishu.cn/open-apis/bot/v2/hook/xxx-xxx secret SECxxx } }Sql转换基于 Calcite可以直接写标准 SQLFieldMapper负责拼出告警文案两个转换各自声明plugin_input/plugin_output管道关系清晰。2. 启动任务流式任务不会自行退出用后台方式提交nohup ./bin/seatunnel.sh --config jobs/log-alert.conf -m local /dev/null 21 3. 看结果向 Kafka 发一条测试消息几秒内飞书群应收到卡片消息kafka-console-producer.sh --bootstrap-server kafka-01:9092 --topic app-error-events # 输入{service:order-svc,level:ERROR,msg:inventory timeout,trace_id:t-1001}飞书收到 order-svc | inventory timeout | trace: t-1001 这样的消息链路就通了。4. 查监控回到 Web UI切到运行中作业列表重点看三个数消费 lag 是否持续增长、filtered_log输出速率、Feishu sink 的写出条数。lag 持续上涨说明下游推送限频跟不上此时应调大parallelism或合并推送。生产化部署从单机到分离集群单机模式只适合验证。上生产时官方建议采用分离集群模式Master 只管调度与 REST APIWorker 只跑任务两者互不干扰Master 重启不会把运行中作业全部冲掉理由与对比见 Zeta 部署文档。集群节点接入步骤每台节点部署同一份 SeaTunnel 目录config/hazelcast-master.yaml与config/hazelcast-worker.yaml分别指向 Master 与 Worker 组网配置修改 member-list 为本集群真实内网 IP并放行5801端口先起 Master再起 Worker# Master 节点10.30.1.11 sh bin/seatunnel-cluster.sh -m master -c config/hazelcast-master.yaml # Worker 节点10.30.1.12 / 10.30.1.13 依次执行 sh bin/seatunnel-cluster.sh -m worker -c config/hazelcast-worker.yamlmember-list必须与 Hazelcast 端口匹配默认 5801组网失败 90% 出在这一行。提交作业时把模式从 local 换成 cluster 即可配置不变./bin/seatunnel.sh --config jobs/order2dw.conf -m clusterMaster 节点监控界面如下可看到本集群 Master 的资源与负载情况Worker 节点面板展示各执行节点的 Slot 占用与任务分布多团队共享用资源隔离分车道多个团队共用一个集群时直接混跑会互相抢资源。SeaTunnel 用Tag标签 资源队列实现隔离给不同团队的任务打不同 tag再把队列与物理队列绑定每个队列可限制并行度上限与 Slot 上限。落地思路很简单为每个团队建独立队列并绑定唯一 tag关键团队的队列给更高的并行度上限队列打满时新任务排队等待而不是挤占别人资源详细规则见 资源隔离文档定时场景可以把 SeaTunnel 作业挂到 Azkaban 等调度器上由调度器负责触发与重跑排障手册三个高频问题的定位路径问题一Worker 起不来 / 集群只见 Master现象Worker 进程在但 Web UI 上看不到它作业卡住不调度。可能原因5801 端口不通member-list写的是外网 IP 或漏写本机防火墙拦截。排查命令# 在 Worker 节点上测 Master 端口连通性 curl -s telnet://10.30.1.11:5801 # 检查本机监听与 Hazelcast 配置 ss -lntp | grep 5801 grep -A4 member-list config/hazelcast-worker.yaml问题二报 ClassNotFoundException / 插件加载失败现象提交作业后报错类找不到日志指向连接器包。可能原因config/plugin_config里没声明该连接器导致install-plugin.sh没下载或 connectors 目录被误清理。排查命令grep -v ^# config/plugin_config | grep -v ^$ ls connectors/ | grep -E jdbc|doris sh bin/install-plugin.sh # 补齐缺失插件问题三跑一会儿就 OutOfMemoryError现象任务运行数十分钟后进程被杀logs/下出现 OOM 堆转储。可能原因-Xmx与物理内存不匹配单分片数据量过大fetch_size或 batch 攒批阈值太高元数据空间膨胀。排查命令# 查看默认 JVM 参数G1GC、Metaspace 上限 cat config/jvm_options # 定位 OOM 堆转储文件用 jhat/JFR 分析大对象 ls -lh /tmp/seatunnel/dump/zeta-server/ # 临时降压调小 fetch_size / batch.max_rows调大 parallelism处理原则先看堆转储确认是大结果集还是长生命周期对象再决定调参数还是改切分方式。调优清单参数、建议值与监控对接参数所在配置建议值适用场景-Xmxconfig/jvm_options物理内存的 50%~60%大吞吐同步任务parallelism作业env块对齐源端分片数Kafka 分区/Jdbc 分片提高管道并行度checkpoint.intervalconfig/seatunnel.yaml / 作业env30s~60s流任务容错与延迟权衡fetch_sizeJdbc source1000~5000库压力大或内存吃紧时调小batch.max_rows / max_bytesDoris 等 sink5 万行 / 50 MB 内攒批与实时性平衡backup-countconfig/seatunnel.yaml1集群副本数影响恢复速度对接 Prometheus / Grafana 的做法在作业env块里打开指标上报让 Prometheus 抓取各节点指标再用 Grafana 拼面板env { parallelism 4 metrics { reporter [prometheus] prometheus.port 5005 } }重点盯四个图各 pipeline 吞吐曲线、Kafka 消费 lag、JVM 堆占用与 GC 时间、队列等待长度。吞吐曲线长期锯齿状优先查 checkpoint 间隔与下游限流lag 缓慢爬升则扩 Worker。一页速查复制即用# ---- 安装与验证 ---- sh bin/install-plugin.sh # 按 plugin_config 下载连接器 ./bin/seatunnel.sh --config job.conf -m local # 本地模式跑作业 ./bin/seatunnel.sh --config job.conf -m cluster # 提交到集群 # ---- 集群启停分离模式---- sh bin/seatunnel-cluster.sh -m master -c config/hazelcast-master.yaml sh bin/seatunnel-cluster.sh -m worker -c config/hazelcast-worker.yaml # ---- 高频排查 ---- ss -lntp | grep 5801 # Hazelcast 端口监听 grep -A4 member-list config/hazelcast-worker.yaml ls connectors/ # 确认插件已就位 ls -lh /tmp/seatunnel/dump/zeta-server/ # 找 OOM 堆转储 curl -s http://localhost:8080 # Web UI 健康检查常用入口中文文档、配置文件目录、连接器文档、Transform 文档。路线图与你的下一步项目近期的能力方向在代码仓库里已经能看到影子AI CLI用对话方式生成与校验作业配置AI CLI 文档Edge Agent把采集能力下沉到边缘节点edge-agent 模块持续扩充连接器CDC、对象存储与 HTTP 类连接器更新最频繁给你三个马上可以做的动作把本文任务一里的库名、表名换成你自己的一套明天早上就能跑第一批真实同步在测试环境搭两节点分离集群故意杀一次 Worker观察作业如何从 Checkpoint 恢复给生产集群接上 Prometheus按调优清单里的四个图建第一版 Grafana 面板跑通第一个真实任务只是开始。把监控看板搭起来之后SeaTunnel 的容错与弹性才真正为你所用——从这一篇开始数据搬运会变成一件有抓手的事。【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考