乐趣区

关于flink:FlinkOperatorID生成逻辑及Chain策略

在 StreamGraph 翻译为 JobGraph 的过程中 Flink 会为每一个算子生成对应的 OperatorID,并传递到 Jobvertex 中。JobVertex 是 JobGraph 中的节点,每个 JobVertex 蕴含一个或多个算子 chain 在一起的算子链。如果 JobVertex 只蕴含一个算子,则 JobVertex 的 id 就是这个算子的 OperatorID,如果 JobVertex 蕴含了多个算子 chain 在一起的算子链,则 JobVertex 的 id 是这个算子链的头部算子的 OperatorID。每个 OperatorID 惟一标识一个算子,Flink 状态复原时也是通过 OperatorID 找到以后节点对应的状态。

入口函数

之前提到,OperatorID 是在 StreamGraph 翻译为 JobGraph 的过程中生成的,其入口函数为 StreamingJobGraphGenerator#createJobGraph:

// Generate deterministic hashes for the nodes in order to identify them across
// submission iff they didn't change.
Map<Integer, byte[]> hashes =
  defaultStreamGraphHasher.traverseStreamGraphAndGenerateHashes(streamGraph);

// Generate legacy version hashes for backwards compatibility
List<Map<Integer, byte[]>> legacyHashes = new ArrayList<>(legacyStreamGraphHashers.size());
for (StreamGraphHasher hasher : legacyStreamGraphHashers) {legacyHashes.add(hasher.traverseStreamGraphAndGenerateHashes(streamGraph));
}
  • defaultStreamGraphHasher:默认实现为 StreamGraphHasherV2,用于计算每个节点的 OperatorID,哈希的对象依据 StreamNode 是否设置了 transformationUID 会有变动。
  • legacyHashes:只蕴含一个 StreamGraphUserHashHasher,如果用户给算子设置了 userHash,则这里会抽取用户设置的 userHash 作为 OperatorID。

StreamGraphHasherV2

  1. 找出所有的 source 算子,增加到 remaining 队列;
  2. 对 remaining 队列采取广度遍历算法,计算每个节点的 OperatorID。
public Map<Integer, byte[]> traverseStreamGraphAndGenerateHashes(StreamGraph streamGraph) {
    // The hash function used to generate the hash
    final HashFunction hashFunction = Hashing.murmur3_128(0);
    final Map<Integer, byte[]> hashes = new HashMap<>();

    Set<Integer> visited = new HashSet<>();
    Queue<StreamNode> remaining = new ArrayDeque<>();

    // We need to make the source order deterministic. The source IDs are
    // not returned in the same order, which means that submitting the same
    // program twice might result in different traversal, which breaks the
    // deterministic hash assignment.
    List<Integer> sources = new ArrayList<>();
    for (Integer sourceNodeId : streamGraph.getSourceIDs()) {sources.add(sourceNodeId);
    }
    Collections.sort(sources);

    //
    // Traverse the graph in a breadth-first manner. Keep in mind that
    // the graph is not a tree and multiple paths to nodes can exist.
    //

    // 将 source 节点放入队列
    // Start with source nodes
    for (Integer sourceNodeId : sources) {remaining.add(streamGraph.getStreamNode(sourceNodeId));
        visited.add(sourceNodeId);
    }

    // 广度遍历
    StreamNode currentNode;
    while ((currentNode = remaining.poll()) != null) {
        // Generate the hash code. Because multiple path exist to each
        // node, we might not have all required inputs available to
        // generate the hash code.
        // 如果生成失败,阐明该节点依赖的节点的哈希尚未计算结束,则把该节点从 visited 拿出,期待下一次遍历
        // 如果生成胜利,则把该节点的上游节点放入待遍历的队列和 visited 队列,放入 visited 队列的起因是
        if (generateNodeHash(
                currentNode,
                hashFunction,
                hashes,
                streamGraph.isChainingEnabled(),
                streamGraph)) {
            // Add the child nodes
            for (StreamEdge outEdge : currentNode.getOutEdges()) {StreamNode child = streamGraph.getTargetVertex(outEdge);

                if (!visited.contains(child.getId())) {remaining.add(child);
                    visited.add(child.getId());
                }
            }
        } else {
            // We will revisit this later.
            visited.remove(currentNode.getId());
        }
    }

    return hashes;
}

generateNodeHash 办法执行逻辑如下图所示:

依据用户是否需给 StreamNode 设置了 transformationUID 会将不同的数据作为哈希对象:

  • generateUserSpecifiedHash,将用户设置的 transformationUID 作为源数据计算哈希:

    private byte[] generateUserSpecifiedHash(StreamNode node, Hasher hasher) {hasher.putString(node.getTransformationUID(), Charset.forName("UTF-8"));
    
      return hasher.hash().asBytes();
    }
  • generateDeterministicHash,依据作业的拓扑构造计算 OperatorID:

    private byte[] generateDeterministicHash(
      StreamNode node,
      Hasher hasher,
      Map<Integer, byte[]> hashes,
      boolean isChainingEnabled,
      StreamGraph streamGraph) {
    
      // Include stream node to hash. We use the current size of the computed
      // hashes as the ID. We cannot use the node's ID, because it is
      // assigned from a static counter. This will result in two identical
      // programs having different hashes.
      generateNodeLocalHash(hasher, hashes.size());
    
      // Include chained nodes to hash
      for (StreamEdge outEdge : node.getOutEdges()) {if (isChainable(outEdge, isChainingEnabled, streamGraph)) {
    
          // Use the hash size again, because the nodes are chained to
          // this node. This does not add a hash for the chained nodes.
          generateNodeLocalHash(hasher, hashes.size());
        }
      }
    
      byte[] hash = hasher.hash().asBytes();
    
      // Make sure that all input nodes have their hash set before entering
      // this loop (calling this method).
      for (StreamEdge inEdge : node.getInEdges()) {byte[] otherHash = hashes.get(inEdge.getSourceId());
    
        // Sanity check
        if (otherHash == null) {
          throw new IllegalStateException(
            "Missing hash for input node"
            + streamGraph.getSourceVertex(inEdge)
            + ". Cannot generate hash for"
            + node
            + ".");
        }
    
        for (int j = 0; j < hash.length; j++) {hash[j] = (byte) (hash[j] * 37 ^ otherHash[j]);
        }
      }
    
      // ... debug log
      return hash;
    }
  1. 在哈希源数据 buffer 中放入 hashes.size();
  2. 查看该算子 chain 的上游算子数量,每有一个就往 buffer 中放一次 hashes.size();
  3. 对 buffer 计算哈希失去该算子的哈希值;
  4. 找出该算子的上游算子的 hashes,与该算子的哈希值做位操作。

Chain 策略

判断两个算子能 chain 在一起的条件如下:

  • 用户全局容许启用 chain:默认处于开启状态,可通过 StreamExecutionEnvironment#disableOperatorChaining 禁用;
  • 上游算子只有以后算子一个上游;
  • 两个算子同属一个 slotSharingGroup;
  • 算子的 chain 策略不能是 NEVER:默认是 ALWAYS,一些 transformations 可通过 setChainingStrategy 批改;
  • 算子之间的数据转发应用 ForwardPartitioner;
  • ShuffleMode 不能是 BATCH;
  • 上下游算子并发度统一;
  • 该 StreamGraph 的 chaining 配置为 true:默认为 true,用户可通过 StreamGraph::setChaining 批改。

总结

OperatorID 的生成逻辑能够简要概括如下:

  1. 如果用户在创立 DataStream 时设置了 userHash,则应用该 userHash 作为 OperatorID;
  2. 如果用户在创立 DataStream 时设置了 transformationUID,则将 transformationUID 进行一次哈希计算的后果作为 OperatorID;
  3. 默认状况下,依据以后算子的地位以及和上游算子 chain 的状况计算哈希值,并和上游算子的哈希值做位操作后取得 OperatorID。
退出移动版