Flink原理与实现:如何生成ExecutionGraph及物理执行图

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介:

阅读本文之前,请先阅读Flink原理与实现系列前面的几篇文章 :

Flink 原理与实现:架构和拓扑概览
Flink 原理与实现:如何生成 StreamGraph
Flink 原理与实现:如何生成 JobGraph

ExecutionGraph生成过程

StreamGraph和JobGraph都是在client生成的,这篇文章将描述如何生成ExecutionGraph以及物理执行图。同时会讲解一个作业提交后如何被调度和执行。

client生成JobGraph之后,就通过submitJob提交至JobMaster。
在其构造函数中,会生成ExecutionGraph:

    this.executionGraph = ExecutionGraphBuilder.buildGraph(...)

看下这个方法,比较长,略过了一些次要的代码片断:


     // 流式作业中,schedule mode固定是EAGER的
        executionGraph.setScheduleMode(jobGraph.getScheduleMode());
        executionGraph.setQueuedSchedulingAllowed(jobGraph.getAllowQueuedScheduling());

     // 设置json plan
     // ...

     // 检查executableClass(即operator类),设置最大并发
     // ...
     
        // 按拓扑顺序,获取所有的JobVertex列表
        List<JobVertex> sortedTopology = jobGraph.getVerticesSortedTopologicallyFromSources();
        
        // 根据JobVertex列表,生成execution graph
        executionGraph.attachJobGraph(sortedTopology);
        
        // checkpoint检查

可以看到,生成execution graph的代码,主要是在最后一行,即ExecutionGraph.attachJobGraph方法:

    public void attachJobGraph(List<JobVertex> topologiallySorted) throws JobException, IOException {
       // 遍历job vertex
        for (JobVertex jobVertex : topologiallySorted) {
            // 根据每一个job vertex,创建对应的ExecutionVertex
            ExecutionJobVertex ejv = new ExecutionJobVertex(this, jobVertex, 1, rpcCallTimeout, createTimestamp);
            // 将创建的ExecutionJobVertex与前置的IntermediateResult连接起来
            ejv.connectToPredecessors(this.intermediateResults);

            ExecutionJobVertex previousTask = this.tasks.putIfAbsent(jobVertex.getID(), ejv);

        // sanity check
        // ...
        
            this.verticesInCreationOrder.add(ejv);
        }
    }

可以看到,创建ExecutionJobVertex的重点就在它的构造函数中:

     // 上面是并行度相关的设置
     
     // 序列化后的TaskInformation,这个信息很重要
     // 后面deploy的时候会将TaskInformation分发到具体的Task中。
        this.serializedTaskInformation = new SerializedValue<>(new TaskInformation(
            jobVertex.getID(),
            jobVertex.getName(),
            parallelism,
            maxParallelism,
            // 这个就是Task将要执行的Operator的类名
            jobVertex.getInvokableClassName(),
            jobVertex.getConfiguration()));
     
     // ExecutionVertex列表,按照JobVertex并行度设置      
        this.taskVertices = new ExecutionVertex[numTaskVertices];
        
        this.inputs = new ArrayList<>(jobVertex.getInputs().size());
        
        // slot sharing和coLocation相关代码
        // ...
        
        // 创建intermediate results,这是由当前operator的出度确定的,如果当前operator只向下游一个operator输出,则为1
        // 注意一个IntermediateResult包含多个IntermediateResultPartition
        this.producedDataSets = new IntermediateResult[jobVertex.getNumberOfProducedIntermediateDataSets()];

        for (int i = 0; i < jobVertex.getProducedDataSets().size(); i++) {
            final IntermediateDataSet result = jobVertex.getProducedDataSets().get(i);

            this.producedDataSets[i] = new IntermediateResult(
                    result.getId(),
                    this,
                    numTaskVertices,
                    result.getResultType());
        }

        // 根据job vertex的并行度,创建对应的ExecutionVertex列表。
        // 即,一个JobVertex/ExecutionJobVertex代表的是一个operator,而
        // 具体的ExecutionVertex则代表了每一个Task
        for (int i = 0; i < numTaskVertices; i++) {
            ExecutionVertex vertex = new ExecutionVertex(
                    this, i, this.producedDataSets, timeout, createTimestamp, maxPriorAttemptsHistoryLength);

            this.taskVertices[i] = vertex;
        }
        
        // sanity check
        // ...
        
        // set up the input splits, if the vertex has any
        // 这是batch相关的代码
        // ...
             
        finishedSubtasks = new boolean[parallelism];

ExecutionJobVertex和ExecutionVertex是创建完了,但是ExecutionEdge还没有创建呢,接下来看一下attachJobGraph方法中这一行代码:

    ejv.connectToPredecessors(this.intermediateResults);

这个方法代码如下:

     // 获取输入的JobEdge列表
        List<JobEdge> inputs = jobVertex.getInputs();
                
        // 遍历每条JobEdge        
        for (int num = 0; num < inputs.size(); num++) {
            JobEdge edge = inputs.get(num);
            
            // 获取当前JobEdge的输入所对应的IntermediateResult
            IntermediateResult ires = intermediateDataSets.get(edge.getSourceId());
            if (ires == null) {
                throw new JobException("Cannot connect this job graph to the previous graph. No previous intermediate result found for ID "
                        + edge.getSourceId());
            }
            
            // 将IntermediateResult加入到当前ExecutionJobVertex的输入中。
            this.inputs.add(ires);
            
            // 为IntermediateResult注册consumer
            // consumerIndex跟IntermediateResult的出度相关
            int consumerIndex = ires.registerConsumer();
            
            for (int i = 0; i < parallelism; i++) {
                ExecutionVertex ev = taskVertices[i];
                // 将ExecutionVertex与IntermediateResult关联起来
                ev.connectSource(num, ires, edge, consumerIndex);
            }
        }

看下ExecutionVertex.connectSource方法代码:

    public void connectSource(int inputNumber, IntermediateResult source, JobEdge edge, int consumerNumber) {

     // 只有forward的方式的情况下,pattern才是POINTWISE的,否则均为ALL_TO_ALL
        final DistributionPattern pattern = edge.getDistributionPattern();
        final IntermediateResultPartition[] sourcePartitions = source.getPartitions();

        ExecutionEdge[] edges;

        switch (pattern) {
            case POINTWISE:
                edges = connectPointwise(sourcePartitions, inputNumber);
                break;

            case ALL_TO_ALL:
                edges = connectAllToAll(sourcePartitions, inputNumber);
                break;

            default:
                throw new RuntimeException("Unrecognized distribution pattern.");

        }

        this.inputEdges[inputNumber] = edges;

        // 之前已经为IntermediateResult添加了consumer,这里为IntermediateResultPartition添加consumer,即关联到ExecutionEdge上
        for (ExecutionEdge ee : edges) {
            ee.getSource().addConsumer(ee, consumerNumber);
        }
    }

connectAllToAll方法:

        ExecutionEdge[] edges = new ExecutionEdge[sourcePartitions.length];

        for (int i = 0; i < sourcePartitions.length; i++) {
            IntermediateResultPartition irp = sourcePartitions[i];
            edges[i] = new ExecutionEdge(irp, this, inputNumber);
        }

        return edges;

看这个方法之前,需要知道,ExecutionVertex的inputEdges变量,是一个二维数据。它表示了这个ExecutionVertex上每一个input所包含的ExecutionEdge列表。

即,如果ExecutionVertex有两个不同的输入:输入A和B。其中输入A的partition=1, 输入B的partition=8,那么这个二维数组inputEdges如下(为简短,以irp代替IntermediateResultPartition)

[ ExecutionEdge[ A.irp[0]] ]
[ ExecutionEdge[ B.irp[0], B.irp[1], ..., B.irp[7] ]

所以上面的代码就很容易理解了。

到这里为止,ExecutionJobGraph就创建完成了。接下来看下这个ExecutionGraph是如何转化成Task并开始执行的。


Task调度和执行

接下来我们以最简单的mini cluster为例讲解一下Task如何被调度和执行。

简单略过client端job的提交和StreamGraph到JobGraph的翻译,以及上面ExecutionGraph的翻译。

提交后的job的流通过程大致如下:

env.execute('<job name>')
  --> MiniCluster.runJobBlocking(jobGraph)
  --> MiniClusterDispatcher.runJobBlocking(jobGraph)
  --> MiniClusterDispatcher.startJobRunners
    --> JobManagerRunner.start
    --> JobMaster.<init> (build ExecutionGraph)

创建完JobMaster之后,JobMaster就会进行leader election,得到leader之后,会回调grantLeadership方法,从而调用jobManager.start(leaderSessionID);开始运行job。

JobMaster.start 
    --> JobMaster.startJobExecution(这里还没开始执行呢..)
    --> resourceManagerLeaderRetriever.start(new ResourceManagerLeaderListener());    

重点是在下面这行,获取到resource manage之后,就会回调ResourceManagerLeaderListener.notifyLeaderAddress,整个调用流如下:

ResourceManagerLeaderListener.notifyLeaderAddress
    --> JobMaster.notifyOfNewResourceManagerLeader
    --> ResourceManagerConnection.start
    --> ResourceManagerConnection.onRegistrationSuccess(callback,由flink rpc框架发送并回调)
    --> JobMaster.onResourceManagerRegistrationSuccess

然后终于来到了最核心的调度代码,在JobMaster.onResourceManagerRegistrationSuccess方法中:

    executionContext.execute(new Runnable() {
        @Override
        public void run() {
            try {
                executionGraph.restoreExternalCheckpointedStore();
                executionGraph.setQueuedSchedulingAllowed(true);
                executionGraph.scheduleForExecution(slotPool.getSlotProvider());
            }
            catch (Throwable t) {
                executionGraph.fail(t);
            }
        }
    });

ExecutionGraph.scheduleForExecution --> ExecutionGraph.scheduleEager

这个方法会计算所有的ExecutionVertex总数,并为每个ExecutionVertex分配一个SimpleSlot(暂时不考虑slot sharing的情况),然后封装成ExecutionAndSlot,顾名思义,即ExecutionVertex + Slot(更为贴切地说,应该是ExecutionAttempt + Slot)。

然后调用execAndSlot.executionAttempt.deployToSlot(slot);进行deploy,即Execution.deployToSlot

这个方法先会进行一系列状态迁移和检查,然后进行deploy,比较核心的代码如下:

        final TaskDeploymentDescriptor deployment = vertex.createDeploymentDescriptor(
            attemptId,
            slot,
            taskState,
            attemptNumber);

        // register this execution at the execution graph, to receive call backs
        vertex.getExecutionGraph().registerExecution(this);
            
        final TaskManagerGateway taskManagerGateway = slot.getTaskManagerGateway();        final Future<Acknowledge> submitResultFuture = taskManagerGateway.submitTask(deployment, timeout);

ExecutionVertex.createDeploymentDescriptor方法中,包含了从Execution Graph到真正物理执行图的转换。如将IntermediateResultPartition转化成ResultPartition,ExecutionEdge转成InputChannelDeploymentDescriptor(最终会在执行时转化成InputGate)。

最后通过RPC方法提交task,实际会调用到TaskExecutor.submitTask方法中。
这个方法会创建真正的Task,然后调用task.startTaskThread();开始task的执行。

在Task构造函数中,会根据输入的参数,创建InputGate, ResultPartition, ResultPartitionWriter等。

startTaskThread方法,则会执行executingThread.start,从而调用Task.run方法。
它的最核心的代码如下:

     // ...
     
        // now load the task's invokable code
        invokable = loadAndInstantiateInvokable(userCodeClassLoader, nameOfInvokableClass);

      // ...
      invokable.setEnvironment(env);
      
      // ...
      this.invokable = invokable;
      invokable.invoke();
      
      // task finishes or fails, do cleanup
      // ...

这里的invokable即为operator对象实例,通过反射创建。具体地,即为OneInputStreamTask,或者SourceStreamTask等。这个nameOfInvokableClass是哪里生成的呢?其实早在生成StreamGraph的时候,这就已经确定了,见StreamGraph.addOperator方法:

        if (operatorObject instanceof StoppableStreamSource) {
            addNode(vertexID, slotSharingGroup, StoppableSourceStreamTask.class, operatorObject, operatorName);
        } else if (operatorObject instanceof StreamSource) {
            addNode(vertexID, slotSharingGroup, SourceStreamTask.class, operatorObject, operatorName);
        } else {
            addNode(vertexID, slotSharingGroup, OneInputStreamTask.class, operatorObject, operatorName);
        }

这里的OneInputStreamTask.class即为生成的StreamNode的vertexClass。这个值会一直传递,当StreamGraph被转化成JobGraph的时候,这个值会被传递到JobVertex的invokableClass。然后当JobGraph被转成ExecutionGraph的时候,这个值被传入到ExecutionJobVertex.TaskInformation.invokableClassName中,一直传到Task中。

那么用户真正写的逻辑代码在哪里呢?比如word count中的Tokenizer,去了哪里呢?
OneInputStreamTask的基类StreamTask,包含了headOperator和operatorChain。当我们调用dataStream.flatMap(new Tokenizer())的时候,会生成一个StreamFlatMap的operator,这个operator是一个AbstractUdfStreamOperator,而用户的代码new Tokenizer,即为它的userFunction。

所以再串回来,以OneInputStreamTask为例,Task的核心执行代码即为OneInputStreamTask.invoke方法,它会调用StreamTask.run方法,这是个抽象方法,最终会调用其派生类的run方法,即OneInputStreamTask, SourceStreamTask等。

OneInputStreamTask的run方法代码如下:

    final OneInputStreamOperator<IN, OUT> operator = this.headOperator;
    final StreamInputProcessor<IN> inputProcessor = this.inputProcessor;
    final Object lock = getCheckpointLock();
        
    while (running && inputProcessor.processInput(operator, lock)) {
        // all the work happens in the "processInput" method
    }

就是一直不停地循环调用inputProcessor.processInput(operator, lock)方法,即StreamInputProcessor.processInput方法:

    public boolean processInput(OneInputStreamOperator<IN, ?> streamOperator, final Object lock) throws Exception {
     // ...
     
        while (true) {
            if (currentRecordDeserializer != null) {
           // ...
           
                if (result.isFullRecord()) {
                    StreamElement recordOrMark = deserializationDelegate.getInstance();
                    
              // 处理watermark,则框架处理
                    if (recordOrMark.isWatermark()) {
                       // watermark处理逻辑
                       // ...
                        continue;
                    } else if(recordOrMark.isLatencyMarker()) {
                        // 处理latency mark,也是由框架处理
                        synchronized (lock) {
                            streamOperator.processLatencyMarker(recordOrMark.asLatencyMarker());
                        }
                        continue;
                    } else {
                        // ***** 这里是真正的用户逻辑代码 *****
                        StreamRecord<IN> record = recordOrMark.asRecord();
                        synchronized (lock) {
                            numRecordsIn.inc();
                            streamOperator.setKeyContextElement1(record);
                            streamOperator.processElement(record);
                        }
                        return true;
                    }
                }
            }

        // 其他处理逻辑
        // ...
        }
    }

上面的代码中,streamOperator.processElement(record);才是真正处理用户逻辑的代码,以StreamFlatMap为例,即为它的processElement方法:

    public void processElement(StreamRecord<IN> element) throws Exception {
        collector.setTimestamp(element);
        userFunction.flatMap(element.getValue(), collector);
    }

这样,整个调度和执行逻辑就全部串起来啦。

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
相关文章
|
10天前
|
Java 流计算
【极数系列】Flink搭建入门项目Demo & 秒懂Flink开发运行原理(05)
【极数系列】Flink搭建入门项目Demo & 秒懂Flink开发运行原理(05)
29 3
|
3月前
|
消息中间件 缓存 监控
Flink背压原理以及解决优化
Flink背压原理以及解决优化
85 0
|
3月前
|
SQL 算法 API
读Flink源码谈设计:图的抽象与分层
前阵子组里的小伙伴问我“为什么Flink从我们的代码到真正可执行的状态,要经过这么多个graph转换?这样做有什么好处嘛?”我早期看到这里的设计时的确有过相同的疑惑,当时由于手里还在看别的东西,查阅过一些资料后就翻页了。如今又碰到了这样的问题,不妨就在这篇文章中好好搞清楚。
530 0
读Flink源码谈设计:图的抽象与分层
|
5月前
|
程序员 API 数据安全/隐私保护
Flink--8、时间语义、水位线(事件和窗口、水位线和窗口的工作原理、生产水位线、水位线的传递、迟到数据的处理)
Flink--8、时间语义、水位线(事件和窗口、水位线和窗口的工作原理、生产水位线、水位线的传递、迟到数据的处理)
|
5天前
|
SQL Java 数据库连接
Flink内存问题之超出物理内存如何解决
Apache Flink是由Apache软件基金会开发的开源流处理框架,其核心是用Java和Scala编写的分布式流数据流引擎。本合集提供有关Apache Flink相关技术、使用技巧和最佳实践的资源。
|
26天前
|
SQL 并行计算 大数据
【大数据技术攻关专题】「Apache-Flink零基础入门」手把手+零基础带你玩转大数据流式处理引擎Flink(基础加强+运行原理)
关于Flink服务的搭建与部署,由于其涉及诸多实战操作而理论部分相对较少,小编打算采用一个独立的版本和环境来进行详尽的实战讲解。考虑到文字描述可能无法充分展现操作的细节和流程,我们决定以视频的形式进行分析和介绍。因此,在本文中,我们将暂时不涉及具体的搭建和部署步骤。
370 3
【大数据技术攻关专题】「Apache-Flink零基础入门」手把手+零基础带你玩转大数据流式处理引擎Flink(基础加强+运行原理)
|
2月前
|
存储 NoSQL MongoDB
阿里云 Flink 原理分析与应用:深入探索 MongoDB Schema Inference
本文整理自阿里云 Flink 团队归源老师关于阿里云 Flink 原理分析与应用:深入探索 MongoDB Schema Inference 的研究。
46859 2
阿里云 Flink 原理分析与应用:深入探索 MongoDB Schema Inference
|
2月前
|
Java 数据处理 分布式数据库
Flink中的Exactly-Once语义是什么?请解释其作用和实现原理。
Flink中的Exactly-Once语义是什么?请解释其作用和实现原理。
12 0
|
3月前
|
消息中间件 Kafka API
2021年最新最全Flink系列教程_Flink原理初探和流批一体API(二.五)
2021年最新最全Flink系列教程_Flink原理初探和流批一体API(二.五)
35 0
|
3月前
|
消息中间件 关系型数据库 MySQL
2021年最新最全Flink系列教程_Flink原理初探和流批一体API(二)
2021年最新最全Flink系列教程_Flink原理初探和流批一体API(二)
120 0

相关产品

  • 云迁移中心