内容简介:上面代码中UnsortedGrouping并不是一个DataSet实现,而是一个用来处理groupBy操作的中间结构,它继承自Grouping抽象类,类定义如下所示:对于流式Job程序提交运行的核心代码,如下所示:上面代码中的Plan和StreamGraph,都是Pipeline接口的具体实现,这两个实现分别用来表示批式和流式Job程序的拓扑,内部封装了用于构建生成JobGraph所需要的全部信息。
使用Flink开发批式或流式Job,除了基本的处理逻辑与实际应用场景相关,我们更关心的是Flink提供的基本框架,是如何在API层面进行统一处理的,或者说尽量使API统一,这样有助于我们对Flink框架更深入地理解。目前使用Flink 1.10版本开发批式和流式Job,在API层面来看,大部分还是比较统一的,但是由于批式和流式场景还是有一定的差异,想要完全统一还是有一定难度。
Flink数据流编程模型的分层设计,如下图所示:
这里,我们关注的是上面的Core APIs层,结合批式和流式Job开发与提交的过程进行分析,对比对于批式和流式场景下,如何生成最终的JobGraph,以及生成的过程有什么不同。
编程API设计
Flink编程API中,我们开发的数据处理Job程序,都是始于一个Source,对应的输入数据源,终于一个Sink,对应输出的存储系统,中间是各种丰富的处理算子。但是对于批式和流式编程API,从代码层面对应的抽象基本上是名称不同,具体逻辑功能比较一致:批式编程API对批式Job DAG中每个节点的抽象使用的是DataSet,而流式编程API中对应的是DataStream。
对于批式Job DAG中,DataSet的类设计体系,如下图所示:
相关的类都在包org.apache.flink.api.java下面,通过上图可以看出,主要分为4类:DataSource、DataSink、SingleInputOperator、TwoInputOperator。其中,DataSink并没有继承自DataSet,但是作为批式Job DAG的输出节点抽象,也还是与上图中各个Operator有直接或间接的关系。
在编写批式Job过程中,输入是一个DataSource(实际也是DataSet),处理中每经过一个转换操作(Transformation),都会生成一个新的类型的DataSet,这在API上看是统一的,实际底层稍微有一点不同。比如,经过groupBy操作后,返回的是一个UnsortedGrouping,它不是一个DataSet实现,而是一种中间结构,封装了很多有用的信息以供翻译过程中使用,通过使用这种中间结构能够更好地处理复杂的转换操作,通过下面代码来看可能更直观一些,在DataSet中查看groupBy()方法代码,如下所示:
public UnsortedGrouping<T> groupBy(int... fields) { return new UnsortedGrouping<>(this, new Keys.ExpressionKeys<>(fields, getType())); }
上面代码中UnsortedGrouping并不是一个DataSet实现,而是一个用来处理groupBy操作的中间结构,它继承自Grouping抽象类,类定义如下所示:
@Public public abstract class Grouping<T> { protected final DataSet<T> inputDataSet; protected final Keys<T> keys; protected Partitioner<?> customPartitioner; public Grouping(DataSet<T> set, Keys<T> keys) { if (set == null || keys == null) { throw new NullPointerException(); } if (keys.isEmpty()) { throw new InvalidProgramException("The grouping keys must not be empty."); } this.inputDataSet = set; this.keys = keys; } @Internal public DataSet<T> getInputDataSet() { return this.inputDataSet; } @Internal public Keys<T> getKeys() { return this.keys; } @Internal public Partitioner<?> getCustomPartitioner() { return this.customPartitioner; } }
对于流式Job DAG中,类设计方面稍有不同,Flink使用了DataStream和StreamOperator这两个类设计体系。我们先看DataStream类设计体系,如下图所示:
DataStream表示在流式Job DAG中每一步转换操作之前与之后,都对应着一个DataStream的数据结构,它内部封装了与转换操作相关的处理逻辑,其实就是StreamOperator。对应上图中,我们举几个编写流式处理程序的例子说明:调用StreamExecutionEnvironment.readTextFile()时会生成一个DataStreamSource,调用keyBy()时会生成一个KeyedStream,调用split()时会生成一个SplitStream,调用iterate()时会生成一个IterativeStream。
下面看下StreamOperator类的设计体系,如下图所示:
StreamOperator类体系中,可以分为两类,一类是OneInputStreamOperator,表示只有一个输入的StreamOperator,比如编程时调用filter()时,内部通过一个OneInputStreamOperator(具体实现为StreamFilter)来封装一个FilterFunction;另一个是TwoInputStreamOperator表示具有两个输入的StreamOperator,比如编程时调用coGroup()或join()时,会对应两个输入DataStream,而DataStream中封装的就是StreamOperator。
生成JobGraph对象
编写批式Job程序,使用执行上线文环境对象ExecutionEnvironment,而流式使用的是StreamExecutionEnvironment。通过用户编程API构建好DAG Job后,都是通过调用执行上线文环境对象的execute()方法提交Job去运行。无论是批式Job还是流式Job,它们在提交执行过程中,有相同的流程,也有不同的流程,通过识别这个过程中涉及相同/不同的API对象,我们抽象出如下流程概念图:
上图中,左侧是批式Job通过API构建并提交到计算集群,基于DataSet进行编程实现,始于DataSource,终于DataSink;右侧是流式Job通过API构建并提交到计算集群,基于DataStream进行编程实现,始于DataStreamSource,终于DataStreamSink。中间部分,跨两个不同环境上下文对象是在提交Job过程中公共的抽象。
对于批式Job程序提交,核心代码如下所示:
final Plan plan = createProgramPlan(jobName); final PipelineExecutorFactory executorFactory = executorServiceLoader.getExecutorFactory(configuration); CompletableFuture<JobClient> jobClientFuture = executorFactory .getExecutor(configuration) .execute(plan, configuration);
对于流式Job程序提交运行的核心代码,如下所示:
StreamGraph streamGraph = getStreamGraphGenerator().setJobName(jobName).generate(); if (clearTransformations) { this.transformations.clear(); } … … final PipelineExecutorFactory executorFactory = executorServiceLoader.getExecutorFactory(configuration); CompletableFuture<JobClient> jobClientFuture = executorFactory .getExecutor(configuration) .execute(streamGraph, configuration);
上面代码中的Plan和StreamGraph,都是Pipeline接口的具体实现,这两个实现分别用来表示批式和流式Job程序的拓扑,内部封装了用于构建生成JobGraph所需要的全部信息。
上面代码是构建生成JobGraph的主要逻辑,先是通过getExecutor()获取到一个PipelineExecutor,然后调用PipelineExecutor的execute()来构建并提交JobGraph。这里,PipelineExecutor表示执行Flink Job的方式,比如,本地执行使用LocalExecutor,或提交到YARN集群上执行使用YarnJobClusterExecutor,或提交到Kubernetes集群上执行使用KubernetesSessionClusterExecutor,等等。
无论是提交批式还是流式Job,最终都被转换成JobGraph对象,构建JobGraph的处理逻辑是完全统一的。构建JobGraph的代码逻辑,如下代码所示:
public static JobGraph getJobGraph( Pipeline pipeline, Configuration optimizerConfiguration, int defaultParallelism) { FlinkPipelineTranslator pipelineTranslator = getPipelineTranslator(pipeline); return pipelineTranslator.translateToJobGraph(pipeline, optimizerConfiguration, defaultParallelism); }
上面,输入的Pipeline对象,批式编程对应具体实现Plan,流式编程对应具体实现StreamGraph。由于用户编程API的不同,这里面选择了不同的FlinkPipelineTranslator来对输入的Pipeline对象进行翻译,其中批式对应的是PlanTranslator,流式对应的是StreamGraphTranslator,最终通过翻译,都生成一个统一的JobGraph。当然,调用translateToJobGraph()方法进行翻译处理的逻辑有很大的不同。
本文基于 署名-非商业性使用-相同方式共享 4.0 许可协议发布,欢迎转载、使用、重新发布,但务必保留文章署名时延军(包含链接:http://shiyanjun.cn),不得用于商业目的,基于本文修改后的作品务必以相同的许可发布。如有任何疑问,请与我联系。
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持 码农网
猜你喜欢:- java 并发编程-AQS源码分析
- Golang之面向接口编程及使用分析
- 并发编程(六)——AbstractQueuedSynchronizer 之 Condition 源码分析
- 并发编程(五)——AbstractQueuedSynchronizer 之 ReentrantLock源码分析
- 原 荐 Java异步编程——深入源码分析FutureTask
- 网络编程(一)TCP数据包的分析
本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们。