企业官网建设流程全解析

Apache SeaTunnel Engine 本地快速上手：从单机验证 FakeSource 管线到 MySQL 批量入 Doris

【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel

本文是 Apache SeaTunnel 内置引擎（SeaTunnel Engine / Zeta）的本地快速启动指南，覆盖两条路径：先用-m local在单机上验证安装、插件与作业配置，再在需要多节点执行时平滑迁移到集群部署。读完本文你可以独立完成从部署 SeaTunnel、安装连接器插件、编写source-transform-sink三段式 HOCON 作业配置、运行批任务并解读输出日志，到扩展为 MySQL → Doris 真实批量同步的完整过程。

先理解两条路径

SeaTunnel Engine 既可以用于单机快速试用，也可以组成多节点集群运行。本文页面围绕这两条路径组织，请根据你的目标选择：

建议：当你想在单机上验证配置与作业管线时，使用本文的本地模式；当你需要多节点执行、资源隔离或更接近预发/生产的环境时，再使用集群部署指南。

Part 1：单机快速启动（本地模式）

本路径用于在单台机器上验证安装、连接器与作业配置。下面所有命令都以-m local方式启动 SeaTunnel Engine。

开始前的准备工作

如果你是第一次接触 SeaTunnel，建议按顺序先阅读以下文档：

• Getting Started 总览
• Deployment 部署说明
• Job 配置指南

本文的示例管线使用了FakeSource（模拟数据源）、FieldMapper（字段重命名转换）和Console（控制台输出 Sink）三个插件。运行示例前，请确保所需插件已安装。

在${SEATUNNEL_HOME}/config/plugin_config中声明需要的连接器插件名，然后执行安装脚本：

--seatunnel-connectors-- connector-fake connector-console --end--

sh bin/install-plugin.sh

在 Windows 上使用批处理脚本：

bin\install-plugin.cmd

补充说明：仓库根目录的 config/plugin_config 是当前发行版中插件声明文件的真实样例，其中区块标记为--connectors-v2--（自 2.3.x 起）而非早期文档中的--seatunnel-connectors--，你只需按所用发行版文档对应的标记格式填写插件名即可。所有受支持的连接器及其在plugin_config中的对应名称，可在${SEATUNNEL_HOME}/connectors/plugins-mapping.properties中查到。

第 1 步：部署 SeaTunnel 与连接器

开始之前，请确保已按 Deployment 部署说明 下载并部署好 SeaTunnel 发行包：

• 需要安装 Java 8 或 11（理论上高于 Java 8 的版本也可行）并设置JAVA_HOME；
• 下载apache-seatunnel-<version>-bin.tar.gz二进制包并解压；
• 从 2.2.0-beta 起二进制包默认不再携带连接器依赖，首次使用必须先运行sh bin/install-plugin.sh（或 Windows 下的bin\install-plugin.cmd）安装连接器，也可以从 Apache Maven 仓库手动下载连接器 JAR 放入connectors/目录（2.3.5 之前的版本放在connectors/seatunnel目录）。

如果你已经安装了全部连接器，可以保留现有环境；如果希望第一次运行尽量精简，上文声明的connector-fake与connector-console两个插件足以支撑本页示例作业。

第 2 步：编写作业配置文件定义作业

编辑config/v2.batch.config.template，它决定了 SeaTunnel 启动后数据输入、处理与输出的方式和逻辑。仓库根目录的 config/v2.batch.config.template 是随发行版提供的同款模板文件。以下是与上述示例应用一致的配置文件内容：

env { parallelism = 1 job.mode = "BATCH" } source { FakeSource { plugin_output = "fake" row.num = 16 schema = { fields { name = "string" age = "int" } } } } transform { FieldMapper { plugin_input = "fake" plugin_output = "fake1" field_mapper = { age = age name = new_name } } } sink { Console { plugin_input = "fake1" } }

这份配置的四个区块职责如下：

• env：环境级配置。parallelism = 1指定作业并行度为 1；job.mode = "BATCH"声明批模式。
• source：FakeSource模拟生成数据，通过plugin_output = "fake"将输出表命名为fake；row.num = 16表示生成 16 条数据；schema.fields声明了name（string 类型）与age（int 类型）两个字段。
• transform：FieldMapper读取输入表fake（plugin_input），按field_mapper映射规则处理：age = age表示字段保持原名，name = new_name表示把name重命名为new_name，处理结果输出到表fake1（plugin_output）。
• sink：Console消费输入表fake1（plugin_input），把每一行数据打印到控制台。

关于配置体系的更多信息，可参考 配置基础概念。

结合源码可以进一步理解各配置项的底层语义：

• row.num的含义是每个并行度生成的数据条数，定义于 FakeSourceOptions.java，默认值为 5。示例中parallelism = 1、row.num = 16，因此共输出 16 行数据。
• FakeSource 是“有界还是无界”取决于作业模式：在 FakeSource.java 中可以看到，BATCH模式返回Boundedness.BOUNDED（有界），流模式则返回UNBOUNDED（无界）。
• Console Sink 支持log.print.data（是否打印数据，默认true）与log.print.delay.ms（每条数据打印间隔毫秒数，默认 0），定义于 ConsoleSinkOptions.java。
• FakeSource 还支持split.num、split.read-interval、string.length、rows（逐行指定模板数据）、各类型的*.min/*.max/*.template/*.fake.mode等大量参数，可用于构造更贴近真实业务的测试数据，详见 FakeSourceOptions.java。
• FieldMapper转换插件位于seatunnel-transforms-v2模块的 fieldmapper 包 下，除字段重命名外还支持按映射删除字段等能力。

第 3 步：运行 SeaTunnel 应用

使用以下命令启动应用：

:::tip

自 2.3.1 版本起，seatunnel.sh中的-e参数已被弃用，请改用-m。

:::

cd "apache-seatunnel-${version}" ./bin/seatunnel.sh --config ./config/v2.batch.config.template -m local

在 Windows 上，从 SeaTunnel 目录运行对应的批处理入口：

cd apache-seatunnel-3.0.0 bin\seatunnel.cmd --config config\v2.batch.config.template -m local

观察输出：命令运行后，控制台会打印运行日志，这是判断命令是否执行成功的重要信号。

从源码层面看，-m/--master参数由 ClientCommandArgs.java 定义，支持local与cluster两种取值，默认值为cluster；其中-e与--deploy-mode已在 2.3.1 起标记为 deprecated。此外，本地模式下 SeaTunnel 引擎会以MASTER_AND_WORKER角色嵌入进程运行（参见 ClientExecuteCommand.java 中关于 local mode 的处理逻辑），并在默认情况下暴露 REST/UI 端点，便于你查看作业状态。

SeaTunnel 控制台会打印类似如下的日志：

2022-12-19 11:01:45,417 INFO org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - output rowType: new_name<STRING>, age<INT> 2022-12-19 11:01:46,489 INFO org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=1: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: CpiOd, 8520946 2022-12-19 11:01:46,490 INFO org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=2: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: eQqTs, 1256802974 2022-12-19 11:01:46,490 INFO org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=3: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: UsRgO, 2053193072 2022-12-19 11:01:46,490 INFO org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=4: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: jDQJj, 1993016602 2022-12-19 11:01:46,490 INFO org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=5: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: rqdKp, 1392682764 2022-12-19 11:01:46,490 INFO org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=6: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: wCoWN, 986999925 2022-12-19 11:01:46,490 INFO org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=7: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: qomTU, 72775247 2022-12-19 11:01:46,490 INFO org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=8: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: jcqXR, 1074529204 2022-12-19 11:01:46,490 INFO org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=9: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: AkWIO, 1961723427 2022-12-19 11:01:46,490 INFO org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=10: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: hBoib, 929089763 2022-12-19 11:01:46,490 INFO org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=11: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: GSvzm, 827085798 2022-12-19 11:01:46,490 INFO org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=12: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: NNAYI, 94307133 2022-12-19 11:01:46,490 INFO org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=13: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: EexFl, 1823689599 2022-12-19 11:01:46,490 INFO org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=14: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: CBXUb, 869582787 2022-12-19 11:01:46,490 INFO org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=15: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: Wbxtm, 1469371353 2022-12-19 11:01:46,491 INFO org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=16: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: mIJDt, 995616438

解读这些日志可以发现两个关键点：

1. 首行日志打印了经过转换后的输出行类型output rowType: new_name<STRING>, age<INT>——name字段已被FieldMapper重命名为new_name，证明transform环节生效；
2. 后续每行是ConsoleSinkWriter（源码）逐条输出的SeaTunnelRow，每条数据包含INSERT行类型与两个字段值，rowIndex从 1 到 16，与row.num = 16完全对应。

扩展示例：MySQL 到 Doris 的批模式同步

本地模式验证通过后，把示例中的模拟源与打印 Sink 替换为真实连接器即可投入实际同步。下面演示最常见的「MySQL 批量读入、Doris 批量写出」场景。

第 1 步：下载连接器

首先，将连接器名称添加到${SEATUNNEL_HOME}/config/plugin_config文件中，然后执行命令安装连接器（当然，你也可以从 Apache Maven 仓库手动下载连接器，放入connectors/目录）。最后，确保connector-jdbc和connector-doris两个连接器都位于${SEATUNNEL_HOME}/connectors/目录下。

# 配置连接器名称。 --seatunnel-connectors-- connector-jdbc connector-doris --end--

# 安装连接器。 sh bin/install-plugin.sh

第 2 步：放置 MySQL 驱动

你需要下载 MySQL 的 JDBC 驱动 JAR 包，并把它放到${SEATUNNEL_HOME}/lib/目录下，以便 JDBC 连接器能够加载 MySQL 驱动。

第 3 步：添加作业配置文件定义作业

cd seatunnel/job/ vim st.conf

env { parallelism = 2 job.mode = "BATCH" } source { Jdbc { url = "jdbc:mysql://localhost:3306/test" driver = "com.mysql.cj.jdbc.Driver" connection_check_timeout_sec = 100 user = "user" password = "pwd" table_path = "test.table_name" query = "select * from test.table_name" } } sink { Doris { fenodes = "doris_ip:8030" username = "user" password = "pwd" database = "test_db" table = "table_name" sink.enable-2pc = "true" sink.label-prefix = "test-cdc" doris.config = { format = "json" read_json_by_line="true" } } }

配置要点：

• env.parallelism = 2：批任务以并行度 2 执行，相比示例作业提升吞吐。
• JdbcSource：通过url指定 MySQL 连接串、driver指定驱动类com.mysql.cj.jdbc.Driver、connection_check_timeout_sec设置连接检查超时（秒）、user/password设置账号密码、table_path指定库表、query指定读取 SQL。
• DorisSink：fenodes为 Doris FE 地址（ip:port），username/password为账号密码，database/table为目标库表；sink.enable-2pc = "true"开启两阶段提交保证精确一次语义，sink.label-prefix设置导入 label 前缀；doris.config内通过format = "json"与read_json_by_line = "true"指定 JSON 逐行写入格式。

关于配置的更多信息，请参考 配置基础概念。

第 4 步：运行 SeaTunnel 应用

使用以下命令启动应用：

cd seatunnel/ ./bin/seatunnel.sh --config ./job/st.conf -m local

检查输出：命令运行后，可以在控制台看到输出信息，可以将其视为命令成功或失败的指示。SeaTunnel 控制台会打印类似下面的统计信息：

*********************************************** Job Statistic Information *********************************************** Start Time : 2024-08-13 10:21:49 End Time : 2024-08-13 10:21:53 Total Time(s) : 4 Total Read Count : 1000 Total Write Count : 1000 Total Failed Count : 0 ***********************************************

这份「Job Statistic Information」是批作业最直接的验收证据：Total Read Count与Total Write Count均为 1000，Total Failed Count为 0，说明 MySQL 中的 1000 行数据在 4 秒内全部写入 Doris，且无失败记录。

:::tip

如果希望优化作业，可参考连接器文档：Source-MySQL 与 Sink-Doris。

:::

Part 2：集群部署

如果你已经在本地验证了作业，并希望跨多个节点运行 SeaTunnel Engine，请继续阅读 SeaTunnel Engine(Zeta) 部署指南。

该部署指南覆盖以下内容：

• 本地模式、混合集群模式与分离集群模式的部署场景；
• 混合集群模式与分离集群模式的具体部署步骤；
• 如何选择合适的部署模式。

建议：当你想在单机上验证配置与作业管线时，使用本页的本地模式；当你需要多节点执行、资源隔离或更接近预发/生产的环境时，再使用部署指南。

延伸阅读

• 想要一条贯穿部署、快速启动与配置的引导式阅读路径，从 Getting Started 总览 开始。
• 准备把示例中的 Source 与 Sink 替换为真实连接器时，参考 Job 配置指南。
• 希望继续走一遍经过验证的「源到目标」演练，可以接着看 MySQL CDC 到 Kafka；其他管线形态见 MySQL CDC 到 Doris、JDBC 到 S3、Kafka 到 Iceberg、Http 到 JDBC、File 到 StarRocks 或 多表 CDC。
• 开始编写你自己的配置文件：选择想用的 连接器，并根据连接器文档配置参数。
• 想要部署多节点 SeaTunnel Engine 集群，继续阅读 SeaTunnel Engine(Zeta) 部署指南。
• 想进一步了解 SeaTunnel Engine 本身，可阅读 SeaTunnel Engine(Zeta) 介绍。

【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel