一篇文章搞定数据同步工具SeaTunnel

文章目录

- 前言
- 第 1 章 Seatunnel 概述
- 第 2 章 Seatunnel 安装和使用
- 第 3 章 SeaTunnel 基本原理
- 第 4章应用案例
- 后记

前言

SeaTunnel安装包及代码：

链接: https://pan.baidu.com/s/1JvgAZpqoOPJ0ecfxUbLo4Q 提取码: pur8
–来自百度网盘超级会员v4的分享

第 1 章 Seatunnel 概述

1.1 SeaTunnel 是什么

SeaTunnel 是一个简单易用的数据集成框架，在企业中，由于开发时间或开发部门不通用，往往有多个异构的、运行在不同的软硬件平台上的信息系统同时运行。数据集成是把不同来源、格式、特点性质的数据在逻辑上或物理上有机地集中，从而为企业提供全面的数据共享。 SeaTunnel 支持海量数据的实时同步。它每天可以稳定高效地同步数百亿数据。并已用于近 100 家公司的生产。

SeaTunnel 的前身是 Waterdrop （中文名：水滴）自 2021 年 10 月 12 日更名为 SeaTunnel。 2021 年 12 月 9 日， SeaTunnel 正式通过 Apache 软件基金会的投票决议，以全票通过的优秀表现正式成为 Apache 孵化器项目。 2022 年 3 月 18 日社区正式发布了首个 Apache 版本 v2.1.0。

1.2 SeaTunnel 在做什么

本质上，SeaTunnel 不是对 Saprk 和 Flink 的内部修改，而是在 Spark 和 Flink 的基础上做了一层包装。它主要运用了控制反转的设计模式，这也是 SeaTunnel 实现的基本思想。

SeaTunnel 的日常使用，就是编辑配置文件。编辑好的配置文件由 SeaTunnel 转换为具体的 Spark 或 Flink 任务。如图所示。

1.3 SeaTunnel 的应用场景

SeaTunnel 适用于以下场景	SeaTunnel 的特点
⚫ 海量数据的同步⚫ 海量数据的集成⚫ 海量数据的 ETL⚫ 海量数据聚合⚫ 多源数据处理	⚫ 基于配置的低代码开发，易用性高，方便维护。⚫ 支持实时流式传输⚫ 离线多源数据分析⚫ 高性能、海量数据处理能力⚫ 模块化的插件架构，易于扩展⚫ 支持用 SQL 进行数据操作和数据聚合⚫ 支持 Spark structured streaming⚫ 支持 Spark 2.x

目前 SeaTunnel 的长板是他有丰富的连接器，又因为它以 Spark 和 Flink 为引擎。所以可以很好地进行分布式的海量数据同步。通常 SeaTunnel 会被用来做出仓入仓工具，或者被用来进行数据集成。比如，唯品会就选择用 SeaTunnel 来解决数据孤岛问题，让 ClcikHouse

集成到了企业中先前的数据系统之中。如图所示：

下图是 SeaTunnel 的工作流程：

1.5 SeaTunnel 目前的插件支持

1.5.1 Spark 连接器插件(Source)

Spark 连接器插件	数据库类型	Source	Sink
Batch	Fake	√
	ElasticSearch	√	√
	File	√	√
	Hive	√	√
	Hudi	√	√
	Jdbc	√	√
	MongoDB	√	√
	Neo4j	√
	Phoenix	√	√
	Redis	√	√
	Tidb	√	√
	Clickhouse		√
	Doris		√
	Email		√
	Hbase	√	√
	Kafka		√
	Console		√
	Kudu	√	√
	Redis	√	√
Stream	FakeStream	√
	KafkaStream	√
	SocketSTream`	√

1.5.2 Flink 连接器插件(Source)

数据库类型	Source	Sink
Druid	√	√
Fake	√
File	√	√
InfluxDb	√	√
Jdbc	√	√
Kafka	√	√
Socket	√
Console		√
Doris		√
ElasticSearch		√

1.5.3 Spark & Flink 转换插件

转换插件	Spark	Flink
Add
CheckSum
Convert
Date
Drop

Grok
Json	√
Kv
Lowercase
Remove
Rename
Repartition
Replace
Sample
Split	√	√
Sql	√	√
Table
Truncate
Uppercase
Uuid

这部分内容来官网，可以看出社区目前规划了大量的插件，但截至 V2.1.0 可用的

transform 插件的数量还是很少的。同学们有兴趣也可以在业余时间尝试参与开源贡献。

官方网址：https://seatunnel.apache.org/zh-CN/

第 2 章 Seatunnel 安装和使用

注意 v2.1.0 中有少量bug，要想一次性跑通所有示例程序，需使用我们自己编译的包，可以在资料包里获取。具体如何修改源码，可以参考文档第 5 章。

2.1 SeaTunnel 的环境依赖

截至 SeaTunnel V2.1.0。

SeaTunnel 支持 Spark 2.x（尚不支持 Spark 3.x）。支持 Flink 1.9.0 及其以上的版本。 Java 版本需要>=1.8

我们演示时使用的是 flink 版本是 1.13.0

2.2 SeaTunnel 的下载和安装

1）使用 wget 下载 SeaTunnel，使用-O 参数将文件命名为 seatunnel-2.1.0.tar.gz

2）解压下载好的 tar.gz 包

3）查看解压的目标路径，apache-seatunnel-incubating-2.1.0 的目录就是我们已经安装好的 seatunnel 。Incubating 的意思是孵化中。

2.3 SeaTunnel 的依赖环境配置

在 config/目录中有一个 seatunnel-env.sh 脚本。我们可以看一下里面的内容。

这个脚本中声明了 SPARK_HOME 和 FLINK_HOME 两个路径。默认情况下 seatunnel- env.sh 中的 SPARK_HOME 和 FLINK_HOME 就是系统环境变量中的 SPARK_HOME 和 FLINK_HOME。

在 shell 脚本中:- 的意思是如果:-前的内容为空，则替换为后面的。

例如，环境变量中没有 FLINK_HOME。那么 SeaTunnel 运行时会将 FLINK_HOME 设为/opt/flink。

如果你机器上的环境变量 SPARK_HOME 指向了 3.x 的一个版本。但是想用2.x 的 Spark 来试一下 SeaTunnel。这种情况下，如果你不想改环境变量，那就直接在 seatunnel-env.sh 中将 2.x 的路径赋值给 SPARK_HOME 即可。

比如：

2.4 示例 1: SeaTunnel 快速开始

我们先跑一个官方的flink 案例。来了解它的基本使用。

1）选择任意路径，创建一个文件。这里我们选择在 SeaTunnel 的 config 路径下创建一个

example01.conf

2）在文件中编辑如下内容

3）开启 flink 集群

4）开启一个 netcat 服务来发送数据

5）使用 SeaTunnel 来提交任务。

在 bin 目录下有以下内容

start-seatunnel-flink.sh 是用来提交flink 任务的。 start-seatunnel-spark.sh 是用来提交 Spark 任务的。这里我们用 flink 演示。所以使用 start-seatunnel-flink.sh。

用–config 参数指定我们的应用配置文件。

等待弹出 Job 已经提交的提示

6）在 netcat 上发送数据

7）在 Flink webUI 上查看输出结果。

8）小结

至此，我们已经跑完了一个官方案例。它以 Socket 为数据源。经过 SQL 的处理，最终输出到控制台。在这个过程中，我们并没有编写具体的 flink 代码，也没有手动去打jar 包。我们只是将数据的处理流程声明在了一个配置文件中。

在背后，是 SeaTunnel 帮我们把配置文件翻译为具体的 flink 任务。配置化，低代码，易维护是 SeaTunnel 最显著的特点。

第 3 章 SeaTunnel 基本原理

3.1 SeaTunnel 的启动脚本

3.1.1 启动脚本的参数

截至目前， SeaTunnel 有两个启动脚本。

提交 spark 任务用 start-seatunnel-spark.sh。

提交 flink 任务则用 start-seatunnel-flink.sh。

本文档主要是结合 flink 来使用 seatunnel 的，所以用 start-seatunnel-flink.sh 来讲解。

start-seatunnle-flink.sh 可以指定 3 个参数

分别是：

–config 应用配置的路径

–variable 应用配置里的变量赋值

–check 检查 config 语法是否合法

3.1.2 --check 参数

截至本文档撰写时的 SeaTunnel 版本 v2.1.0 。check 功能还尚在开发中，因此–check 参数是一个虚设。目前 start-seatunnel-flink.sh 并不能对应用配置文件的语法合法性进行检查。而且 start-seatunnel-flink.sh 中目前没有对–check 参数的处理逻辑。

需要注意！使用过程中，如果没有使用–check 参数，命令行一闪而过。那就是你的配置文件语法有问题。

3.1.3 --config 参数和–variable 参数

–config 参数用来指定应用配置文件的路径。

–variable 参数可以向配置文件传值。配置文件内是支持声明变量的。然后我们可以通过命令行给配置中的变量赋值。

变量声明语法如下。

在配置文件的任何位置都可以声明变量。并用命令行参数–variable key=value 的方式

将变量值传进去，你也可以用它的短命令形式 -i key=value。传递参数时， key 需要和配置

文件中声明的变量名保持一致。

如果需要传递多个参数，那就在命令行里面传递多个-i 或–variable key=value。

比如：

bin/start-seatunnel-flink.sh --config/xxx.sh -i age=18 -i sex=man

3.1.4 示例2：配置中使用变量

1）我们在 example01.conf 的基础上创建 example02.conf。

2）修改文件

3）给 sql 插件声明一个变量，红色的是我们修改的地方。最终的配置文件如下。

4）开启 netcat 服务

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-kEBHcnvJ-1690266375143)(file:private/var/folders/wp/353tb1gx3hx7znx2kr1frrhr0000gn/T/com.kingsoft.wpsoffice.mac/wps-chinamanor/ksohtml//wps23.png)]

5）使用 SeaTunnel 来提交任务。 -i age=18 往命令行中

bin/start-seatunnel-flink.sh --config config/example01.sh -i age=18

6）接着，我们用 nc 发送几条数据看看效果。

7）在 flink 的 webUI 上我们看一下控制台的输出。最终发现未满 18 岁的李四被过滤掉了。

8）小结

通过传递变量，我们可以实现配置文件的复用。让同一份配置文件就能满足不同的业务需求。

3.1.5 给 flink 传递参数

在启动脚本的尾部，我们可以看到，start-seatunnel-flink.sh 会执行(exec)一条命令，这个命令会使用 flink 的提交脚本去向集群提交一个任务。而且在调用 bin/flink run 的时候，还传递了 PARAMS 作为 flink run 的参数。

如下图所示，我们可知，凡是–config 和 --variable 之外的命令行参数都被放到 PARAMS 变量中，最后相当于给 flink run 传递了参数。注意！命令行参数解析过程中没有涉及–check 参数处理。这也是为什么说它目前不支持–check 操作。

比如，我们可以在 seatunnel 启动脚本中，指定 flink job 并行度。

3.2 SeaTunnel 的配置文件

3.2.1 应用配置的 4 个基本组件

我们从 SeaTunnel 的 app 配置文件开始讲起。

一个完整的 SeaTunnel 配置文件应包含四个配置组件。分别是：

env{} source{} --> transform{} --> sink{}

在 Source 和 Sink 数据同构时，如果业务上也不需要对数据进行转换，那么 transform 中的内容可以为空。具体需根据业务情况来定。

3.2.2 env 块

env 块中可以直接写spark 或 flink 支持的配置项。比如并行度，检查点间隔时间。检查点 hdfs 路径等。在 SeaTunnel 源码的 ConfigKeyName 类中，声明了 env 块中所有可用的 key。

如图所示：

3.2.3 SeaTunnel 中的核心数据结构 Row

Row 是 SeaTunnel 中数据传递的核心数据结构。对 flink 来说， source 插件需要给下游的转换插件返回一个 DataStream，转换插件接到上游的 DataStream进行处理后需要再给下游返回一个 DataStream。最后 Sink 插件将转换插件处理好的 DataStream输出到外部的数据系统。

如图所示：

因为 DataStream可以很方便地和 Table 进行互转，所以将 Row 当作核心数据结

构可以让转换插件同时具有使用代码（命令式）和 sql （声明式）处理数据的能力。

3.2.4 source 块

source 块是用来声明数据源的。 source 块中可以声明多个连接器。比如：

需要注意的是，所有的 source 插件中都可以声明 result_table_name 。如果你声明了 result_table_name 。SeaTunnel 会将 source 插件输出的 DataStream转换为 Table 并注册在Table 环境中。当你指定了result_table_name ，那么你还可以指定field_name ，在注册时，给 Table 重设字段名(后面的案例中会讲解)。

因为不同 source 所需的配置并不一样，所以对 source 连接器的配置最好参考官方的文档。

3.2.5 transform 块

目前社区对插件做了很多规划，但是截至 v2.1.0 版本，可用的插件总共有两个，一个是 Split，另一个是 sql。

transform{}块中可以声明多个转换插件。所有的转换插件都可以使用 source_table_name，和 result_table_name。同样，如果我们声明了 result_table_name，那么我们就能声明 field_name。

我们需要着重了解一下 Split 插件和sql 插件的实现。但在此

在 SeaTunnel 中，一个转换插件的实现类最重要的逻辑在下述四个方法中。

1）处理批数据，DataSet进， DataSet出

DataSet processBatch(FlinkEnvironment env, DataSet data)

2）处理流数据，DataStram进， DataStream出

DataStream processStream(FlinkEnvironment env, DataStream dataStream)

3）函数名叫注册函数。实际上，这是一个约定，它只不过是每个 transform 插件作用于流

之后调用的一个函数。

4）处理一些预备工作，通常是用来解析配置。

Split 插件的实现

现在我们需要着重看一下 Split 插件的实现。

先回顾一下我们之前 example01.conf 中关于 transform 的配置。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-85Ou6Qic-1690266375149)(private/var/folders/wp/353tb1gx3hx7znx2kr1frrhr0000gn/T/com.kingsoft.wpsoffice.mac/wps-chinamanor/ksohtml//wps38.png)]

但是，需要注意，tranform 接口其实是留给了我们直接操作数据的能力的。也就是 processStream 方法。那么，一个 transform 插件其实同时履行了 process 和 udf 的职责，这是违反单一职责原则的。那要判断一个转换插件在做什么就只能从源码和文档的方面来加以

区分了。