Flink 编程 API 设计分析

使用Flink开发批式或流式Job,除了基本的处理逻辑与实际应用场景相关,我们更关心的是Flink提供的基本框架,是如何在API层面进行统一处理的,或者说尽量使API统一,这样有助于我们对Flink框架更深入地理解。目前使用Flink 1.10版本开发批式和流式Job,在API层面来看,大部分还是比较统一的,但是由于批式和流式场景还是有一定的差异,想要完全统一还是有一定难度。
Flink数据流编程模型的分层设计,如下图所示:
Flink-Layerd-Abstraction
这里,我们关注的是上面的Core APIs层,结合批式和流式Job开发与提交的过程进行分析,对比对于批式和流式场景下,如何生成最终的JobGraph,以及生成的过程有什么不同。

编程API设计

Flink编程API中,我们开发的数据处理Job程序,都是始于一个Source,对应的输入数据源,终于一个Sink,对应输出的存储系统,中间是各种丰富的处理算子。但是对于批式和流式编程API,从代码层面对应的抽象基本上是名称不同,具体逻辑功能比较一致:批式编程API对批式Job DAG中每个节点的抽象使用的是DataSet,而流式编程API中对应的是DataStream。
对于批式Job DAG中,DataSet的类设计体系,如下图所示:
CD.DataSet
相关的类都在包org.apache.flink.api.java.operators下面,通过上图可以看出,主要分为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类设计体系,如下图所示:
CD.DataStream
DataStream表示在流式Job DAG中每一步转换操作之前与之后,都对应着一个DataStream的数据结构,它内部封装了与转换操作相关的处理逻辑,其实就是StreamOperator。对应上图中,我们举几个编写流式处理程序的例子说明:调用StreamExecutionEnvironment.readTextFile()时会生成一个DataStreamSource,调用keyBy()时会生成一个KeyedStream,调用split()时会生成一个SplitStream,调用iterate()时会生成一个IterativeStream。
下面看下StreamOperator类的设计体系,如下图所示:
CD.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对象,我们抽象出如下流程概念图:
Flink-Job-Submission
上图中,左侧是批式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()方法进行翻译处理的逻辑有很大的不同。

Creative Commons License

本文基于署名-非商业性使用-相同方式共享 4.0许可协议发布,欢迎转载、使用、重新发布,但务必保留文章署名时延军(包含链接:http://shiyanjun.cn),不得用于商业目的,基于本文修改后的作品务必以相同的许可发布。如有任何疑问,请与我联系

发表评论

电子邮件地址不会被公开。 必填项已用*标注

您可以使用这些HTML标签和属性: <a href="" title=""> <abbr title=""> <acronym title=""> <b> <blockquote cite=""> <cite> <code> <del datetime=""> <em> <i> <q cite=""> <strike> <strong>