快速上手 - apache/seatunnel

环境准备与部署

在开始使用 SeaTunnel 之前，需要确保系统满足以下基本要求。

Java 环境要求

SeaTunnel 需要 Java 运行环境。根据官方文档，系统需要安装 Java 8 或 Java 11，理论上高于 Java 8 的版本也可以工作。安装完成后，需要正确设置 JAVA_HOME 环境变量（docs/en/getting-started/locally/deployment.md:10-28）。

下载 SeaTunnel 二进制包

可以通过以下两种方式获取 SeaTunnel：

方式一：直接下载二进制包

访问 SeaTunnel 下载页面获取最新版本的 seatunnel-<version>-bin.tar.gz 文件，或通过命令行下载：

shell
1export version="3.0.0"
2wget "https://archive.apache.org/dist/seatunnel/${version}/apache-seatunnel-${version}-bin.tar.gz"
3tar -xzvf "apache-seatunnel-${version}-bin.tar.gz"

（docs/en/getting-started/locally/deployment.md:10-28）

方式二：从源码构建

从 GitHub 仓库或下载页面获取源码后执行构建：

shell
1cd seatunnel
2sh ./mvnw clean install -DskipTests -Dskip.spotless=true
3# 获取二进制包
4cp seatunnel-dist/target/apache-seatunnel-3.0.0-bin.tar.gz /The-Path-You-Want-To-Copy
5
6cd /The-Path-You-Want-To-Copy
7tar -xzvf "apache-seatunnel-${version}-bin.tar.gz"

（docs/en/getting-started/locally/deployment.md:61-78）

从源码构建时，二进制包会包含所有连接器插件和必要的依赖（如 MySQL 驱动），无需单独安装。

安装 Connector 插件

从 2.2.0-beta 版本开始，二进制包默认不再包含连接器依赖。首次使用时需要安装连接器：

bash
1sh bin/install-plugin.sh

（docs/en/getting-started/locally/deployment.md:30-60）

如需指定特定版本（以 3.0.0 为例）：

bash
1sh bin/install-plugin.sh 3.0.0

（docs/en/getting-started/locally/deployment.md:30-60）

通常不需要所有连接器插件，可以通过配置 config/plugin_config 指定所需插件。例如，要让示例应用正常工作，需要 connector-console 和 connector-fake 插件：

plugin_config
1--seatunnel-connectors--
2connector-fake
3connector-console
4--end--

（docs/en/getting-started/locally/deployment.md:30-60）

所有支持的连接器及其配置名称可在 ${SEATUNNEL_HOME}/connectors/plugins-mapping.properties 文件中找到。

使用 SeaTunnel Engine 快速上手

SeaTunnel Engine（代号 Zeta）是 SeaTunnel 内置的计算引擎，无需依赖外部引擎即可运行数据同步任务。这是最简单的快速上手方式。

配置作业文件

编辑 config/v2.batch.config.template 文件定义数据同步作业。以下配置使用 FakeSource 生成模拟数据，通过 FieldMapper 转换后输出到控制台：

hocon
1env {
2  parallelism = 1
3  job.mode = "BATCH"
4}
5
6source {
7  FakeSource {
8    plugin_output = "fake"
9    row.num = 16
10    schema = {
11      fields {
12        name = "string"
13        age = "int"
14      }
15    }
16  }
17}
18
19transform {
20  FieldMapper {
21    plugin_input = "fake"
22    plugin_output = "fake1"
23    field_mapper = {
24      age = age
25      name = new_name
26    }
27  }
28}
29
30sink {
31  Console {
32    plugin_input = "fake1"
33  }
34}

（docs/en/getting-started/locally/quick-start-seatunnel-engine.md:11-52）

本地模式运行

使用以下命令启动作业：

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

（docs/en/getting-started/locally/quick-start-seatunnel-engine.md:56-95）

注意：从 2.3.1 版本开始，seatunnel.sh 中的 -e 参数已弃用，应使用 -m 参数。

验证输出

作业运行成功后，控制台将输出类似以下日志：

shell
12022-12-19 11:01:45,417 INFO  org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - output rowType: name<STRING>, age<INT>
22022-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
32022-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
4...

（docs/en/getting-started/locally/quick-start-seatunnel-engine.md:56-95）

看到上述输出即表示作业运行成功。

使用 Flink 快速上手

如果已有 Flink 集群，可以利用 Flink 作为执行引擎运行 SeaTunnel 作业。

Flink 环境配置

首先下载 Flink（要求版本 >= 1.12.0），可参考 Flink 官方文档。然后在 ${SEATUNNEL_HOME}/config/seatunnel-env.sh 中设置 FLINK_HOME 环境变量指向 Flink 安装目录（docs/en/getting-started/locally/quick-start-flink.md:11-16）。

配置作业文件

编辑 config/v2.streaming.conf.template 文件，配置内容与 SeaTunnel Engine 示例相同（docs/en/getting-started/locally/quick-start-flink.md:17-60）。

启动命令

根据 Flink 版本选择对应的启动脚本：

Flink 1.12.x - 1.14.x 版本：

shell
1cd "apache-seatunnel-${version}"
2./bin/start-seatunnel-flink-13-connector-v2.sh --config ./config/v2.streaming.conf.template

Flink 1.15.x - 1.18.x 版本：

shell
1cd "apache-seatunnel-${version}"
2./bin/start-seatunnel-flink-15-connector-v2.sh --config ./config/v2.streaming.conf.template

（docs/en/getting-started/locally/quick-start-flink.md:62-104）

验证输出

成功运行后，控制台输出示例：

shell
1fields : name, age
2types : STRING, INT
3row=1 : elWaB, 1984352560
4row=2 : uAtnp, 762961563
5row=3 : TQEIB, 2042675010
6...

（docs/en/getting-started/locally/quick-start-flink.md:62-104）

使用 Spark 快速上手

如果已有 Spark 集群，可以使用 Spark 作为执行引擎。

Spark 环境配置

下载 Spark（要求版本 >= 2.4.0），可参考 Spark 官方文档。在 ${SEATUNNEL_HOME}/config/seatunnel-env.sh 中设置 SPARK_HOME 环境变量指向 Spark 安装目录（docs/en/getting-started/locally/quick-start-spark.md:11-16）。

配置作业文件

编辑 config/seatunnel.streaming.conf.template 文件，配置内容与前面示例类似（docs/en/getting-started/locally/quick-start-spark.md:17-60）。

启动命令

shell
1cd "apache-seatunnel-${version}"
2./bin/start-seatunnel-spark-connector-v2.sh --config ./config/seatunnel.streaming.conf.template

（docs/en/getting-started/locally/quick-start-spark.md:63-118）

验证输出

成功运行后，控制台将显示类似 Flink 模式的输出结果（docs/en/getting-started/locally/quick-start-spark.md:63-118）。

进阶示例MySQL 到 Doris

以下示例演示如何将 MySQL 数据同步到 Doris 数据仓库。

下载连接器

首先在 ${SEATUNNEL_HOME}/config/plugin_config 文件中添加所需连接器：

bash
1--seatunnel-connectors--
2connector-jdbc
3connector-doris
4--end--

然后执行安装命令：

bash
1sh bin/install-plugin.sh

（docs/en/getting-started/locally/quick-start-seatunnel-engine.md:97-118）

确保 connector-jdbc 和 connector-doris 已存在于 ${SEATUNNEL_HOME}/connectors/ 目录中。

放置 MySQL 驱动

下载 MySQL JDBC 驱动 JAR 包并放置到 ${SEATUNNEL_HOME}/lib/ 目录（docs/en/getting-started/locally/quick-start-seatunnel-engine.md:97-118）。

配置作业文件

创建 seatunnel/job/st.conf 文件：

bash
1cd seatunnel/job/
2vim st.conf

配置内容如下：

hocon
1env {
2  parallelism = 2
3  job.mode = "BATCH"
4}
5
6source {
7    Jdbc {
8        url = "jdbc:mysql://localhost:3306/test"
9        driver = "com.mysql.cj.jdbc.Driver"
10        connection_check_timeout_sec = 100
11        user = "user"
12        password = "pwd"
13        table_path = "test.table_name"
14        query = "select  * from test.table_name"
15    }
16}
17
18sink {
19   Doris {
20          fenodes = "doris_ip:8030"
21          username = "user"
22          password = "pwd"
23          database = "test_db"
24          table = "table_name"
25          sink.enable-2pc = "true"
26          sink.label-prefix = "test-cdc"
27          doris.config = {
28            format = "json"
29            read_json_by_line="true"
30          }
31      }
32}

（docs/en/getting-started/locally/quick-start-seatunnel-engine.md:120-187）

运行作业

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

（docs/en/getting-started/locally/quick-start-seatunnel-engine.md:120-187）

验证输出

作业成功完成后，控制台将显示统计信息：

shell
1***********************************************
2           Job Statistic Information
3***********************************************
4Start Time                : 2024-08-13 10:21:49
5End Time                  : 2024-08-13 10:21:53
6Total Time(s)             :                   4
7Total Read Count          :                1000
8Total Write Count         :                1000
9Total Failed Count        :                   0
10***********************************************

（docs/en/getting-started/locally/quick-start-seatunnel-engine.md:120-187）

运行模式对比

不同运行模式的适用场景如下表所示：

运行模式	依赖组件	适用场景	启动命令示例
SeaTunnel Engine (本地)	无	开发测试、独立部署、无外部集群环境	`./bin/seatunnel.sh --config xxx -m local`
SeaTunnel Engine (集群)	SeaTunnel Engine 集群	生产环境、高可用需求	需要确认（未在提供的源文件中找到详细说明）
Flink	Flink 集群	已有 Flink 基础设施、需要 Flink 生态集成	`./bin/start-seatunnel-flink-*.sh --config xxx`
Spark	Spark 集群	已有 Spark 基础设施、需要 Spark 生态集成	`./bin/start-seatunnel-spark-*.sh --config xxx`