flink JobGraph解析

水军大提督  论坛元老 | 2025-2-14 17:08:30 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 1010|帖子 1010|积分 3030

JobGraph组成


JobGraph主要是StreamGraph经过优化后天生的，主要优化的就是对符合条件节点举行chain，这样可以淘汰数据流动的序列化和传输。
JobGraph主要由三部分组成。


  • JobVertex：图的顶点。输入是一个JobEdge，输出是IntermediateDataSet。它可以对应多个StreamNode，将多个operator归并到一起。
  • IntermediateDataSet：中间效果集。是JobVertex处置惩罚后天生的效果集，为了方便卑鄙复用，producer 是 JobVertex ，consumer 是 JobEdge。
  • JobEdge：边。JobGraph的传输管道。source 是 IntermediateDataSet，target 是 JobVertex。即数据通过 JobEdge 由 IntermediateDataSet 传递给目标 JobVertex 。
JobVertex



  • operatorIDs：该 job 节点包罗的全部 operator ids，以深度优先方式存储 ids
  • results：job 节点盘算出的中间效果
  • inputs：输入数据的边列表

IntermediateDataSet



  • producer：生产者，JobVertex
  • consumers：消费边，可以对应多个，但是必须具有相同的分区器和并行性
  • resultType：运行时使用的分区类型

    • BLOCKING 阻塞，批处置惩罚模式
    • PIPELINED 管道非阻塞，流处置惩罚模式


JobEdge



  • target：edge的输出，JobVertex
  • source：edge的源，IntermediateDataSet
  • distributionPattern：决定了在上游节点（生产者）的子任务和卑鄙节点（消费者）之间的连接模式

    • ALL_TO_ALL：每个生产子任务都连接到消费任务的每个子任务
    • POINTWISE：每个生产子任务都连接到使用任务的一个或多个子任务


JobGraph天生

入口是在StreamingJobGraphGenerator的createJobGraph方法

createJobGraph过程比较多，重点是三步：

  • 为各个StreamNode天生hash值，这样在故障规复的时候可以识别
  • 天生JobVertex
  • 天生JobEdge、IntermediateDataSet

天生JobVertex（setChaining）

从 Source StreamNode 实例开始设置 task chain，它将会递归地创建全部的 JobVertex 实例。
buildChainedInputsAndGetHeadInputs会得到chain的出发点集合，然后遍历举行createChain

buildChainedInputsAndGetHeadInputs
  1. private Map<Integer, OperatorChainInfo> buildChainedInputsAndGetHeadInputs(
  2.         final Map<Integer, byte[]> hashes, final List<Map<Integer, byte[]>> legacyHashes) {
  4.     // 可以chain的source，单独处理这种节点
  5.     final Map<Integer, ChainedSourceInfo> chainedSources = new HashMap<>();
  6.     // chain的起点(不能chain的souce节点、可以chain的souce节点的下一个节点)
  7.     final Map<Integer, OperatorChainInfo> chainEntryPoints = new HashMap<>();
  9.     // 遍历streamGraph的所有source node
  10.     for (Integer sourceNodeId : streamGraph.getSourceIDs()) {
  11.         final StreamNode sourceNode = streamGraph.getStreamNode(sourceNodeId);
  13.         if (sourceNode.getOperatorFactory() instanceof SourceOperatorFactory
  14.                 && sourceNode.getOutEdges().size() == 1) {
  15.             // 要求source node的outEdge只有一个。有多个出边的source不能chain
  16.             // as long as only NAry ops support this chaining, we need to skip the other parts
  17.             final StreamEdge sourceOutEdge = sourceNode.getOutEdges().get(0);
  18.             final StreamNode target = streamGraph.getStreamNode(sourceOutEdge.getTargetId());
  19.             final ChainingStrategy targetChainingStrategy =
  20.                     target.getOperatorFactory().getChainingStrategy();
  22.             if (targetChainingStrategy == ChainingStrategy.HEAD_WITH_SOURCES
  23.                     && isChainableInput(sourceOutEdge, streamGraph)) {
  24.                 final OperatorID opId = new OperatorID(hashes.get(sourceNodeId));
  25.                 final StreamConfig.SourceInputConfig inputConfig =
  26.                         new StreamConfig.SourceInputConfig(sourceOutEdge);
  27.                 final StreamConfig operatorConfig = new StreamConfig(new Configuration());
  28.                 setOperatorConfig(sourceNodeId, operatorConfig, Collections.emptyMap());
  29.                 setOperatorChainedOutputsConfig(operatorConfig, Collections.emptyList());
  30.                 // we cache the non-chainable outputs here, and set the non-chained config later
  31.                 opNonChainableOutputsCache.put(sourceNodeId, Collections.emptyList());
  33.                 // sources的index都是0
  34.                 operatorConfig.setChainIndex(0); // sources are always first
  35.                 operatorConfig.setOperatorID(opId);
  36.                 operatorConfig.setOperatorName(sourceNode.getOperatorName());
  37.                 chainedSources.put(
  38.                         sourceNodeId, new ChainedSourceInfo(operatorConfig, inputConfig));
  40.                 final SourceOperatorFactory<?> sourceOpFact =
  41.                         (SourceOperatorFactory<?>) sourceNode.getOperatorFactory();
  42.                 final OperatorCoordinator.Provider coord =
  43.                         sourceOpFact.getCoordinatorProvider(sourceNode.getOperatorName(), opId);
  45.                 // chainEntryPoints中添加(targetNodeId, chainInfo)
  46.                 final OperatorChainInfo chainInfo =
  47.                         chainEntryPoints.computeIfAbsent(
  48.                                 sourceOutEdge.getTargetId(),
  49.                                 (k) ->
  50.                                         new OperatorChainInfo(
  51.                                                 sourceOutEdge.getTargetId(),
  52.                                                 hashes,
  53.                                                 legacyHashes,
  54.                                                 chainedSources,
  55.                                                 streamGraph));
  56.                 chainInfo.addCoordinatorProvider(coord);
  57.                 chainInfo.recordChainedNode(sourceNodeId);
  58.                 continue;
  59.             }
  60.         }
  62.         chainEntryPoints.put(
  63.                 sourceNodeId,
  64.                 new OperatorChainInfo(
  65.                         sourceNodeId, hashes, legacyHashes, chainedSources, streamGraph));
  66.     }
  68.     return chainEntryPoints;
  69. }
复制代码

createChain
在创建chain的过程中，一个chain完成后，在头结点创建一个JobVertex。
  1. private List<StreamEdge> createChain(
  2.         final Integer currentNodeId,
  3.         final int chainIndex,
  4.         final OperatorChainInfo chainInfo,
  5.         final Map<Integer, OperatorChainInfo> chainEntryPoints) {
  7.     Integer startNodeId = chainInfo.getStartNodeId();
  8.     if (!builtVertices.contains(startNodeId)) {
  9.         // transitiveOutEdges 过渡的出边集合，就是两个StreamNode不能再进行chain的那条边，用于生成JobEdge
  10.         List<StreamEdge> transitiveOutEdges = new ArrayList<StreamEdge>();
  12.         // chainableOutputs 两个StreamNode可以进行chain的出边集合
  13.         // nonChainableOutputs 两个StreamNode不能进行chain的出边
  14.         List<StreamEdge> chainableOutputs = new ArrayList<StreamEdge>();
  15.         List<StreamEdge> nonChainableOutputs = new ArrayList<StreamEdge>();
  17.         StreamNode currentNode = streamGraph.getStreamNode(currentNodeId);
  19.         for (StreamEdge outEdge : currentNode.getOutEdges()) {
  20.             if (isChainable(outEdge, streamGraph)) {
  21.                 chainableOutputs.add(outEdge);
  22.             } else {
  23.                 nonChainableOutputs.add(outEdge);
  24.             }
  25.         }
  27.         for (StreamEdge chainable : chainableOutputs) {
  28.             // 如果存在可以chain的边，那么就继续往这条边的target operator进行chain。
  29.             // transitiveOutEdges最终返回给首次调用栈的是不能再继续chain的那条边
  30.             transitiveOutEdges.addAll(
  31.                     createChain(
  32.                             chainable.getTargetId(),
  33.                             chainIndex + 1,
  34.                             chainInfo,
  35.                             chainEntryPoints));
  36.         }
  38.         for (StreamEdge nonChainable : nonChainableOutputs) {
  39.             //如果存在了不可chain的边，说明该边就是StreamNode chain之间的过渡边，添加到transitiveOutEdges中，
  40.             //继续对该边的target StreamNode进行新的createChain操作，意味着一个新的chain
  41.             transitiveOutEdges.add(nonChainable);
  42.             createChain(
  43.                     nonChainable.getTargetId(),
  44.                     1, // operators start at position 1 because 0 is for chained source inputs
  45.                     chainEntryPoints.computeIfAbsent(
  46.                             nonChainable.getTargetId(),
  47.                             (k) -> chainInfo.newChain(nonChainable.getTargetId())),
  48.                     chainEntryPoints);
  49.         }
  51.         chainedNames.put(
  52.                 currentNodeId,
  53.                 createChainedName(
  54.                         currentNodeId,
  55.                         chainableOutputs,
  56.                         Optional.ofNullable(chainEntryPoints.get(currentNodeId))));
  57.         chainedMinResources.put(
  58.                 currentNodeId, createChainedMinResources(currentNodeId, chainableOutputs));
  59.         chainedPreferredResources.put(
  60.                 currentNodeId,
  61.                 createChainedPreferredResources(currentNodeId, chainableOutputs));
  63.         // 添加当前的StreamNode到chain中
  64.         OperatorID currentOperatorId =
  65.                 chainInfo.addNodeToChain(
  66.                         currentNodeId,
  67.                         streamGraph.getStreamNode(currentNodeId).getOperatorName());
  69.         if (currentNode.getInputFormat() != null) {
  70.             getOrCreateFormatContainer(startNodeId)
  71.                     .addInputFormat(currentOperatorId, currentNode.getInputFormat());
  72.         }
  74.         if (currentNode.getOutputFormat() != null) {
  75.             getOrCreateFormatContainer(startNodeId)
  76.                     .addOutputFormat(currentOperatorId, currentNode.getOutputFormat());
  77.         }
  79.         // chain的头结点创建JobVertex
  80.         StreamConfig config =
  81.                 currentNodeId.equals(startNodeId)
  82.                         ? createJobVertex(startNodeId, chainInfo)
  83.                         : new StreamConfig(new Configuration());
  85.         tryConvertPartitionerForDynamicGraph(chainableOutputs, nonChainableOutputs);
  87.         setOperatorConfig(currentNodeId, config, chainInfo.getChainedSources());
  89.         setOperatorChainedOutputsConfig(config, chainableOutputs);
  91.         // we cache the non-chainable outputs here, and set the non-chained config later
  92.         // 缓存不能chain的出边集合
  93.         opNonChainableOutputsCache.put(currentNodeId, nonChainableOutputs);
  95.         if (currentNodeId.equals(startNodeId)) {
  96.             // 头结点
  97.             chainInfo.setTransitiveOutEdges(transitiveOutEdges);
  98.             chainInfos.put(startNodeId, chainInfo);
  100.             config.setChainStart();
  101.             config.setChainIndex(chainIndex);
  102.             config.setOperatorName(streamGraph.getStreamNode(currentNodeId).getOperatorName());
  103.             config.setTransitiveChainedTaskConfigs(chainedConfigs.get(startNodeId));
  105.         } else {
  106.             chainedConfigs.computeIfAbsent(
  107.                     startNodeId, k -> new HashMap<Integer, StreamConfig>());
  109.             config.setChainIndex(chainIndex);
  110.             StreamNode node = streamGraph.getStreamNode(currentNodeId);
  111.             config.setOperatorName(node.getOperatorName());
  112.             chainedConfigs.get(startNodeId).put(currentNodeId, config);
  113.         }
  115.         config.setOperatorID(currentOperatorId);
  117.         if (chainableOutputs.isEmpty()) {
  118.             // chain尾节点
  119.             config.setChainEnd();
  120.         }
  121.         return transitiveOutEdges;
  123.     } else {
  124.         return new ArrayList<>();
  125.     }
  126. }
复制代码

判定是否chainable


  • 公用一个slotGroup
  • 上卑鄙operator可以chain
  • partitioner和exchangeMode可以chain（forward）
  • 并行度一样
  • 允许chain
  • 不能是联合操作

createJobVertex

  • 创建对应的operator集合
  • 创建JobVertex（InputOutputFormatVertex是一种特殊的 JobVertex，它用于处置惩罚输入输出格式相干的任务，比方读取和写入文件、数据库等）
  • 添加对应的上游数据集
  • 缓存JobVertex相干信息


天生JobEdge、IntermediateDataSet（setAllVertexNonChainedOutputsConfigs）

遍历jobVertices，调用connect连接起来。


connect
将两个JobVertex（headVertex、downStreamVertex）连接起来。关键方法是downStreamVertex.connectNewDataSetAsInput

connectNewDataSetAsInput
创建IntermediateDataSet和JobEdge，形成JobVertex->IntermediateDataSet->JobEdge->JobVertex


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

您需要 登录 才可以下载或查看，没有账号？立即注册

x
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

水军大提督

论坛元老
这个人很懒什么都没写!

楼主热帖

标签云

AI 运维 CIO 存储 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表