Flink: 两个递归彻底搞懂operator chain

朴灿烈づ我的快乐病毒、 2023-07-10 08:13 107阅读 0赞

operator chain是指将满足一定条件的operator 链在一起，放在同一个task里面执行，是Flink任务优化的一种方式，在同一个task里面的operator的数据传输变成函数调用关系，这种方式减少数据传输过程。常见的chain例如：source->map->filter，这样的任务链可以chain在一起，那么其内部是如何决定是否能够chain在一起与chain一起之后如何执行就是本篇文章将要剖析的重点。

### 第一个递归：JobGraph生成 ###

Flink中划分了四种图：StreamGraph、JobGraph、ExecutionGraph、物理执行图，前两种StreamGraph、JobGraph是在客户端生成，ExecutionGraph在jobMaster中生成，最后一种物理执行图是一种虚拟的图，不存在的数据结构，运行在每一个TaskExecutor中。我们在Flink Web UI中看到的就是JobGraph，如下：

![format_png][]

JobGraph相对于StreamGraph，可以理解为优化过后的StreamGraph，将能够chain一起的operator chain在一起，上图将source与filter两个operator chain在一起了，这个步骤在生成JobGraph过程中完成。其具体实现在StreamingJobGraphGenerator中：

private JobGraph createJobGraph() {
            .....
            Map<Integer, List<Tuple2<byte[], byte[]>>> chainedOperatorHashes = new HashMap<>();
            setChaining(hashes, legacyHashes, chainedOperatorHashes);
            setPhysicalEdges();
            setSlotSharingAndCoLocation();
            configureCheckpointing();
         ....
        }

重点就在setChaining方法中，在里面调用createChain方法，构造JobVertix的同时完成operator chain的操作，createChain方法：

private List<StreamEdge> createChain(
                Integer startNodeId,
                Integer currentNodeId,
                Map<Integer, byte[]> hashes,
                List<Map<Integer, byte[]>> legacyHashes,
                int chainIndex,
                Map<Integer, List<Tuple2<byte[], byte[]>>> chainedOperatorHashes) {
    
    
          if (!builtVertices.contains(startNodeId)) {
       //chain 的出边
              List<StreamEdge> transitiveOutEdges = new ArrayList<StreamEdge>();
       //能够chain在一起的边
              List<StreamEdge> chainableOutputs = new ArrayList<StreamEdge>();
        //不能够chain一起的边
              List<StreamEdge> nonChainableOutputs = new ArrayList<StreamEdge>();
    
    
              for (StreamEdge outEdge : streamGraph.getStreamNode(currentNodeId).getOutEdges()) {
          //isChainable判断是否能够chain在一起
                  if (isChainable(outEdge, streamGraph)) {
                      chainableOutputs.add(outEdge);
                  } else {
                      nonChainableOutputs.add(outEdge);
                  }
              }
    
    
              for (StreamEdge chainable : chainableOutputs) {
          //能够chain在一起那么遍历下一个节点
                  transitiveOutEdges.addAll(
                          createChain(startNodeId, chainable.getTargetId(), hashes, legacyHashes, chainIndex + 1, chainedOperatorHashes));
              }
    
    
              for (StreamEdge nonChainable : nonChainableOutputs) {
                  transitiveOutEdges.add(nonChainable); 
          //以不能chain在一起的节点为起始点重新开始往下遍历
                  createChain(nonChainable.getTargetId(), nonChainable.getTargetId(), hashes, legacyHashes, 0, chainedOperatorHashes);
              }
    
    
                List<Tuple2<byte[], byte[]>> operatorHashes =
                    chainedOperatorHashes.computeIfAbsent(startNodeId, k -> new ArrayList<>());
    
    
                byte[] primaryHashBytes = hashes.get(currentNodeId);
    
    
                for (Map<Integer, byte[]> legacyHash : legacyHashes) {
                    operatorHashes.add(new Tuple2<>(primaryHashBytes, legacyHash.get(currentNodeId)));
                }
    
    
                chainedNames.put(currentNodeId, createChainedName(currentNodeId, chainableOutputs));
                chainedMinResources.put(currentNodeId, createChainedMinResources(currentNodeId, chainableOutputs));
                chainedPreferredResources.put(currentNodeId, createChainedPreferredResources(currentNodeId, chainableOutputs));
    
    
          //如果currentNodeId=startNodeId 那么就说明是一个chain的起点，则需要创建jobVertix
         //不是则表示是chain的一部分，只需要创建StreamConfig即可
            StreamConfig config = currentNodeId.equals(startNodeId)
                    ? createJobVertex(startNodeId, hashes, legacyHashes, chainedOperatorHashes)
                    : new StreamConfig(new Configuration());
    
    
            setVertexConfig(currentNodeId, config, chainableOutputs, nonChainableOutputs);
    
    
            if (currentNodeId.equals(startNodeId)) {
                config.setChainStart(); //起始chain
                config.setChainIndex(0);
                config.setOperatorName(streamGraph.getStreamNode(currentNodeId).getOperatorName());
                config.setOutEdgesInOrder(transitiveOutEdges);
                config.setOutEdges(streamGraph.getStreamNode(currentNodeId).getOutEdges());
        //连接边
                for (StreamEdge edge : transitiveOutEdges) {
                    connect(startNodeId, edge);
                }
    
    
                config.setTransitiveChainedTaskConfigs(chainedConfigs.get(startNodeId));
    
    
            } else {
        //currentNodeId属于chain的一部分 
                Map<Integer, StreamConfig> chainedConfs = chainedConfigs.get(startNodeId);
    
    
           if (chainedConfs == null) {
                        chainedConfigs.put(startNodeId, new HashMap<Integer, StreamConfig>());
            }
              config.setChainIndex(chainIndex);
              StreamNode node = streamGraph.getStreamNode(currentNodeId);
              config.setOperatorName(node.getOperatorName());
              chainedConfigs.get(startNodeId).put(currentNodeId, config);
          }
    
    
            config.setOperatorID(new OperatorID(primaryHashBytes));
    
    
            if (chainableOutputs.isEmpty()) {
                config.setChainEnd();
            }
      //返回chain的出边
            return transitiveOutEdges;
    
    
        } else {
            return new ArrayList<>();
        }
        }

整个过程就是一个递归的过程，createChain 过程就是不断寻找一个chain的出边，如果邻接的两个operator(source与filter)能够chain在一起，那么就以下一个能够chain一起的operator(filter)为起点，继续寻找，直到找到不能够chain一起的operator(process1)，但是此时并没有立刻返回，而是以当前不能chain再一起的operator(process1)为起点继续往下寻找，直到终点(sink）开始一层一层返回，实际上其构造过程是一个反向过程：sink->process2->process1->(source&filter) 这样的一个过程完成operator chain并且构造JobVertix (可通过debug方式查看其详细过程)。

如何判断两个相邻的operator(StreamNode)能够chain在一起？通过isChainable方法判断：

public static boolean isChainable(StreamEdge edge, StreamGraph streamGraph) {
            StreamNode upStreamVertex = streamGraph.getSourceVertex(edge);
            StreamNode downStreamVertex = streamGraph.getTargetVertex(edge);
    
    
            StreamOperator<?> headOperator = upStreamVertex.getOperator();
            StreamOperator<?> outOperator = downStreamVertex.getOperator();
    
    
            return downStreamVertex.getInEdges().size() == 1 //下游的数据流入只有一个节点
              && outOperator != null
              && headOperator != null
              && upStreamVertex.isSameSlotSharingGroup(downStreamVertex) //在同一个slotGroup中
      && outOperator.getChainingStrategy() == ChainingStrategy.ALWAYS //开启operator chain策略
              && (headOperator.getChainingStrategy() == ChainingStrategy.HEAD ||
          headOperator.getChainingStrategy() == ChainingStrategy.ALWAYS) //head 表示的是一个起点
              && (edge.getPartitioner() instanceof ForwardPartitioner) //直接转发方式
              && upStreamVertex.getParallelism() == downStreamVertex.getParallelism() //并行度相同
              && streamGraph.isChainingEnabled(); //默认允许chain
        }

### 第二个递归：OperatorChain生成 ###

当Execution在deploy的过程中，也就是Task在TaskExecutor启动过程中， 会生成一个OperatorChain对象，在该OperatorChain对象中包含了所有的能够chain在一起的operator(source&filter)，其内部会生成一个名为chainEntryPoint的WatermarkGaugeExposingOutput对象，一个将数据输出的对象，其输出有两种形式：

1.函数调用，将数据推送给chain在一起的下一个operator节点(filter中)                2.输出到下一个没有被chain的operator(process1)

那么chainEntryPoint是如何生成的？

OperatorChain的初始化是在StreamTask中被调用的：

operatorChain = new OperatorChain<>(this, recordWriters);
    headOperator = operatorChain.getHeadOperator();

在其构造函数中调用：

this.chainEntryPoint = createOutputCollector(
                    containingTask, //当前的streamTask
                    configuration, //chain的第一个节点的StreamConfig
                    chainedConfigs, //该chain的所有StreamConfig
                    userCodeClassloader,
                    streamOutputMap,
                    allOps);

createOutputCollector获取当前节点的out，

private <T> WatermarkGaugeExposingOutput<StreamRecord<T>> createOutputCollector(
                StreamTask<?, ?> containingTask,
                StreamConfig operatorConfig,
                Map<Integer, StreamConfig> chainedConfigs,
                ClassLoader userCodeClassloader,
                Map<StreamEdge, RecordWriterOutput<?>> streamOutputs,
                List<StreamOperator<?>> allOperators) {
            List<Tuple2<WatermarkGaugeExposingOutput<StreamRecord<T>>, StreamEdge>> allOutputs = new ArrayList<>(4);
    
    
            // 当前operator的网络方式输出(filter->process1)
          for (StreamEdge outputEdge : operatorConfig.getNonChainedOutputs(userCodeClassloader)) {
              @SuppressWarnings("unchecked")
              RecordWriterOutput<T> output = (RecordWriterOutput<T>) streamOutputs.get(outputEdge);
    
    
              allOutputs.add(new Tuple2<>(output, outputEdge));
          }
    
    
            // chain out (source->filter)
            for (StreamEdge outputEdge : operatorConfig.getChainedOutputs(userCodeClassloader)) {
                int outputId = outputEdge.getTargetId();
                StreamConfig chainedOpConfig = chainedConfigs.get(outputId);
         //寻找被chain在一起的下一个operator(filter)的out
            WatermarkGaugeExposingOutput<StreamRecord<T>> output = createChainedOperator(
                containingTask,
                chainedOpConfig,
                chainedConfigs,
                userCodeClassloader,
                streamOutputs,
                allOperators,
                outputEdge.getOutputTag());
            allOutputs.add(new Tuple2<>(output, outputEdge));
            }
    
    
      //最后将当前节点的out返回
        .....
    
    
            }
        }

createChainedOperator方法：

private <IN, OUT> WatermarkGaugeExposingOutput<StreamRecord<IN>> createChainedOperator(
                StreamTask<?, ?> containingTask,
                StreamConfig operatorConfig,
                Map<Integer, StreamConfig> chainedConfigs,
                ClassLoader userCodeClassloader,
                Map<StreamEdge, RecordWriterOutput<?>> streamOutputs,
                List<StreamOperator<?>> allOperators,
                OutputTag<IN> outputTag) {
            // 调用createOutputCollector 获取当前operator的out
            WatermarkGaugeExposingOutput<StreamRecord<OUT>> chainedOperatorOutput = createOutputCollector(
                containingTask,
                operatorConfig,
                chainedConfigs,
                userCodeClassloader,
                streamOutputs,
                allOperators);
    
    
            // 获取当前的StreamOperator
            OneInputStreamOperator<IN, OUT> chainedOperator = operatorConfig.getStreamOperator(userCodeClassloader);
       //作为当前的out 传入setup方法中
            chainedOperator.setup(containingTask, operatorConfig, chainedOperatorOutput);
    
    
            allOperators.add(chainedOperator);
    
    
      //将被chain的operator(filter)传给上一个operator(source)的out
      //那么在out中就可以直接调用filter的处理source的输出数据了
            WatermarkGaugeExposingOutput<StreamRecord<IN>> currentOperatorOutput;
            if (containingTask.getExecutionConfig().isObjectReuseEnabled()) {
                currentOperatorOutput = new ChainingOutput<>(chainedOperator, this, outputTag);
            }
            else {
                TypeSerializer<IN> inSerializer = operatorConfig.getTypeSerializerIn1(userCodeClassloader);
                currentOperatorOutput = new CopyingChainingOutput<>(chainedOperator, inSerializer, outputTag, this);
            }
        ....
            return currentOperatorOutput;
        }

可以看出整个构造chainEntryPoint的过程也是一个递归的过程，会不断寻找被chain在一起的下一个operator的out，直到下游没有可chain的位置，返回网络out作为最后一个operator的out，然后通过使用当前operator构造前一个operator的out，同样是一个反向构造out的过程(filterOut->sourceOut)。简化一下逻辑代码：

createOut(currentOperator){
    
    
        if(currentOperator.isNetworkOut){
            currOut=networkOut();
        }else{
            currOut=creatOut(nextOperstor);
        }
        currentOperator.setOut(currOut);
        preOut=CopyingOut(currentOperator);
        return preOut;
    }

最终得到的chainEntryPoint就是headOperator的out，这样在其内部不断的通过out调用operator的方式实现了chain的函数调用链关系。

### 总结 ###

透过以上分析，operator chain就是将满足一定条件的的operator通过函数调用方式传递数据，避免了数据传输的中间过程。

![format_png 1][]

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6X3BuZy9TN3RJdm81eFlNNUt5bFRhc29SajJaQ3M1cEZGSVQ5NFl5Z2ZIbG9pYjRpYW1yQlpTNXc4bzFFUzFxUzl3ZXBaTVg2VkwzUWtoQWYyWlppY25OeWtYdnlLUS82NDA?x-oss-process=image/format,png
[format_png 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6X3BuZy9TN3RJdm81eFlNNTdzV2VYeTJpY0xTdHY4MUNRU1ZINUFKa0J4aWFPWWRPRWlhaWNMM1R1akJnSWJHM0ZZaWNoZ0pXaWF2QWhtRXI3OG1BS1hvb3h3dkxUZVNCQS82NDA?x-oss-process=image/format,png