![SeaTunnel 新手]()
欢迎来到 Apache SeaTunnel 的世界！这份文档旨在帮助新手快速了解 SeaTunnel 的核心功能、基本架构，并完成第一个数据同步任务。
- 什么是 Apache SeaTunnel？
Apache SeaTunnel 是一个非常易于使用、高性能、支持实时流式和离线批处理的海量数据集成平台。它的目标是解决常见的数据集成问题，如数据源多样性、同步场景复杂性以及资源消耗高的问题。
核心特性
- 丰富的数据源支持：支持 100+ 种 Connector，涵盖主流数据库、云存储、SaaS 服务等。
- 批流一体：同一套 Connector 代码同时支持批处理（离线）和流处理（实时）。
- 高性能：支持多引擎（Zeta, Flink, Spark），提供高吞吐、低延迟的数据同步能力。
- 简单易用：通过简单的配置文件（Config）即可定义复杂的数据同步任务。
- 架构与环境
2.1 架构图
SeaTunnel 采用了解耦的设计架构，Source、Transform、Sink 插件与具体的执行引擎（Engine）是分离的。
![ST architecture]()
2.2 操作系统支持
SeaTunnel 基于 Java 开发，理论上支持所有安装了 JDK 的操作系统。
| 操作系统 | 适用场景 | 说明 | | :--- | :--- | :--- | | Linux (CentOS, Ubuntu, etc.) | 生产环境 (推荐) | 稳定性高，适合长期运行服务。 | | macOS | 开发/测试 | 适合开发者本地调试和编写 Config。 |
2.3 环境准备
在开始安装 SeaTunnel 之前，请确保你的环境满足以下要求：
- JDK 版本 ：必须安装 Java 8 或 Java 11 。
- 可以通过命令
java -version 检查。
- 确保设置了
JAVA_HOME 环境变量。
- 核心组件深度解析
在使用 SeaTunnel 之前，深入理解其核心组件的内部机制有助于你更好地调优和排查问题。
3.1 Source (数据源)
Source 负责从外部系统读取数据，并将其转换为 SeaTunnel 内部的行格式（SeaTunnelRow）。
- Enumerator (枚举器) ：运行在 Master 节点（Coordinator）。负责发现数据分片（Splits）。例如，在 JDBC Source 中，Enumerator 会根据
partition_column 的最大值和最小值计算出多个查询范围（Splits）。
- Reader (读取器)：运行在 Worker 节点。负责接收 Enumerator 分配的 Splits，并真正执行读取操作。多个 Reader 并行工作，极大提高了读取效率。
- Checkpoint 支持：对于流式作业，Source 还需要支持状态保存（如 Kafka 的 Offset），以便在故障恢复时实现断点续传。
3.2 Transform (数据转换)
Transform 负责在数据从 Source 流向 Sink 的过程中对数据进行处理。
- 无状态转换 ：大多数 Transform（如
Sql, Filter, Replace）是无状态的，即处理当前行不需要依赖其他行的数据。
- Schema 变更：Transform 可以改变数据的 Schema（增加、删除、修改字段），下游 Sink 会感知到这种变化。
3.3 Sink (数据目标)
Sink 负责将 SeaTunnel 处理后的数据写入到外部系统。
- Writer (写入器)：运行在 Worker 节点。负责将数据写入目标系统。通常支持批量写入以提高吞吐量。
- Committer (提交器) ：运行在 Master 节点（可选）。对于支持事务的 Sink（如文件系统、Iceberg），需要一个全局的 Committer 来在 Checkpoint 完成时统一提交事务（二阶段提交），从而实现 Exactly-Once（精确一次）语义。
3.4 执行流程
-
解析配置：SeaTunnel 解析配置文件，构建逻辑执行计划。
-
资源分配：Master 节点根据并行度申请资源。
-
任务分发：Enumerator 生成分片，分发给 Reader。
-
数据流转 ：Reader -> Transform -> Writer。
-
状态提交：周期性触发 Checkpoint，保存状态并提交事务。
-
支持的 Connector 及其优缺点分析
SeaTunnel 支持超过 100 种 Connector，以下是几类最常用的 Connector 及其特性分析：
4.1 关系型数据库 (JDBC)
支持列表: MySQL, PostgreSQL, Oracle, SQLServer, DB2, Teradata, Dameng(达梦), OceanBase, TiDB 等。
- 优点 ：
- 通用性强：只要有 JDBC 驱动即可连接几乎所有 SQL 数据库。
- 功能完善 ：支持列投影（只读部分列）、并行读取（基于
partition_column 切分）、Exactly-Once（取决于实现）。
- 自动建表：部分 Connector 支持在目标端自动创建表结构。
- 缺点 ：
- 性能瓶颈 ：受限于 JDBC 协议和单机驱动性能，超大规模数据读取可能需要精细调优（如
fetch_size）。
- 源库压力：如果并行度设置过高，可能打满源库连接池或 CPU。
4.2 消息队列
支持列表: Kafka, Pulsar, RocketMQ, Amazon DynamoDB Streams 等。
- 优点 ：
- 高吞吐：天生适合大规模流数据处理，支持削峰填谷。
- 格式丰富：支持 JSON, Avro, Protobuf, Canal-JSON, Debezium-JSON 等多种序列化格式。
- Exactly-Once：支持端到端的精确一次语义（依赖 Checkpoint 机制）。
- 缺点 ：
- 配置复杂：涉及 Offset 管理、序列化 Schema 配置、Consumer Group 管理等。
- 数据可见性：相比数据库，数据在 Topic 中不够直观，调试稍显麻烦。
4.3 变更数据捕获 (CDC)
支持列表: MySQL-CDC, PostgreSQL-CDC, Oracle-CDC, MongoDB-CDC, SQLServer-CDC, TiDB-CDC 等。
- 优点 ：
- 实时性：毫秒级捕获数据库增删改操作。
- 无锁读取：SeaTunnel 的 CDC 实现了无锁并行快照算法，极大降低了对源库的影响。
- 断点续传：支持从 Binlog/WAL 指定位置恢复。
- Schema Evolution：支持表结构变更同步（部分支持）。
- 缺点 ：
- 权限要求：通常需要较高的数据库权限（如 REPLICATION SLAVE）。
- 依赖日志：源库必须开启 Binlog（或 WAL），且保留时间需足够长。
4.4 文件系统 & 云存储
支持列表: LocalFile, HDFS, S3, OSS, GCS, FTP, SFTP 等。
- 优点 ：
- 海量存储：适合数据湖场景，成本低廉。
- 格式支持：原生支持 Parquet, ORC, Avro, JSON, CSV, Excel, Text 等。
- 压缩支持：支持 Snappy, Gzip, Lzo 等多种压缩算法。
- 缺点 ：
- 小文件问题：流式写入时，如果 Checkpoint 间隔太短，容易产生大量小文件（SeaTunnel 有文件合并功能但会增加复杂度）。
4.5 NoSQL & 其他
支持列表: Elasticsearch, Redis, MongoDB, Cassandra, HBase, InfluxDB, ClickHouse, Doris, StarRocks 等。
- 特点：针对各数据库特性进行了优化，例如 ClickHouse/StarRocks 支持 Stream Load 高速导入，Elasticsearch 支持批量写入。
- Transform 实战演练 (附带详细注释)
Transform 插件用于在 Source 和 Sink 之间处理数据。以下是几个常用 Transform 的配置示例。
5.1 Sql Transform (最推荐)
使用 SQL 语法对数据进行处理，支持重命名、计算、常量添加、过滤等。
transform {
Sql {
# 输入表名，必须与 Source 的 result_table_name 一致
plugin_input = "fake"
# 输出表名，供后续 Transform 或 Sink 使用
plugin_output = "fake_sql"
# SQL 查询语句
# 1. name as full_name: 字段重命名
# 2. age + 1: 数值计算
# 3. 'US' as country: 增加常量列
# 4. where age > 10: 数据过滤
query = "select name as full_name, age + 1 as next_year_age, 'US' as country from fake where age > 10"
}
}
5.2 Filter Transform
用于删除或保留指定字段（注意：不是过滤行，是过滤列/字段）。
transform {
Filter {
plugin_input = "fake"
plugin_output = "fake_filter"
# 仅保留 name 和 age 字段，其他字段会被丢弃
include_fields = ["name", "age"]
# 或者使用 exclude_fields 删除指定字段
# exclude_fields = ["card"]
}
}
5.3 Replace Transform
用于字符串替换，支持正则表达式。
transform {
Replace {
plugin_input = "fake"
plugin_output = "fake_replace"
# 需要替换的字段名
replace_field = "name"
# 匹配模式（旧字符串）
pattern = " "
# 替换后的字符串（新字符串）
replacement = "_"
# 是否使用正则表达式，这里设为 true，表示 pattern 是一个正则
is_regex = true
# 是否只替换第一个匹配项
replace_first = true
}
}
5.4 Split Transform
将一个字符串字段拆分为多个字段。
transform {
Split {
plugin_input = "fake"
plugin_output = "fake_split"
# 分隔符，这里使用空格
separator = " "
# 需要拆分的源字段
split_field = "name"
# 拆分后生成的新字段名列表
output_fields = ["first_name", "last_name"]
}
}
- 快速安装
对于新手，推荐直接下载编译好的二进制发行包进行体验。
步骤 1: 下载
前往 SeaTunnel 下载页面 下载最新版本的二进制包（例如 apache-seatunnel-2.3.x-bin.tar.gz）。
步骤 2: 解压
tar -xzvf apache-seatunnel-2.3.x-bin.tar.gz
cd apache-seatunnel-2.3.x
步骤 3: 安装 Connector 插件
SeaTunnel 的 Connector 是插件化的。首次使用需要下载插件：
sh bin/install-plugin.sh
注意：该命令会根据 config/plugin_config 文件中的配置，从 Maven 中央仓库下载常用插件（如 connector-fake, connector-console 等）。如果下载速度慢，请耐心等待或配置 Maven 镜像。
💡 技巧：配置 Maven 国内镜像加速下载
如果遇到下载速度极慢或超时失败的情况，建议配置 Maven 阿里云镜像。
- 找到或创建 Maven 配置文件：
~/.m2/settings.xml (Windows 下为 C:\Users\你的用户名\.m2\settings.xml)。
- 添加如下镜像配置：
<settings>
<mirrors>
<mirror>
<id>aliyunmaven</id>
<mirrorOf>*</mirrorOf>
<name>阿里云公共仓库</name>
<url>https://maven.aliyun.com/repository/public</url>
</mirror>
</mirrors>
</settings>
保存后再次运行 sh bin/install-plugin.sh 即可享受高速下载。
- 实战：第一个 SeaTunnel 任务
我们将创建一个简单的任务：生成一些随机数据（FakeSource），并将其打印到控制台（Console Sink）。
步骤 1: 创建配置文件
在 config 目录下创建一个名为 hello_world.conf 的文件，内容如下：
env {
# 并行度设置：决定了有多少个线程同时执行任务。
# 设置为 1 表示单线程执行，适合测试；生产环境可根据资源调大。
parallelism = 1
# 作业模式：
# BATCH (批处理)：一次性处理完数据后结束（如离线同步）。
# STREAMING (流处理)：持续运行，实时处理数据（如实时同步）。
job.mode = "BATCH"
}
source {
# FakeSource 是一个虚拟数据源，用于生成测试数据
FakeSource {
# result_table_name: 将此数据源产生的数据注册为一个临时表，表名为 "fake"
# 后续的 Transform 或 Sink 可以通过这个名字引用这份数据
result_table_name = "fake"
# row.num: 指定生成数据的总行数，这里生成 16 行数据
row.num = 16
# schema: 定义数据的结构（字段名和类型）
schema = {
fields {
name = "string" # 定义一个名为 name 的字符串字段
age = "int" # 定义一个名为 age 的整型字段
}
}
}
}
transform {
# Sql Transform: 使用 SQL 语句对数据进行处理
Sql {
# plugin_input: 指定输入数据来源，这里引用了 Source 中定义的 "fake" 表
plugin_input = "fake"
# plugin_output: 指定处理后的结果表名，命名为 "fake_transformed"
# 下游的 Sink 将使用这个名字来获取处理后的数据
plugin_output = "fake_transformed"
# query: 执行的 SQL 查询语句
# 这里演示了选择 name 和 age 字段，并新增一个常量字段 new_field
query = "select name, age, 'new_field_val' as new_field from fake"
}
}
sink {
# Console Sink: 将数据输出打印到控制台（标准输出）
Console {
# plugin_input: 指定要输出的数据来源，这里引用了 Transform 输出的 "fake_transformed" 表
plugin_input = "fake_transformed"
}
}
步骤 2: 运行任务
使用 SeaTunnel 自带的 Zeta 引擎运行该任务。
执行命令：
./bin/seatunnel.sh --config ./config/hello_world.conf -e local
命令详解：
./bin/seatunnel.sh: 启动脚本，默认使用 Zeta 引擎。--config (或 -c): 指定配置文件的路径。这里我们指定了刚才创建的 hello_world.conf。-e local (或 -m local): 指定执行模式。
local: 本地模式。SeaTunnel 会在当前进程中启动一个轻量级的 Zeta 引擎集群来运行任务，任务结束后集群关闭。适合开发和测试。cluster: 集群模式。任务会提交到已经运行的 SeaTunnel 集群中执行。适合生产环境。
步骤 3: 查看结果与日志分析
任务启动后，终端会输出大量日志。我们需要关注以下关键信息：
-
任务提交成功 ： 看到 Job execution started 表示配置文件解析通过，任务已提交给引擎。
-
数据处理过程 ： 由于我们使用的是 Console Sink，数据会直接打印在日志中。你应能看到类似如下的输出：
...
INFO ...ConsoleSinkWriter - subtaskIndex=0 rowIndex=1: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: CpiOd, 12345, new_field_val
INFO ...ConsoleSinkWriter - subtaskIndex=0 rowIndex=2: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: eQqTs, 67890, new_field_val
...
subtaskIndex: 并行任务的编号。rowIndex: 当前处理的行号。SeaTunnelRow: 打印出的具体数据内容。
-
任务结束 ： 最后看到 Job Execution Status: FINISHED 表示任务执行成功结束。
8. 常见问题排查 (Troubleshooting)
如果在运行过程中遇到报错，请参考以下常见问题进行排查：
🔴 问题 1: command not found: java 或 JAVA_HOME is not set
- 现象：运行脚本时直接报错，提示找不到 Java。
- 原因：环境未安装 Java 或未配置环境变量。
- 解决 ：
- 运行
java -version 确认 Java 8 或 11 已安装。
- 设置环境变量：
export JAVA_HOME=/path/to/your/java (建议写入 ~/.bashrc 或 ~/.zshrc)。
🔴 问题 2: Exception in thread "main" ... ClassNotFoundException
- 现象 ：报错提示找不到某个类，例如
ClassNotFoundException: org.apache.seatunnel.connectors.seatunnel.fake.source.FakeSourceFactory。
- 原因 ：Connector 插件未安装。默认包中只有引擎核心，没有包含具体的数据源插件。
- 解决 ：
- 确保你执行过
sh bin/install-plugin.sh。
- 检查
connectors/seatunnel 目录下是否有对应的 jar 包（例如 connector-fake-*.jar）。
🔴 问题 3: Config file not valid 或 HOCONSyntaxError
- 现象：提示配置文件格式错误。
- 原因 ：
hello_world.conf 中的括号 {} 不匹配，或者关键字拼写错误。
- 解决 ：仔细检查配置文件语法。SeaTunnel 使用 HOCON 格式，确保每一层级的
{ 和 } 都是成对出现的。
🔴 问题 4: 任务卡住不动
- 现象：日志停止更新，但任务没有结束。
- 原因：可能是资源不足（CPU/内存），或者在流模式（STREAMING）下这是正常现象（流任务是无休止运行的）。
- 解决 ：
- 如果是 BATCH 模式卡住，检查
plugin_config 里的内存设置。
- 检查是否在
env 中错误地设置了 job.mode = "STREAMING"。
- 进阶学习资源
Apache SeaTunnel 批流一体、生态丰富、部署轻便，入门有指南，实战有案例。即刻上手探索，加入开源社区，让数据流转更简单，为数据工程高效赋能！祝你学习愉快！
