Apache Seatunnel - 架构解析

时间：2023-02-09 12:33:45

概述

​​Seatunnel​​ 是一个非常易用，高性能、支持​​实时流式​​和​​离线批处理​​的海量数据处理产品，架构于​​Apache Spark​​ 和 ​​Apache Flink​​之上，开源项目地址：https://github.com/apache/incubator-seatunnel

版本演变

Seatunnel原名为Waterdrop，在更名之后正式孵化为Apache项目，同时对于两个名字也对应了不同的版本，Waterdrop指1.x版本，Seatunnel指2.x版本，对于1.x和2.x有以下区别：

关键功能

1.x

2.x



支持spark



yes



yes



支持flink



no



yes



主要开发语言



scala



java



主要构建工具



sbt



maven

为什么我们需要Seatunnel

Apache Spark和Apache Flink对于分布式数据处理和流式数据处理来说是一个伟大的进步，但较高的使用门槛让数据处理人员需要学习spark和flink复杂的运行机制和api才能够使用的更加顺畅，为降低数据处理门槛，且让spark和flink变得更加易用，减少学习成本，加快分布式数据处理在生产环境的落地，Seatunnel应运而生。

基于当前大多数数据处理工作的一些思考

  1. 更多的数据处理是重复的
  2. 数据处理的代码是冗余的
  3. 在数据处理工作中有一部分的比例是数据同步工作，在离线数仓计算完成之后，往往会将ads层数据同步至对查询专门优化过的OLAP​​数据库​​（ck、es等）中以提供前端​​报表​​展示的功能，这些功能是否可以沉淀？是否可以复用？
  4. 在数据处理过程中，可能会有多种异构数据源接入的需求，例如file、redis、hdfs、kafka、mysql….，在面对这种异构数据源集成的需求时如何去更好的应对？
  5. 在当前越来越多大数据框架面世的基础上，大数据处理的方向慢慢变向了sql化和低代码化，在业务看来无论底层有多少数据都会是落成一张表或是多张表，如果可以使用sql就能够计算海量数据并快速获取正确结果，对于整个业务部门对于数据的利用将更加高效
  6. 假设企业中需要组建数据中台，如何对外快速提供数据处理的中台能力

Seatunnel可以解决的业务痛点

  1. 背靠spark和flink两大分布式数据框架，天生具有分布式数据处理的能力，使业务可以更加专注于数据的价值挖掘与处理，而不是专注于底层技术对于大数据的兼容和开发
  2. 利用spark和flink分布式框架对于异构数据源的兼容，可以实现快速的异构数据源同步和接入
  3. 高度抽象业务处理逻辑，减少代码的冗余和重复开发

Seatunnel优势与缺点

优势

  1. 简单易用，灵活配置，无需开发
  2. 模块化和插件化
  3. 支持利用SQL做数据处理和聚合
  4. 由于其高度封装的计算引擎架构，可以很好的与中台进行融合，对外提供分布式计算能力

缺点

  1. Spark支持2.2.0 - 2.4.8，不支持spark3.x
  2. Flink支持1.9.0，目前flink已经迭代至1.14.x，无法向上兼容
  3. Spark作业虽然可以很快配置，但相关人员还需要懂一些参数的调优才能让作业效率更优

相关竞品及对比

  • FlinkX，现已更名为chunjun
  • StreamX
  • DataX

关键功能

Seatunnel

FlinkX

StreamX

DataX



spark是否支持



yes



no



yes



no



flink是否支持



yes，高版本兼容性不好



yes，高版本兼容性不好



yes，高版本兼容性好



no



部署难度



轻松



中等



较难



容易



主要功能对比



etl、数据同步



数据同步



flink任务可视化部署



数据同步

Seatunnel核心理念与内核原理

核心概念

  1. 整个Seatunnel设计的核心是利用设计模式中的“控制翻转”或者叫“依赖注入”，主要概括为以下两点：
  1. 上层不依赖底层，两者都依赖抽象
  2. 流程代码与业务逻辑应该分离
  1. 对于整个数据处理过程，大致可以分为以下几个流程：​​输入​​ -> ​​转换​​ -> ​​输出​​，对于更复杂的数据处理，实质上也是这几种行为的组合：

Apache Seatunnel - 架构解析

Apache Seatunnel - 架构解析

Apache Seatunnel - 架构解析

  1. Seatunnel对于这几种数据处理的行为进行高度的抽象，在基于这层比较完善的抽象之上，对业务数据处理过程中的80%重复操作进行沉淀，做成可热插拔的插件，这样业务处理逻辑与整个数据处理实现了解耦，让用户更专注于业务的落地与实施。

内核原理

  1. 插件的动态注册使用了java spi技术，保证了框架的灵活扩展，设计思路参考了presto、es等，有兴趣的同学可以下去自行研究，es使用了google guice，presto使用的就是上面提到的java spi
  2. 在以上理论基础上，数据的转换需要做一个统一的抽象与转化，很契合的是spark或者flink都已经为我们做好了这个工作，spark的DataSet，flink的DataSet、DataStream都已经是对接入数据的一个高度抽象，本质上对数据的处理就是对这些数据结构的转换，同时这些数据在接入进来之后可以注册成上下文中的表，基于表就可以使用SQL进行处理
  3. 整个Seatunnel通过配置文件生成的是一个spark job或者flink job
  4. 技术栈包括以下：
  1. Java
  2. Scala
  3. Flink
  4. Spark
  5. Java spi

Spark插件体系架构设计

Apache Seatunnel - 架构解析

Flink插件体系架构设计

Apache Seatunnel - 架构解析

程序执行流程

Apache Seatunnel - 架构解析

最上层插件抽象实现细节

public interface Plugin<T> extends Serializable {
// 配置文件的key
String RESULT_TABLE_NAME = "result_table_name";
String SOURCE_TABLE_NAME = "source_table_name";

// 设置每个插件的config
void setConfig(Config config);

// 获取插件的配置 
Config getConfig();

// 对于config的校验
CheckResult checkConfig();

// 插件前准备
void prepare(T prepareEnv);
}

复制

Spark插件上层抽象实现细节

批处理Batch

Source

trait BaseSparkSource[Data] extends BaseSource[SparkEnvironment] {

protected var config: Config = ConfigFactory.empty()

override def setConfig(config: Config): Unit = this.config = config

override def getConfig: Config = config

def getData(env: SparkEnvironment): Data;

}

复制

Transform

trait BaseSparkTransform extends BaseTransform[SparkEnvironment] {

protected var config: Config = ConfigFactory.empty()

override def setConfig(config: Config): Unit = this.config = config

override def getConfig: Config = config

def process(data: Dataset[Row], env: SparkEnvironment): Dataset[Row];

}

复制

Output

trait BaseSparkSink[OUT] extends BaseSink[SparkEnvironment] {

protected var config: Config = ConfigFactory.empty()

override def setConfig(config: Config): Unit = this.config = config

override def getConfig: Config = config

def output(data: Dataset[Row], env: SparkEnvironment): OUT;

}

复制

流处理Stream

trait SparkStreamingSource[T] extends BaseSparkSource[DStream[T]] {

def beforeOutput(): Unit = {}

def afterOutput(): Unit = {}

def rdd2dataset(sparkSession: SparkSession, rdd: RDD[T]): Dataset[Row]

def start(env: SparkEnvironment, handler: Dataset[Row] => Unit): Unit = {
getData(env).foreachRDD(rdd => {
val dataset = rdd2dataset(env.getSparkSession, rdd)
handler(dataset)
    })
  }

}

复制

Flink插件上层抽象实现细节

批处理Batch

Source

public interface FlinkBatchSource<T> extends BaseFlinkSource {

DataSet<T> getData(FlinkEnvironment env);
}

复制

Transform

public interface FlinkBatchTransform<IN, OUT> extends BaseFlinkTransform {

DataSet<OUT> processBatch(FlinkEnvironment env, DataSet<IN> data);

}

复制

Output

public interface FlinkBatchSink<IN, OUT> extends BaseFlinkSink {

DataSink<OUT> outputBatch(FlinkEnvironment env, DataSet<IN> inDataSet);

}

复制

流处理Stream

Source

public interface FlinkStreamSource<T> extends BaseFlinkSource {

DataStream<T> getData(FlinkEnvironment env);

}

复制

Transform

public interface FlinkStreamTransform<IN, OUT> extends BaseFlinkTransform {

DataStream<OUT> processStream(FlinkEnvironment env, DataStream<IN> dataStream);
}

复制

Output

public interface FlinkStreamSink<IN, OUT> extends BaseFlinkSink {

DataStreamSink<OUT> outputStream(FlinkEnvironment env, DataStream<IN> dataStream);

}

复制

自定义插件步骤

  1. 针对不同的框架和插件类型继承对应的接口，接口中的核心处理方法
  2. 在java spi中注册
  3. 将自己定义的jar包放在Seatunnel主jar包的plugins目录下

Java spi原理解析

概念

SPI全称Service Provider Interface，是Java提供的一套用来被第三方实现或者扩展的接口，它可以用来启用框架扩展和替换组件，SPI的作用就是为这些被扩展的API寻找服务实现

API和SPI的区别

API-(​​Application Programming Interface​​)大多数情况下，都是​​实现方​​制定接口并完成对接口的实现，​​调用方​​仅仅依赖接口调用，且无权选择不同实现。从使用人员上来说，API 直接被应用开发人员使用，SPI-(​​Service Provider Interface​​)是​​调用方​​来制定接口规范，提供给外部来实现​​调用方​​选择自己需要的外部实现。从使用人员上来说，SPI 被框架扩展人员使用

实现demo

  1. 定义接口
package com.tyrantlucifer;

public interface Animal {
void shut();
 }

复制

  1. 定义main函数，使用service loader进行动态加载
package com.tyrantlucifer;

import java.util.ServiceLoader;

public class Main {
public static void main(String[] args) {
ServiceLoader<Animal> services = ServiceLoader.load(Animal.class);
for (Animal service : services) {
service.shut();
         }
     }
 }

复制

  1. 实现接口
package com.tyrantlucifer;

public class Cat implements Animal {
public void shut() {
System.out.println("cat shut miao miao!!!");
     }
 }

复制

package com.tyrantlucifer;

public class Dog implements Animal{
public void shut() {
System.out.println("dog shut wang wang!!!");
     }
 }

复制

注册spi，需要在resources/META-INF/services下新建以接口全类名的文件，比如我们这次的接口​​com.tyrantlucifer.Animal​​，那么就新建一个​​com.tyrantlucifer.Animal​​文件，并在文件中添加自己的实现类：

Apache Seatunnel - 架构解析

com.tyrantlucifer.Cat
com.tyrantlucifer.Dog

复制

Seatunnel demo演示

  1. Spark
spark {
spark.streaming.batchDuration = 5
spark.app.name = "seatunnel"
spark.ui.port = 13000
 }

input {
socketStream {}
 }

filter {
split {
fields = ["msg", "name"]
delimiter = ","
   }
 }

output {
stdout {}
 }

复制

  1. Flink
env {
execution.parallelism = 1
 }

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 {}
 }

复制

  1. 自定义插件
class MyStdout extends BaseOutput {

var config: Config = ConfigFactory.empty()

/**
    * Set Config.
    * */
override def setConfig(config: Config): Unit = {
this.config = config
   }

/**
    * Get Config.
    * */
override def getConfig(): Config = {
this.config
   }

override def checkConfig(): (Boolean, String) = {
if (!config.hasPath("limit") || (config.hasPath("limit") && config.getInt("limit") >= -1)) {
       (true, "")
     } else {
       (false, "please specify [limit] as Number[-1, " + Int.MaxValue + "]")
     }
   }

override def prepare(spark: SparkSession): Unit = {
super.prepare(spark)

val defaultConfig = ConfigFactory.parseMap(
Map(
"limit" -> 100,
"format" -> "plain" // plain | json | schema
       )
     )
config = config.withFallback(defaultConfig)
   }

override def process(df: Dataset[Row]): Unit = {

val limit = config.getInt("limit")

var format = config.getString("format")
if (config.hasPath("serializer")) {
format = config.getString("serializer")
     }
format match {
case "plain" => {
if (limit == -1) {
df.show(Int.MaxValue, false)
         } else if (limit > 0) {
df.show(limit, false)
         }
       }
case "json" => {
if (limit == -1) {
df.toJSON.take(Int.MaxValue).foreach(s => println(s))

         } else if (limit > 0) {
df.toJSON.take(limit).foreach(s => println(s))
         }
       }
case "schema" => {
df.printSchema()
       }
     }
   }
 }
标签：

相关文章