皮皮网

1.SeaTunnel连接器V1到V2的码分架构演进与探究
2.有什么关于 Spark 的书推荐？
3.Spark Core读取ES的分区问题分析

SeaTunnel连接器V1到V2的架构演进与探究

       核心概念

       SeaTunnel设计的核心是利用设计模式中的控制翻转或依赖注入，主要包括以下两点：

       数据处理过程大致分为输入 -> 转换 -> 输出，码分更复杂的码分数据处理实质上也是这些行为的组合。

       内核原理

       SeaTunnel将数据处理的码分各种行为抽象成Plugin，并使用SPI技术进行动态注册，码分设计思路保证了框架的码分微擎源码1.8灵活扩展。在以上理论基础上，码分数据的码分转换与处理还需要做统一的抽象，如著名的码分异构数据源同步工具DataX，也对数据单条记录做了统一抽象。码分

       SeaTunnel V1架构体系中，码分由于背靠Spark和Flink两大分布式计算框架，码分框架已经为我们做好了数据源抽象的码分工作，Flink的码分DataStream、Spark的码分process hacker源码DataFrame已经是对接入数据源的高度抽象。在此基础上，我们只需要在插件中处理这些数据抽象即可。同时，借助Flink和Spark提供的SQL接口，还可以将每次处理完的数据注册成表，方便用SQL进行处理，减少代码的开发量。

       实际上，SeaTunnel的最终目的是自动生成一个Spark或Flink作业，并提交到集群中运行。

       SeaTunnel连接器V1 API解析架构概览

       目前在项目dev分支下，SeaTunnel连接器V1 API所在的模块如图所示：

       seatunnel-api-base

       在基础模块中，有以下代码：

       为了更清晰地理解这些类之间的关系，笔者制作了一张简单的view源码阅读UML类图：

       整个API的组成可以大体分为三部分：构建层接收命令参数构建执行器，执行器初始化上下文，上下文注册插件并启动插件，至此，整个作业开始运行。

       seatunnel-api-spark

       在Spark引擎API层有以下代码：

       同样，笔者整理了一张UML类图来表示它们之间的关系：

       整个流程与Base模块一致，在此不再赘述，有兴趣的读者可以自行查看源码。

       seatunnel-api-flink

       在Flink引擎API层有以下代码：

       同样，笔者整理了一张UML类图来表示它们之间的关系：

       整个流程与Base模块一致，在此不再赘述，有兴趣的读者可以自行查看源码。

       SeaTunnel连接器V1运行原理启动器模块概览

       整个项目的最外层启动类都放在以下模块中：

       与连接器V1有关的模块如下：

       执行流程

       为了更好地理解SeaTunnel V1的启动流程，笔者制作了一张简单的服务php源码时序图：

       程序最外层的启动由start-seatunnel-${ engine}.sh开始，用户将配置文件从脚本传入，脚本调用org.apache.seatunnel.core.spark.SparkStarter或org.apache.seatunnel.core.flink.FlinkStarter。实际上，这个类只做一个工作：将所有参数拼接成spark-submit或flink命令，然后脚本接收spark-submit或flink命令并提交到集群中。提交到集群中真正执行job的类实际上是org.apache.seatunnel.spark.SeatunnelSpark或org.apache.seatunnel.flink.SeatunnelFlink。读者如果想直接深入了解作业启动核心流程的话，推荐阅读这两个类的源码。

       执行原理SparkFlinkSeaTunnel连接器V2 API解析架构概览

       目前在项目dev分支下，SeaTunnel连接器V2 API所在的模块如图所示：

       数据抽象

       SeaTunnel连接器V2 API在数据层面做了抽象，定义了自己的数据类型，这是与连接器V1最大的不同点。连接器V1使用的是引擎数据抽象的能力，但连接器V2自己提供了这个异构数据源统一的打字测试源码能力。

       在所有的Source连接器和Sink连接器中，处理的都是SeaTunnelRow类型数据，同时SeaTunnel也对内设置了数据类型规范。所有通过Source接入进来的数据会被对应的连接器转化为SeaTunnelRow送到下游。

       API Common

       在API common包下有以下接口的定义：

       在这里，由于篇幅关系，只介绍比较核心的几个接口：

       具体接口中有哪些方法，读者可以自行阅读对应类的源码，在此不再赘述。

       API Source

       在API source包下有以下接口的定义：

       在这里，由于篇幅关系，只介绍比较核心的几个接口：

       API Sink

       在API sink包下有以下接口的定义：

       在这里，由于篇幅关系，只介绍比较核心的几个接口：

       小结

       连接器V2在架构分层上与计算引擎进行解耦，定义了自己的元数据定义以及数据类型定义，在API层和计算引擎层增加了翻译层，将SeaTunnel自定义的数据源通过翻译层接入到引擎中，从而真正实现接口和引擎分离的目的。

       SeaTunnel连接器V2运行原理启动器模块概览

       整个项目的最外层启动类都放在以下模块中：

       与连接器V2有关的模块如下：

       执行流程

       为了更好地理解SeaTunnel V2的启动流程，笔者制作了一张简单的时序图：

       程序最外层的启动由start-seatunnel-${ engine}-new-connector.sh开始，用户根据将配置文件从脚本传入，脚本调用org.apache.seatunnel.core.spark.SparkStarter或org.apache.seatunnel.core.flink.FlinkStarter。实际上，这个类只做一个工作：将所有参数拼接成spark-submit或flink命令，然后脚本接收spark-submit或flink命令并提交到集群中。提交到集群中真正执行job的类实际上是org.apache.seatunnel.spark.SeatunnelSpark或org.apache.seatunnel.flink.SeatunnelFlink。读者如果想直接深入了解作业启动核心流程的话，推荐阅读这两个类的源码，连接器V2和连接器V1的启动流程基本一致。

       SeaTunnel V2 on Spark

       SeaTunnel Source连接器V2将异构数据源接入，生成以SeaTunnelRow为基本单位的数据源，在翻译层实现了Spark DataSource API V2，翻译层使得Spark可以接入以SeaTunnelRow为基本单位的数据源，从而实现无缝接入Spark的目的。

       关于Spark DataSource API V2的详细信息，读者可以参考：/session/apache-spark-data-source-v2。由于这篇文章的主题并不是介绍Spark的特性，所以在此不再赘述。

       SeaTunnel V2 on Flink

       SeaTunnel Source连接器V2将异构数据源接入，生成以SeaTunnelRow为基本单位的数据源，同时在翻译层实现了Flink source function和Flink sink function。翻译层使得Flink可以接入以SeaTunnelRow为基本单位的数据源，从而实现无缝接入Flink的目的。

       关于Flink source Function和Flink sink function的详细信息，读者可以参考：/elastic/elasticsearch-hadoop这个gradle工程，可以直接导入idea，然后切换到7.x版本。

       接下来，找到ScalaEsRDD，发现getPartitions方法是在其父类中实现的，方法内容如下：

       esPartitions是一个lazy型的变量：

       这种声明的原因是什么呢？

       lazy+transient的原因大家可以思考一下。

       RestService.findPartitions方法只是创建客户端获取分片等信息，然后调用，分两种情况调用两个方法：

       a).findSlicePartitions

       这个方法实际上是在5.x及以后的ES版本，同时配置了

       之后，才会执行。实际上就是将ES的分片按照指定大小进行拆分，必然要先进行分片大小统计，然后计算出拆分的分区数，最后生成分区信息。具体代码如下：

       实际上，分片就是通过游标方式，对_doc进行排序，然后按照分片计算得到的分区偏移进行数据读取，组装过程是通过SearchRequestBuilder.assemble方法实现的。

       这个实际上会浪费一定的性能，如果真的要将ES与Spark结合，建议合理设置分片数。

       b).findShardPartitions方法

       这个方法没有疑问，一个RDD分区对应于ES index的一个分片。

       3.总结

       以上就是Spark Core读取ES数据时，分片和RDD分区的对应关系分析。默认情况下，一个ES索引分片对应Spark RDD的一个分区。如果分片数过大，且ES版本在5.x及以上，可以配置参数

       进行拆分。