SeaTunnel 高性能分布式数据集成平台_Adobee Chen

irpas 02-07 3710

介绍

为什么我们需要SeaTunnel

SeaTunnel的特点

SeaTunnel的工作流程

SeaTunnel支持的插件

Connector

Trandform

环境依赖

安装部署(官方案例flink引擎)

1.下载安装包

2.解压

?3.开启flink

4.修改config/seatunnel-env.sh，配置flink目录

?5.?vim config/application.conf

6.开启 nc -l -p 9999

7.?./bin/start-seatunnel-flink.sh --config ./config/application.conf

?8.nc -l -p 9999输入xg#1995

?9.查看Task Manager stdout

同步hive数据到clickhouse例子（spark引擎）

hive、clickhouse版本及配置

seaTunnel spark引擎配置?

数据准备

启动

查看结果

介绍

SeaTunnel是一个非常好用的超高性能分布式数据集成平台，支持海量数据的实时同步。每天可以稳定高效地同步数百亿数据，已应用于近百家企业的生产中。

为什么我们需要SeaTunnel

SeaTunnel会尽力解决海量数据同步中可能遇到的问题：

数据丢失和重复任务累积和延迟低吞吐量长周期应用于生产环境缺乏应用运行状态监控 SeaTunnel的特点简单易用，配置灵活，低代码开发实时流式传输离线多源数据分析高性能、海量数据处理能力模块化和插件机制，易于扩展支持SQL数据处理和聚合支持Spark结构化流支持Spark 2.x SeaTunnel的工作流程

Source[Data Source Input] -> Transform[Data Processing] -> Sink[Result Output]

数据处理管道由多个过滤器构成，以满足各种数据处理需求。如果习惯SQL，也可以直接通过SQL构建数据处理管道，简单高效。目前，SeaTunnel 支持的过滤器列表还在扩展中。此外，您可以开发自己的数据处理插件，因为整个系统易于扩展。

SeaTunnel支持的插件 Connector clickhousedorisdruidelasticsearchemailfakefilehbasehivehudiiceberginfluxDbjdbckafkakuduMongoDBNeo4jPhoenixredissockettidb Trandform AddCheckSumConvertDateDropGrokJsnKvLowercaseRemoveRenameRepartitionReplaceSampleSplitSqlTableTruncateUppercaseUuid 环境依赖

1.java运行环境，java>=1.8

2.spark 2.x(尚不支持 Spark 3.x)

如果要在集群环境中运行SeaTunnel,可以使用以下任何Spark集群环境

Spark on YarnSpark Standalone

如果数据量小，或者只是为了功能验证，也可以在没有集群环境的情况下本地启动，因为SeaTunnel支持单机运行。

3.Flink 1.9.0及以上版本

注意：SeaTunnel 2.0支持在Spark和Flink上运行

安装部署(官方案例flink引擎) 1.下载安装包

Apache SeaTunnel

2.解压

?3.开启flink

注：seatunnel 2.1.0支持的flink最高保本是1.13.5，使用1.14.1会报错。（flink支持1.9.0-1.13.5）

4.修改config/seatunnel-env.sh，配置flink目录

?5.?vim config/application.conf ?env { ? # You can set flink configuration here ? execution.parallelism = 1 ? #execution.checkpoint.interval = 10000 ? #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint" } source { ? ? SocketStream{ ? ? ? ? ? result_table_name = "fake" ? ? ? ? ? field_name = "info" ? ? } } transform { ? Split{ ? ? separator = "#" ? ? fields = ["name","age"] ? } ? sql { ? ? sql = "select * from (select info,split(info) as info_row from fake) t1" ? } } sink { ? ConsoleSink {} } 6.开启 nc -l -p 9999 7.?./bin/start-seatunnel-flink.sh --config ./config/application.conf

?8.nc -l -p 9999输入xg#1995

?9.查看Task Manager stdout

同步hive数据到clickhouse例子（spark引擎） hive、clickhouse版本及配置

hive版本：3.1.2?

hive source 配置

clickhouse版本: 21.7.3.14?

clickhouse sink配置

seaTunnel spark引擎配置? 前提

seaTunnel安装前，需要准备好spark。spark 只支持2.x，不支持spark 3.x 。 deply-mode=local,下载解压后，无需任何配置即可提交Spark模式任务。如果任务是Standalone cluster 或者Yarn cluster、Mesos cluster模式，先配置好spark。（这个案例是local模式，解压spark 2.x安装包。如果你的环境是spark 3.x ,要解压一个spark2.x的安装包）

解压spark并在seaTunnel config下配置spark路径

vim config/seatunnel-env.sh

修改SPARK_HOME=/xxxx/spark-2.4.8-bin-hadoop2.7

数据准备 hive建表语句 create table student ( id string comment 'id', name string comment '姓名' )COMMENT '测试'; 插入数据

insert into student values ('1','cy');

insert into student values ('2','cxy');

clickhouse建表语句 create table student ( id String, name String )engine =MergeTree primary key (id) ORDER BY (id) seaTunnel配置

vim hive2ck.config

env { ? spark.sql.catalogImplementation ="hive" ? spark.app.name="hive2ck" ? spark.executor.instances=4 ? spark.executor.cores =4 ? spark.executor.memory = "4g"

}

source { ? ? ?hive { ? ? ? ? ? ? pre_sql ="select id,name from default.student" ? ? ? ? ? ? result_table_name="student" } }

transform {}

sink { ? clickhouse { ? ? ? host = "henghe-051:8123" ? ? ? database ="default" ? ? ? table ="student" ? ? ? fields=["id","name"] ? ? ? username="default" ? ? ? password ="root" ? ? ? bulk_size =20000 } }

注意：必须确保 hive matastore正在使用中。启动命令hive --service metastore服务 default port 9083 cluster,client,local模式,hive-site.xml必须放在$HADOOP_CONF任务提交节点的目录下（或放到$SPARK_HOME/conf）,IDE本地Debug放在resources目录下。

本demo放在了spark-2.4.8-bin-hadoop2.7/conf下??

启动启动命令

?./bin/start-seatunnel-spark.sh --master 'local[2]' --deploy-mode client --config jobs/hive2ck.config

查看结果