@

• 概述
  • 定义
  • 使用场景
  • 特点
  • 工作流程
  • 连接器
  • 转换
  • 为何选择SeaTunnel
• 安装
  • 下载
  • 配置文件
  • 部署模式
  • 入门示例
  • 启动脚本
  • 配置文件使用参数示例
  • Kafka进Kafka出的ETL示例
  • Flink Run传递参数

概述

定义

http://seatunnel.incubator.apache.org/

https://github.com/apache/incubator-seatunnel

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

• SeaTunnel原名WaterDrop（水滴），自2021年 10月 12日更名为 SeaTunnel。2021年 12月 9日，SeaTunnel正式通过 Apache软件基金会的投票决议，以全票通过的优秀表现正式成为 Apache 孵化器项目。2022 年 3 月 18 日社区正式发布了首个 Apache 版本v2.1.0。
• 本质上SeaTunnel 没有对 Spark 和 Flink的改造，而是在这两者的基础上做了一层包装，主要运用控制反转设计模式，这也是 SeaTunnel实现的基本思想。
• SeaTunnel 的日常使用，就是编辑配置文件，编辑好的配置文件由 SeaTunnel 转换为具体的 Spark或 Flink 任务。

使用场景

• 海量数据同步
• 海量数据集成
• 海量数据的ETL
• 海量数据聚合
• 多源数据处理

特点

• 使用方便，配置灵活，近乎低代码开发
• 实时流媒体
• 离线多源数据分析
• 高性能、海量数据处理能力
• 模块化和插件机制，易于扩展
• 支持SQL对数据进行处理和聚合
• 支持Spark structured streaming
• 支持Spark 2.x

工作流程

SeaTunnel 有丰富的连接器且以 Spark 和 Flink 为引擎，所以可以很好地进行分布式的海量数据同步。一般来说使用SeaTunnel作为出仓入仓的工具，或者用于数据集成，主要流程如下：

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

数据处理管道由多个滤波器组成，以满足各种数据处理需求，最简单有效就是通过SQL直接构造数据处理管道。目前，SeaTunnel支持的过滤列表还在扩展中。此外，还可以开发自己的数据处理插件，得益于SeaTunnel系统基于插件化设计思想实现的强可扩展性。

连接器

• Source：Druid、Elasticsearch、Fake（伪造数据主要用于测试）、Feishu、Sheet、File、HBase、Hive、Http、Hudi、Iceberg、InfluxDb、Jdbc、Kafka、Kudu、MongoDb、Phoenix、Redis、Socket、Tidb、WebhookStream、Common OptionsSource common parametersNeo4j自主开发的输入插件。
• Sink： Assert、Clickhouse、 ClickhouseFile、 Console、 Doris、Druid、Elasticsearch、 Email、 File、 Hbase、 Hive、 Hudi、 Iceberg、InfluxDB、 Jdbc、 Kafka、 Kudu、 MongoDB、 Phoenix、 Redis、 TiDb、 Common Options、自主开发的输出插件。
• Flink-sql of SeaTunnel:Flink SQL ElasticSearch 连接器、Flink SQL JDBC 连接器、Flink SQL Kafka 连接、如何使用flink sql 模块。

连接器在实际开发中的使用直接查找官方对应章节即可

转换

用于转换或过滤插件包括如下：添加校验、转换、日期、删除、Grok、Json、KV、大写、小写、删除、重命名、重分区、替换、样本、拆分、Sql、表、截断、Uuid，自主开发的过滤器插件。详细可查阅官网地址

为何选择SeaTunnel

SeaTunnel将尽力解决在海量数据同步过程中可能遇到的问题:

• 数据丢失和重复。
• 任务积累和延迟。
• 低吞吐量。
• 在生产环境中应用的周期长。
• 缺乏应用程序运行状态监控。

安装

下载

Seatunnel支持多种安装方式，包括locally本地的二进制安装、docker安装、k8s安装,我们先以locally的方式安装和演示。

# 下载
wget https://dlcdn.apache.org/incubator/seatunnel/2.1.3/apache-seatunnel-incubating-2.1.3-bin.tar.gz
# 解压
tar -xvf apache-seatunnel-incubating-2.1.3-bin.tar.gz
# 进入目录
cd apache-seatunnel-incubating-2.1.3

配置文件

在config/seatunnel-env.sh中更改设置;如果使用Spark作为引擎，请更改SPARK_HOME，如果使用Flink，请更改FLINK_HOME默认的话，SPARK_HOME和FLINK_HOME用的都是对应的系统环境变量值，如果没有，使用:-后面的值，按需修改即可。我们本篇示例以Flink为主，在前面关于Flink文章提前设置好FLINK_HOME这个环境变量，这里直接使用即可。

在SeaTunnel中，最重要的是配置文件，可以通过配置文件定制自己的数据同步需求，最大限度地发挥SeaTunnel的潜力。配置文件包含几个部分:env、source、transform、sink；相关模块功能如下描述：

• env:用于添加一些引擎可选参数，无论是哪个引擎(Spark还是Flink)，都要在这里填写相应的可选参数。
• source:用于定义SeaTunnel需要在何处获取数据，并使用获取的数据进行下一步操作。可以同时定义多个源。现在支持的源检查SeaTunnel的源。每个源都有自己的特定参数来定义如何获取数据，SeaTunnel还提取每个源将使用的参数，例如result_table_name参数，用于指定当前源生成的数据名称，方便其他模块后续使用，详细可以查阅官方支持source连接器。
• transform:当有了数据源后，业务可能需要进一步处理数据，因此提供转换模块。当然这里使用了“可能”这个词意味着也可以直接将转换视为不存在的，直接从source到sink。与source一样，transform也具有属于每个模块的特定参数，详细可以查阅官方支持变换功能。
• sink：使用SeaTunnel目的是将数据从一个位置同步到另一个位置，因此定义如何以及在何处写入数据是至关重要的；通过SeaTunnel提供的sink模块，可以快速高效地完成这个操作。sink和source非常相似，但区别在于读取和写入，详细可以查阅官方支持sink连接器。
• 其他：当定义多个源和多个接收器时，每个接收器读取哪些数据，每个转换读取哪些数据?使用result_table_name和source_table_name两个关键配置。每个source模块都会配置一个result_table_name，表示该数据源生成的数据源名称，其他转换和接收模块可以使用source_table_name引用相应的数据源名称表示想读取数据进行处理。然后transform作为中间处理模块，可以同时使用result_table_name和source_table_name配置。并不是每个模块都配置了这两个参数，因为在SeaTunnel中有一个默认约定，如果没有配置这两个参数，那么将使用从上一个节点的最后一个模块生成的数据，特别是当只有一个源时就方便多了。

部署模式

在缩放中运行引擎,本节是关于引擎的，不是SeaTunnel本身，不做过多阐述，Spark和Flink都可以运行在不同类型的集群中，并且可以运行在任何规模的集群中。下面只展示建立在Spark或Flink引擎之上的SeaTunnel的基本用法，如果想扩展引擎集群请参阅Spark或Flink官网文档。

• Local Mode(Spark Only)

./bin/start-seatunnel-spark.sh \
--master local[4] \
--deploy-mode client \
--config ./config/application.conf

• Standalone Cluster

bin/start-seatunnel-flink.sh \
--config config-path

# -p 2 specifies that the parallelism of flink job is 2. You can also specify more parameters, use flink run -h to view
bin/start-seatunnel-flink.sh \
-p 2 \
--config config-path

• Yarn Cluster

bin/start-seatunnel-flink.sh \
-m yarn-cluster \
--config config-path

# -ynm seatunnel specifies the name displayed in the yarn webUI as seatunnel, you can also specify more parameters, use flink run -h to view
bin/start-seatunnel-flink.sh \
-m yarn-cluster \
-ynm seatunnel \
--config config-path

• Mesos Cluster(Spark Only)

# cluster mode
./bin/start-seatunnel-spark.sh \
--master mesos://ip:7077 \
--deploy-mode cluster \
--config ./config/application.conf

入门示例

如何部署Flink可以参考前一篇文章《新一代分布式实时流处理引擎Flink入门实战操作篇]》，在前面文章中部署的是最新版本1.15.1的，超过SeaTunnel官网支持Flink版本范围，因此可选择安装flink1.12.7或flink-1.13.6的版本，安装过程步骤参考之前文章

部署好flink1.12.7或flink-1.13.6的Flink,启动Standalone集群

# 环境变量配置flink得Home目录
export FLINK_HOME=/home/commons/flink-1.12.7
# 进入Flink目录
cd /home/commons/flink-1.12.7
# 启动flink的Standalone集群
./bin/start-cluster.sh

在config目录下官网提供flink和spark的多个简易配置模板，上一节说到SeaTunnel的连接器是非常丰富的，入门示例就以flink简单流式处理从Socket数据源读取数据，转换后输出到Console控制台，配置模板使用官网提供的flink.streaming.conf.template，创建flink-streaming-test-01.conf，内容如下：

vi config/flink-streaming-test-01.conf

env {
  # flink 配置
  execution.parallelism = 1
  #execution.checkpoint.interval = 10000
  #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
}
# 配置数据源
source {  
  SocketStream {
    host = hadoop1
    result_table_name = "socket_demo_table"
    field_name = "info"    
  }
}
# 配置转换插件
transform {
  Split{
    separator = "#"
    fields = ["name","age"]
  }
  sql {
    sql = "select info,split(info) from socket_demo_table"
  }
}
# 声明输出
sink {
  ConsoleSink {}
}

保存配置内容后，先在hadoop1开一个连接窗口，开启一个netcat服务来发送数据，nc -lk 9999 监听socket端口

# 启动seatunnel
./bin/start-seatunnel-flink.sh \
--config ./config/flink-streaming-test-01.conf

在hadoop1也即是ckserver1监听端口按#分隔输出下面几条数据

访问Flink的控制台UI页面http://hadoop1:8081/ ，找到刚运行的任务的日志，可以看到已经将info字段拆分为name和age两个字段输出

启动脚本

在前面示例使用start-seatunnle-flink.sh可以指定3个参数，分别是

• –-config参数用来指定应用配置文件的路径。上面入门的示例也已使用到了。

• –variable参数可以向配置文件传值。自从v1.2.4, SeaTunnel支持配置中的变量替换。此特性通常用于定时或非定时脱机处理，以替换时间和日期等变量配置文件内是支持声明变量的，可以通过命令行给配置中的变量赋值，变量声明语法如下：

  transform {
    sql {
      sql = "select * from user_view where city ='"${city}"' and dt = '"${date}"'"
    }
  }
  
  ./bin/start-seatunnel-flink.sh \
  --config ./config/flink-streaming-test-01.conf \
  -i city=shanghai \
  -i date=20190319
  

  • 在配置文件的任何位置都可以声明变量。并用命令行参数–variable key=value的方式将变量值传进去，你也可以用它的短命令形式 -i key=value。传递参数时，key需要和配置文件中声明的变量名保持一致。
  • 如果需要传递多个参数，那就在命令行里面传递多个-i或–variable key=value。

• –-check参数用来检查config语法是否合法。将入门示例故意修改错误格式，在第一行env前面加hello字符串，保存后运行后执行带上--check参数，通过检查出现缺少 env配置的错误提示。

  

配置文件使用参数示例

使用前面的入门示例，输出年龄大于指定输出参数年龄的数据，修改transform的sql语句

sql = "select * from (select info,split(info) as info_record from socket_demo_table) where age > '"${age}"'"

保存配置内容后，先在hadoop1开一个连接窗口，开启一个netcat服务来发送数据，nc -lk 9999 监听socket端口

# 启动seatunnel
./bin/start-seatunnel-flink.sh \
--config ./config/flink-streaming-test-01.conf -i age=25

在hadoop1也即是ckserver1监听端口按#分隔输出下面几条数据

访问Flink的控制台UI页面http://hadoop1:8081/ ，找到刚运行的任务的日志，可以看到已经将info字段拆分为name和age两个字段输出

Kafka进Kafka出的ETL示例

编写配置文件，vi config/flink-streaming-test-02.conf

env {
  # flink 配置
  execution.parallelism = 1
}
# 配置数据源
source {  
  KafkaTableStream {
    consumer.bootstrap.servers = "192.168.5.120:9092,192.168.5.121:9092,192.168.5.122:9092"
    consumer.group.id = "seatunnel-test"
    topics = seatunnel-kafka-in
    result_table_name = test
    format.type = csv
    schema = "[{\"field\":\"name\",\"type\":\"string\"},{\"field\":\"age\",\"type\":\"int\"}]"
    format.field-delimiter = ";"
    format.allow-comments = "true"
    format.ignore-parse-errors = "true"
  }
}
# 配置转换插件
transform {
  sql {
    sql = "select name,age from test where age > '"${age}"'"
  }
}
# 声明输出
sink {
  kafka {
    topics = "seatunnel-kafka-out"
    producer.bootstrap.servers = "192.168.5.120:9092,192.168.5.121:9092,192.168.5.122:9092"
  }
}

创建kafka的测试示例的输入和输出的topic

# 如果先创建过可以执行先删除topic，可选
./kafka-topics.sh --delete --bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092 --topic seatunnel-kafka-in
./kafka-topics.sh --delete --bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092 --topic seatunnel-kafka-out

./kafka-topics.sh --create --bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092 --replication-factor 1 --partitions 3 --topic seatunnel-kafka-in
./kafka-topics.sh --create --bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092 --replication-factor 1 --partitions 3 --topic seatunnel-kafka-out

启动seatunnel

./bin/start-seatunnel-flink.sh --config ./config/flink-streaming-test-02.conf -i age=25

查询Flink的控制台UI页面http://hadoop1:8081/ ，找到刚运行的任务查看其概览可以看到使用Kafka的Source和Sink

kafka生产数据

./kafka-console-producer.sh --broker-list kafka1:9092,kafka2:9092,kafka3:9092 --topic seatunnel-kafka-out

kafka消费数据，结果是正确的

# 消费输出
./kafka-console-consumer.sh --bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092 --topic seatunnel-kafka-in

Flink Run传递参数

-p 2指定flink job的并行度为2，还也可以指定更多的参数，使用flink运行-h查看,

# 将配置文件中env中execution.parallelism注释掉，改为参数传递的方式
# execution.parallelism = 1
# 启动seatunnel
./bin/start-seatunnel-flink.sh --config ./config/flink-streaming-test-02.conf -p 2 -i age=25

访问Flink的控制台UI页面http://hadoop1:8081/ 查看当前job有两个，3个slot使用了2个，可用只剩1个

**本人博客网站 **IT小神 www.itxiaoshen.com