【万字长文】详解Flink作业提交流程（二）

private void addEdgeInternal(Integer upStreamVertexID,
   Integer downStreamVertexID,
   int typeNumber,
   StreamPartitioner<?> partitioner,
   List<String> outputNames,
   OutputTag outputTag,
   ShuffleMode shuffleMode) {
  //当上游是sideoutput时，递归调用，并传入sideoutput信息
  if (virtualSideOutputNodes.containsKey(upStreamVertexID)) {
   int virtualId = upStreamVertexID;
   upStreamVertexID = virtualSideOutputNodes.get(virtualId).f0;
   if (outputTag == null) {
    outputTag = virtualSideOutputNodes.get(virtualId).f1;
   }
   addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, null, outputTag, shuffleMode);
  }
  //当上游是select时，递归调用，并传入select信息
  else if (virtualSelectNodes.containsKey(upStreamVertexID)) {
   int virtualId = upStreamVertexID;
   upStreamVertexID = virtualSelectNodes.get(virtualId).f0;
   if (outputNames.isEmpty()) {
    // selections that happen downstream override earlier selections
    outputNames = virtualSelectNodes.get(virtualId).f1;
   }
   addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, outputNames, outputTag, shuffleMode);
  }
  //当上游是Partition时，递归调用，并传入Partition信息
  else if (virtualPartitionNodes.containsKey(upStreamVertexID)) {
   int virtualId = upStreamVertexID;
   upStreamVertexID = virtualPartitionNodes.get(virtualId).f0;
   if (partitioner == null) {
    partitioner = virtualPartitionNodes.get(virtualId).f1;
   }
   shuffleMode = virtualPartitionNodes.get(virtualId).f2;
   addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, outputNames, outputTag, shuffleMode);
  }
  //不是以上逻辑转换的情况，真正构建StreamEdge
  else {
   StreamNode upstreamNode = getStreamNode(upStreamVertexID);
   StreamNode downstreamNode = getStreamNode(downStreamVertexID);
   // If no partitioner was specified and the parallelism of upstream and downstream
   // operator matches use forward partitioning, use rebalance otherwise.
   //没有指定partitioner时，会为其选择forward或者rebalance
   if (partitioner == null && upstreamNode.getParallelism() == downstreamNode.getParallelism()) {
    partitioner = new ForwardPartitioner<Object>();
   } else if (partitioner == null) {
    partitioner = new RebalancePartitioner<Object>();
   }
   if (partitioner instanceof ForwardPartitioner) {
    if (upstreamNode.getParallelism() != downstreamNode.getParallelism()) {
     throw new UnsupportedOperationException("Forward partitioning does not allow " +
       "change of parallelism. Upstream operation: " + upstreamNode + " parallelism: " + upstreamNode.getParallelism() +
       ", downstream operation: " + downstreamNode + " parallelism: " + downstreamNode.getParallelism() +
       " You must use another partitioning strategy, such as broadcast, rebalance, shuffle or global.");
    }
   }
   if (shuffleMode == null) {
    shuffleMode = ShuffleMode.UNDEFINED;
   }
   //创建StreamEdge,并将该SteramEdge添加到上游的输出，下游的输入。
   StreamEdge edge = new StreamEdge(upstreamNode, downstreamNode, typeNumber, outputNames, partitioner, outputTag, shuffleMode);
   getStreamNode(edge.getSourceId()).addOutEdge(edge);
   getStreamNode(edge.getTargetId()).addInEdge(edge);
  }
 }

private JobGraph createJobGraph() {
     preValidate();
     // make sure that all vertices start immediately
     //设置调度模式
     jobGraph.setScheduleMode(streamGraph.getScheduleMode());
     // Generate deterministic hashes for the nodes in order to identify them across
     // submission iff they didn't change.
     //为每个节点生成确定的hashid作为唯一表示，在提交和执行过程中保持不变。
     Map<Integer, byte[]> hashes = defaultStreamGraphHasher.traverseStreamGraphAndGenerateHashes(streamGraph);
     // Generate legacy version hashes for backwards compatibility
     //为了向后保持兼容，为每个节点生成老版本的hash id
     List<Map<Integer, byte[]>> legacyHashes = new ArrayList<>(legacyStreamGraphHashers.size());
     for (StreamGraphHasher hasher : legacyStreamGraphHashers) {
      legacyHashes.add(hasher.traverseStreamGraphAndGenerateHashes(streamGraph));
     }
     Map<Integer, List<Tuple2<byte[], byte[]>>> chainedOperatorHashes = new HashMap<>();
     //真正对SteramGraph进行转换，生成JobGraph图
     setChaining(hashes, legacyHashes, chainedOperatorHashes);
     setPhysicalEdges();
     //设置共享slotgroup
     setSlotSharingAndCoLocation();
     setManagedMemoryFraction(
      Collections.unmodifiableMap(jobVertices),
      Collections.unmodifiableMap(vertexConfigs),
      Collections.unmodifiableMap(chainedConfigs),
      id -> streamGraph.getStreamNode(id).getMinResources(),
      id -> streamGraph.getStreamNode(id).getManagedMemoryWeight());
     //配置checkpoint
     configureCheckpointing();
     jobGraph.setSavepointRestoreSettings(streamGraph.getSavepointRestoreSettings());
     //如果有之前的缓存文件的配置，则重新读入
     JobGraphGenerator.addUserArtifactEntries(streamGraph.getUserArtifacts(), jobGraph);
     // set the ExecutionConfig last when it has been finalized
     try {
      //设置执行环境配置
      jobGraph.setExecutionConfig(streamGraph.getExecutionConfig());
     }
     catch (IOException e) {
      throw new IllegalConfigurationException("Could not serialize the ExecutionConfig." +
        "This indicates that non-serializable types (like custom serializers) were registered");
     }
     return jobGraph;
    }

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)) {
      List<StreamEdge> transitiveOutEdges = new ArrayList<StreamEdge>();
      List<StreamEdge> chainableOutputs = new ArrayList<StreamEdge>();
      List<StreamEdge> nonChainableOutputs = new ArrayList<StreamEdge>();
      StreamNode currentNode = streamGraph.getStreamNode(currentNodeId);
      //获取当前节点的出边，判断是否符合OperatorChain的条件
      //分为两类：chainableoutputs,nonchainableoutputs
      for (StreamEdge outEdge : currentNode.getOutEdges()) {
       if (isChainable(outEdge, streamGraph)) {
        chainableOutputs.add(outEdge);
       } else {
        nonChainableOutputs.add(outEdge);
       }
      }
      //对于chainable的边，递归调用createchain
      //返回值添加到transitiveOutEdges中
      for (StreamEdge chainable : chainableOutputs) {
       transitiveOutEdges.addAll(
         createChain(startNodeId, chainable.getTargetId(), hashes, legacyHashes, chainIndex + 1, chainedOperatorHashes));
      }
      //对于无法chain在一起的边，边的下游节点作为Operatorchain的Head节点
      //进行递归调用，返回值添加到transitiveOutEdges中
      for (StreamEdge nonChainable : nonChainableOutputs) {
       transitiveOutEdges.add(nonChainable);
       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);
      OperatorID currentOperatorId = new OperatorID(primaryHashBytes);
      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));
      if (currentNode.getInputFormat() != null) {
       getOrCreateFormatContainer(startNodeId).addInputFormat(currentOperatorId, currentNode.getInputFormat());
      }
      if (currentNode.getOutputFormat() != null) {
       getOrCreateFormatContainer(startNodeId).addOutputFormat(currentOperatorId, currentNode.getOutputFormat());
      }
      //如果当前节点是起始节点，则直接创建JobVertex，否则返回一个空的StreamConfig
      StreamConfig config = currentNodeId.equals(startNodeId)
        ? createJobVertex(startNodeId, hashes, legacyHashes, chainedOperatorHashes)
        : new StreamConfig(new Configuration());
      //将StreamNode中的配置信息序列化到Streamconfig中。
      setVertexConfig(currentNodeId, config, chainableOutputs, nonChainableOutputs);
      //再次判断，如果是Chain的起始节点，执行connect（）方法，创建JobEdge和IntermediateDataset
      //否则将当前节点的StreamConfig 添加到chainedConfig中。
      if (currentNodeId.equals(startNodeId)) {
       config.setChainStart();
       config.setChainIndex(0);
       config.setOperatorName(streamGraph.getStreamNode(currentNodeId).getOperatorName());
       for (StreamEdge edge : transitiveOutEdges) {
        connect(startNodeId, edge);
       }
       config.setOutEdgesInOrder(transitiveOutEdges);
       config.setTransitiveChainedTaskConfigs(chainedConfigs.get(startNodeId));
      } else {
       chainedConfigs.computeIfAbsent(startNodeId, k -> new HashMap<Integer, StreamConfig>());
       config.setChainIndex(chainIndex);
       StreamNode node = streamGraph.getStreamNode(currentNodeId);
       config.setOperatorName(node.getOperatorName());
       chainedConfigs.get(startNodeId).put(currentNodeId, config);
      }
      config.setOperatorID(currentOperatorId);
      if (chainableOutputs.isEmpty()) {
       config.setChainEnd();
      }
      return transitiveOutEdges;
     } else {
      return new ArrayList<>();
     }
    }