Apache Flink 进阶（六）：Flink 做业执行深度解析

时间  2019-11-19
标签 apache flink 进阶 执行 深度 解析 栏目 Apache 繁體版
原文   原文链接

做者：岳猛
整理：毛鹤数组

本文根据 Apache Flink 系列直播课程整理而成，由 Apache Flink Contributor、网易云音乐实时计算平台研发工程师岳猛分享。主要分享内容为 Flink Job 执行做业的流程，文章将从两个方面进行分享：一是如何从 Program 到物理执行计划，二是生成物理执行计划后该如何调度和执行。缓存

Flink 四层转化流程

Flink 有四层转换流程，第一层为 Program 到 StreamGraph；第二层为 StreamGraph 到 JobGraph；第三层为 JobGraph 到 ExecutionGraph；第四层为 ExecutionGraph 到物理执行计划。经过对 Program 的执行，可以生成一个 DAG 执行图，即逻辑执行图。以下：session

img1

第一部分将先讲解四层转化的流程，而后将以详细案例讲解四层的具体转化。架构

  • 第一层 StreamGraph 从 Source 节点开始，每一次 transform 生成一个 StreamNode，两个 StreamNode 经过 StreamEdge 链接在一块儿,造成 StreamNode 和 StreamEdge 构成的DAG。
  • 第二层 JobGraph，依旧从 Source 节点开始，而后去遍历寻找可以嵌到一块儿的 operator，若是可以嵌到一块儿则嵌到一块儿，不能嵌到一块儿的单独生成 jobVertex，经过 JobEdge 连接上下游 JobVertex，最终造成 JobVertex 层面的 DAG。
  • JobVertex DAG 提交到任务之后，从 Source 节点开始排序,根据 JobVertex 生成ExecutionJobVertex，根据 jobVertex的IntermediateDataSet 构建IntermediateResult，而后 IntermediateResult 构建上下游的依赖关系，造成 ExecutionJobVertex 层面的 DAG 即 ExecutionGraph。
  • 最后经过 ExecutionGraph 层到物理执行层。

Program 到 StreamGraph 的转化

Program 转换成 StreamGraph 具体分为三步：并发

  • 从 StreamExecutionEnvironment.execute 开始执行程序，将 transform 添加到 StreamExecutionEnvironment 的 transformations。
  • 调用 StreamGraphGenerator 的 generateInternal 方法，遍历 transformations 构建 StreamNode 及 StreamEage。
  • 经过 StreamEdge 链接 StreamNode。

img2

经过 WindowWordCount 来看代码到 StreamGraph 的转化，在 flatMap transform 设置 slot 共享组为 flatMap_sg，并发设置为 4，在聚合的操做中设置 slot 共享组为 sum_sg， sum() 和 counts() 并发设置为 3，这样设置主要是为了演示后面如何嵌到一块儿的，跟上下游节点的并发以及上游的共享组有关。app

WindowWordCount 代码中能够看到，在 readTextFile() 中会生成一个 transform，且 transform 的 ID 是 1；而后到 flatMap() 会生成一个 transform， transform 的 ID 是 2；接着到 keyBy() 生成一个 transform 的 ID 是 3；再到 sum() 生成一个 transform 的 ID 是 4；最后到 counts()生成 transform 的 ID 是 5。框架

img3

transform 的结构如图所示，第一个是 flatMap 的 transform，第二个是 window 的 transform，第三个是 SinkTransform 的 transform。除此以外，还能在 transform 的结构中看到每一个 transform 的 input 是什么。frontend

接下来介绍一下 StreamNode 和 StreamEdge。分布式

  • StreamNode 是用来描述 operator 的逻辑节点，其关键成员变量有 slotSharingGroup、jobVertexClass、inEdges、outEdges以及transformationUID；
  • StreamEdge 是用来描述两个 operator 逻辑的连接边，其关键变量有 sourceVertex、targetVertex。

img4

WindowWordCount transform 到 StreamGraph 转化如图所示，StreamExecutionEnvironment 的 transformations 存在 3 个 transform，分别是 Flat Map（Id 2）、Window（Id 4）、Sink（Id 5）。ide

transform 的时候首先递归处理 transform 的 input，生成 StreamNode，而后经过 StreamEdge 连接上下游 StreamNode。须要注意的是，有些 transform 操做并不会生成StreamNode 如 PartitionTransformtion，而是生成个虚拟节点。

img5

在转换完成后能够看到，streamNodes 有四种 transform 形式，分别为 Source、Flat Map、Window、Sink。

img6

每一个 streamNode 对象都携带并发个数、slotSharingGroup、执行类等运行信息。

StreamGraph 到 JobGraph 的转化

img7

StreamGraph 到 JobGraph 的转化步骤：

  • 设置调度模式，Eager 全部节点当即启动。
  • 广度优先遍历 StreamGraph，为每一个 streamNode 生成 byte 数组类型的 hash 值。
  • 从 source 节点开始递归寻找嵌到一块儿的 operator，不能嵌到一块儿的节点单独生成 jobVertex，可以嵌到一块儿的开始节点生成 jobVertex，其余节点以序列化的形式写入到 StreamConfig，而后 merge 到 CHAINED_TASK_CONFIG，再经过 JobEdge 连接上下游 JobVertex。
  • 将每一个 JobVertex 的入边(StreamEdge)序列化到该 StreamConfig。
  • 根据 group name 为每一个 JobVertext 指定 SlotSharingGroup。
  • 配置 checkpoint。
  • 将缓存文件存文件的配置添加到 configuration 中。
  • 设置 ExecutionConfig。

从 source 节点递归寻找嵌到一块儿的 operator 中，嵌到一块儿须要知足必定的条件，具体条件介绍以下：

  • 下游节点只有一个输入。
  • 下游节点的操做符不为 null。
  • 上游节点的操做符不为 null。
  • 上下游节点在一个槽位共享组内。
  • 下游节点的链接策略是 ALWAYS。
  • 上游节点的链接策略是 HEAD 或者 ALWAYS。
  • edge 的分区函数是 ForwardPartitioner 的实例。
  • 上下游节点的并行度相等。
  • 能够进行节点链接操做。

img8

JobGraph 对象结构如上图所示，taskVertices 中只存在 Window、Flat Map、Source 三个 TaskVertex，Sink operator 被嵌到 window operator 中去了。

为何要为每一个 operator 生成 hash 值？

Flink 任务失败的时候，各个 operator 是可以从 checkpoint 中恢复到失败以前的状态的，恢复的时候是依据 JobVertexID（hash 值)进行状态恢复的。相同的任务在恢复的时候要求 operator 的 hash 值不变，所以可以获取对应的状态。

每一个 operator 是怎样生成 hash 值的？

若是用户对节点指定了一个散列值，则基于用户指定的值可以产生一个长度为 16 的字节数组。若是用户没有指定，则根据当前节点所处的位置，产生一个散列值。

考虑的因素主要有三点：

  • 一是在当前 StreamNode 以前已经处理过的节点的个数，做为当前 StreamNode 的 id，添加到 hasher 中；
  • 二是遍历当前 StreamNode 输出的每一个 StreamEdge，并判断当前 StreamNode 与这个 StreamEdge 的目标 StreamNode 是否能够进行连接，若是能够，则将目标 StreamNode 的 id 也放入 hasher 中，且这个目标 StreamNode 的 id 与当前 StreamNode 的 id 取相同的值；
  • 三是将上述步骤后产生的字节数据，与当前 StreamNode 的全部输入 StreamNode 对应的字节数据，进行相应的位操做，最终获得的字节数据，就是当前 StreamNode 对应的长度为 16 的字节数组。

JobGraph 到 ExexcutionGraph 以及物理执行计划

img9

JobGraph 到 ExexcutionGraph 以及物理执行计划的流程：

  • 将 JobGraph 里面的 jobVertex 从 Source 节点开始排序。
  • 在 executionGraph.attachJobGraph(sortedTopology)方法里面，根据 JobVertex 生成 ExecutionJobVertex，在 ExecutionJobVertex 构造方法里面，根据 jobVertex 的 IntermediateDataSet 构建 IntermediateResult，根据 jobVertex 并发构建 ExecutionVertex，ExecutionVertex 构建的时候，构建 IntermediateResultPartition（每个 Execution 构建 IntermediateResult 数个IntermediateResultPartition ）；将建立的 ExecutionJobVertex 与前置的 IntermediateResult 链接起来。
  • 构建 ExecutionEdge ，链接到前面的 IntermediateResultPartition，最终从 ExecutionGraph 到物理执行计划。

Flink Job 执行流程

Flink On Yarn 模式

img10

基于 Yarn 层面的架构相似 Spark on Yarn 模式，都是由 Client 提交 App 到 RM 上面去运行，而后 RM 分配第一个 container 去运行 AM，而后由 AM 去负责资源的监督和管理。须要说明的是，Flink 的 Yarn 模式更加相似 Spark on Yarn 的 cluster 模式，在 cluster 模式中，dirver 将做为 AM 中的一个线程去运行。Flink on Yarn 模式也是会将 JobManager 启动在 container 里面，去作个 driver 相似的任务调度和分配，Yarn AM 与 Flink JobManager 在同一个 Container 中，这样 AM 能够知道 Flink JobManager 的地址，从而 AM 能够申请 Container 去启动 Flink TaskManager。待 Flink 成功运行在 Yarn 集群上，Flink Yarn Client 就能够提交 Flink Job 到 Flink JobManager，并进行后续的映射、调度和计算处理。

Fink on Yarn 的缺陷

  • 资源分配是静态的，一个做业须要在启动时获取所需的资源而且在它的生命周期里一直持有这些资源。这致使了做业不能随负载变化而动态调整，在负载降低时没法归还空闲的资源，在负载上升时也没法动态扩展。
  • On-Yarn 模式下，全部的 container 都是固定大小的，致使没法根据做业需求来调整 container 的结构。譬如 CPU 密集的做业或许须要更多的核，但不须要太多内存，固定结构的 container 会致使内存被浪费。
  • 与容器管理基础设施的交互比较笨拙，须要两个步骤来启动 Flink 做业: 1.启动 Flink 守护进程；2.提交做业。若是做业被容器化而且将做业部署做为容器部署的一部分，那么将再也不须要步骤2。
  • On-Yarn 模式下，做业管理页面会在做业完成后消失不可访问。
  • Flink 推荐 per job clusters 的部署方式，可是又支持能够在一个集群上运行多个做业的 session 模式，使人疑惑。

在 Flink 版本 1.5 中引入了 Dispatcher，Dispatcher 是在新设计里引入的一个新概念。Dispatcher 会从 Client 端接受做业提交请求并表明它在集群管理器上启动做业。

引入 Dispatcher 的缘由主要有两点:

  • 第一，一些集群管理器须要一个中心化的做业生成和监控实例；
  • 第二，可以实现 Standalone 模式下 JobManager 的角色，且等待做业提交。在一些案例中，Dispatcher 是可选的(Yarn)或者不兼容的(kubernetes)。

资源调度模型重构下的 Flink On Yarn 模式

img11

没有 Dispatcher job 运行过程

客户端提交 JobGraph 以及依赖 jar 包到 YarnResourceManager，接着 Yarn ResourceManager 分配第一个 container 以此来启动 AppMaster，Application Master 中会启动一个 FlinkResourceManager 以及 JobManager，JobManager 会根据 JobGraph 生成的 ExecutionGraph 以及物理执行计划向 FlinkResourceManager 申请 slot，FlinkResoourceManager 会管理这些 slot 以及请求，若是没有可用 slot 就向 Yarn 的 ResourceManager 申请 container，container 启动之后会注册到 FlinkResourceManager，最后 JobManager 会将 subTask deploy 到对应 container 的 slot 中去。

img12

在有 Dispatcher 的模式下

会增长一个过程，就是 Client 会直接经过 HTTP Server 的方式，而后用 Dispatcher 将这个任务提交到 Yarn ResourceManager 中。

新框架具备四大优点，详情以下：

  • client 直接在 Yarn 上启动做业，而不须要先启动一个集群而后再提交做业到集群。所以 client 再提交做业后能够立刻返回。
  • 全部的用户依赖库和配置文件都被直接放在应用的 classpath，而不是用动态的用户代码 classloader 去加载。
  • container 在须要时才请求，再也不使用时会被释放。
  • “须要时申请”的 container 分配方式容许不一样算子使用不一样 profile (CPU 和内存结构)的 container。

新的资源调度框架下 single cluster job on Yarn 流程介绍

img13

single cluster job on Yarn 模式涉及三个实例对象：

  • clifrontend

    • Invoke App code；
    • 生成 StreamGraph，而后转化为 JobGraph；

  • YarnJobClusterEntrypoint（Master）

    • 依次启动 YarnResourceManager、MinDispatcher、JobManagerRunner 三者都服从分布式协同一致的策略；
    • JobManagerRunner 将 JobGraph 转化为 ExecutionGraph ，而后转化为物理执行任务Execution，而后进行 deploy，deploy 过程会向 YarnResourceManager 请求 slot，若是有直接 deploy 到对应的 YarnTaskExecutiontor 的 slot 里面，没有则向 Yarn 的 ResourceManager 申请，带 container 启动之后 deploy。

  • YarnTaskExecutorRunner (slave)

    • 负责接收 subTask，并运行。

整个任务运行代码调用流程以下图：

img14

subTask 在执行时是怎么运行的？

调用 StreamTask 的 invoke 方法，执行步骤以下：
* initializeState()即operator的initializeState()
* openAllOperators() 即operator的open()方法
* 最后调用 run 方法来进行真正的任务处理

咱们来看下 flatMap 对应的 OneInputStreamTask 的 run 方法具体是怎么处理的。

@Override
    protected void run() throws Exception {
        // cache processor reference on the stack, to make the code more JIT friendly
        final StreamInputProcessor<IN> inputProcessor = this.inputProcessor;

        while (running && inputProcessor.processInput()) {
            // all the work happens in the "processInput" method
        }
    }

最终是调用 StreamInputProcessor 的 processInput() 作数据的处理，这里面包含用户的处理逻辑。

public boolean processInput() throws Exception {
        if (isFinished) {
            return false;
        }
        if (numRecordsIn == null) {
            try {
                numRecordsIn = ((OperatorMetricGroup) streamOperator.getMetricGroup()).getIOMetricGroup().getNumRecordsInCounter();
            } catch (Exception e) {
                LOG.warn("An exception occurred during the metrics setup.", e);
                numRecordsIn = new SimpleCounter();
            }
        }

        while (true) {
            if (currentRecordDeserializer != null) {
                DeserializationResult result = currentRecordDeserializer.getNextRecord(deserializationDelegate);

                if (result.isBufferConsumed()) {
                    currentRecordDeserializer.getCurrentBuffer().recycleBuffer();
                    currentRecordDeserializer = null;
                }

                if (result.isFullRecord()) {
                    StreamElement recordOrMark = deserializationDelegate.getInstance();
                    //处理watermark
                    if (recordOrMark.isWatermark()) {
                        // handle watermark
                        //watermark处理逻辑，这里可能引发timer的trigger
                        statusWatermarkValve.inputWatermark(recordOrMark.asWatermark(), currentChannel);
                        continue;
                    } else if (recordOrMark.isStreamStatus()) {
                        // handle stream status
                        statusWatermarkValve.inputStreamStatus(recordOrMark.asStreamStatus(), currentChannel);
                        continue;
                        //处理latency watermark
                    } else if (recordOrMark.isLatencyMarker()) {
                        // handle latency marker
                        synchronized (lock) {
                            streamOperator.processLatencyMarker(recordOrMark.asLatencyMarker());
                        }
                        continue;
                    } else {
                        //用户的真正的代码逻辑
                        // now we can do the actual processing
                        StreamRecord<IN> record = recordOrMark.asRecord();
                        synchronized (lock) {
                            numRecordsIn.inc();
                            streamOperator.setKeyContextElement1(record);
                            //处理数据
                            streamOperator.processElement(record);
                        }
                        return true;
                    }
                }
            }
            
            //这里会进行checkpoint barrier的判断和对齐，以及不一样partition 里面checkpoint barrier不一致时候的，数据buffer，

            final BufferOrEvent bufferOrEvent = barrierHandler.getNextNonBlocked();
            if (bufferOrEvent != null) {
                if (bufferOrEvent.isBuffer()) {
                    currentChannel = bufferOrEvent.getChannelIndex();
                    currentRecordDeserializer = recordDeserializers[currentChannel];
                    currentRecordDeserializer.setNextBuffer(bufferOrEvent.getBuffer());
                }
                else {
                    // Event received
                    final AbstractEvent event = bufferOrEvent.getEvent();
                    if (event.getClass() != EndOfPartitionEvent.class) {
                        throw new IOException("Unexpected event: " + event);
                    }
                }
            }
            else {
                isFinished = true;
                if (!barrierHandler.isEmpty()) {
                    throw new IllegalStateException("Trailing data in checkpoint barrier handler.");
                }
                return false;
            }
        }
    }

streamOperator.processElement(record) 最终会调用用户的代码处理逻辑，假如 operator 是 StreamFlatMap 的话，

@Override
    public void processElement(StreamRecord<IN> element) throws Exception {
        collector.setTimestamp(element);
        userFunction.flatMap(element.getValue(), collector);//用户代码
    }

 

 

 

原文连接

本文为云栖社区原创内容，未经容许不得转载。

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程