网站建设合同编号,小白做电商从什么做起,网站建设保定,网站域名注销流程文章目录 前言第 1 章 Seatunnel 概述第 2 章 Seatunnel 安装和使用第 3 章 SeaTunnel 基本原理第 4章应用案例后记 前言
SeaTunnel安装包及代码#xff1a;
链接: https://pan.baidu.com/s/1JvgAZpqoOPJ0ecfxUbLo4Q 提取码: pur8 –来自百度网盘超级会员v4的分享
第 1 章 … 文章目录 前言第 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 连接器插件数据库类型SourceSinkBatchFake√ElasticSearch√√File√√Hive√√Hudi√√Jdbc√√MongoDB√√Neo4j√Phoenix√√Redis√√Tidb√√Clickhouse√Doris√Email√Hbase√√Kafka√Console√Kudu√√Redis√√StreamFakeStream√KafkaStream√SocketSTream√
1.5.2 Flink 连接器插件(Source)
Flink 连接器插件数据库类型SourceSinkDruid√√Fake√File√√InfluxDb√√Jdbc√√Kafka√√Socket√Console√Doris√ElasticSearch√
1.5.3 Spark Flink 转换插件
转换插件SparkFlinkAddCheckSumConvertDateDrop
GrokJson√KvLowercaseRemoveRenameRepartitionReplaceSampleSplit√√Sql√√TableTruncateUppercaseUuid
这部分内容来官网可以看出社区目前规划了大量的插件但截至 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 keyvalue 的方式
将变量值传进去你也可以用它的短命令形式 -i keyvalue。传递参数时 key 需要和配置
文件中声明的变量名保持一致。
如果需要传递多个参数 那就在命令行里面传递多个-i 或–variable keyvalue。
比如
bin/start-seatunnel-flink.sh --config/xxx.sh -i age18 -i sexman
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 age18 往命令行中
bin/start-seatunnel-flink.sh --config config/example01.sh -i age18
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 的职责这是 违反单一职责原则的。那要判断一个转换插件在做什么就只能从源码和文档的方面来加以
区分了。 最后需要叮嘱的是 指定 soure_table_name 对于 sql 插件的意义不大。因为 sql 插件可 以通过 from 子句来决定从哪个表里抽取数据。
3.2.6 sink 块
Sink 块里可以声明多个 sink 插件 每个 sink 插件都可以指定 source_table_name 。不过 因为不同 Sink 插件的配置差异较大 所以在实现时建议参考官方文档。
3.3 SeaTunnel 的基本原理
SeaTunnel 的工作原理简单明了。
1程序会解析你的应用配置并创建环境
2配置里 source{} transform{} sink{}三个块中的插件最终在程序中以List 集合的方式存 在。
3由Excution 对象来拼接各个插件这涉及到选择 source_table注册 result_table 等流程 注册 udf 等流程。并最终触发执行
可以参考下图 3.4 小结
最后我们用一张图将 SeaTunnel 中的重要概念串起来。 如果你愿意依托 sql 插件和 udf。单个配置文件也可以定义出比较复杂的工作流。但 SeaTunnel 的定位是一个数据集成平台。核心的功能是依托丰富的连接器进行数据同步 数
据处理并不是 SeaTunnel 的长处。所以在 SeaTunnel 中定义复杂的工作流或许是一种不值得 提倡的做法。
需要提醒的是 如果你不指定 source_table_name插件会使用它在配置文件上最近的
上一个插件的输出作为输入。
所以 我们可以通过使用依托表名表环境来实现复杂的工作流。 也可以减少表名的使用实现简单的数据同步通道。 第 4章应用案例
注意 下述示例请使用我们修改编译好的包。
4.1 Kafka 进 Kafka 出的简单 ETL
4.1.1 需求
对 test_csv 主题中的数据进行过滤仅保留年龄在 18 岁以上的记录。
4.1.2 需求实现
1首先 创建为 kafka 创建 test_csv 主题。
kafka-topics.sh --bootstrap-server hadoop102:9092 --create – topic test_csv --partitions 1 --replication-factor 1
2为 kafka 创建 test_sink 主题
kafka-topics.sh --bootstrap-server hadoop102:9092 --create – topic test_sink --partitions 1 --replication-factor 1
3编辑应用配置 4应用配置内容 5提交任务
bin/start-seatunnel-flink.sh --config config/example03.conf -i age18
6起一个 kafka console producer 发送 csv 数据(分号分隔) 7起一个 kafka console consumer 消费数据 我们成功地实现了数据从 kafka 输入经过简单的 ETL 再向 kafka 输出。
4.2 Kafka 输出到 Doris 进行指标统计
4.2.1 需求
使用回话日志统计用户的总观看视频数用户最常会话市场用户最小会话时长用户最后一次会话时间。
4.2.2 需求实现
1在资料中有一个伪数据的生成脚本将它拷贝到服务器的任意位置 2执行以下命令安装 python 脚本需要的两个依赖库 3使用 mysql 客户端连接 doris
[atguiguhadoop102 fake_data]$ mysql -h hadoop102 -P 9030 - uatguigu -p123321
4手动创建 test_db 数据库。 5使用下述 sql 语句建表 7使用 python 脚本向 kafka 中生成伪数据
[atguiguhadoop102 fake_data]$ python3 fake_video.py --bootstrap- server hadoop102:9092 --topic test_video
8查看 doris 中的结果。 后记
猜你想看下一篇文章 教你如何成为开源项目SeaTunnel的贡献者
