devbox
你好赵伟
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

flink streamGraph生成_flink streamgraph select 虚拟节点

作者：你好赵伟 | 2024-07-07 00:37:51

赞

踩

flink streamgraph select 虚拟节点

当完成DataStream的配置后，调用其中环境上下文StreamExecutionEnvironment的getStreamGrahph()方法即可生成关于该DataStream的streamGraph。

  1. @Internal
  2. public StreamGraph getStreamGraph() {
  3.    if (transformations.size() <= 0) {
  4.       throw new IllegalStateException("No operators defined in streaming topology. Cannot execute.");
  5.    }
  6.    return StreamGraphGenerator.generate(this, transformations);
  7. }
  8.  
  9. public static StreamGraph generate(StreamExecutionEnvironment env, List<StreamTransformation<?>> transformations) {
  10.    return new StreamGraphGenerator(env).generateInternal(transformations);
  11. }

生成streamGraph实则是在StreamGraphGenerator的generate()方法中，参数除了调用的env本身之外还需要env当中表示被转换的DataStream中所有操作streamTransformation的数组。

之后会在generate()方法中会生成一个StreamGraphGenerator并调用generateInternal()方法根据DataStream中的所有transformation开始生成streamGraph。

  1. private StreamGraph generateInternal(List<StreamTransformation<?>> transformations) {
  2.    for (StreamTransformation<?> transformation: transformations) {
  3.       transform(transformation);
  4.    }
  5.    return streamGraph;
  6. }

此处会遍历所有的streamTransformation，依次调用transform()方法来转换至streamGraph。

 

在transform()方法中，会根据所有类型的transform分别处理，具体的代码如下。

  1. if (transform instanceof OneInputTransformation<?, ?>) {
  2.    transformedIds = transformOneInputTransform((OneInputTransformation<?, ?>) transform);
  3. } else if (transform instanceof TwoInputTransformation<?, ?, ?>) {
  4.    transformedIds = transformTwoInputTransform((TwoInputTransformation<?, ?, ?>) transform);
  5. } else if (transform instanceof SourceTransformation<?>) {
  6.    transformedIds = transformSource((SourceTransformation<?>) transform);
  7. } else if (transform instanceof SinkTransformation<?>) {
  8.    transformedIds = transformSink((SinkTransformation<?>) transform);
  9. } else if (transform instanceof UnionTransformation<?>) {
  10.    transformedIds = transformUnion((UnionTransformation<?>) transform);
  11. } else if (transform instanceof SplitTransformation<?>) {
  12.    transformedIds = transformSplit((SplitTransformation<?>) transform);
  13. } else if (transform instanceof SelectTransformation<?>) {
  14.    transformedIds = transformSelect((SelectTransformation<?>) transform);
  15. } else if (transform instanceof FeedbackTransformation<?>) {
  16.    transformedIds = transformFeedback((FeedbackTransformation<?>) transform);
  17. } else if (transform instanceof CoFeedbackTransformation<?>) {
  18.    transformedIds = transformCoFeedback((CoFeedbackTransformation<?>) transform);
  19. } else if (transform instanceof PartitionTransformation<?>) {
  20.    transformedIds = transformPartition((PartitionTransformation<?>) transform);
  21. } else if (transform instanceof SideOutputTransformation<?>) {
  22.    transformedIds = transformSideOutput((SideOutputTransformation<?>) transform);
  23. } else {
  24.    throw new IllegalStateException("Unknown transformation: " + transform);
  25. }

以一个map操作节点为例子，如果是map类型的transformation，那么将会在此处通过transformOneInputTransform()方法构造成streamGraph中的streamNode。

  1. private <IN, OUT> Collection<Integer> transformOneInputTransform(OneInputTransformation<IN, OUT> transform) {
  2.  
  3.    Collection<Integer> inputIds = transform(transform.getInput());
  4.  
  5.    // the recursive call might have already transformed this
  6.    if (alreadyTransformed.containsKey(transform)) {
  7.       return alreadyTransformed.get(transform);
  8.    }
  9.  
  10.    String slotSharingGroup = determineSlotSharingGroup(transform.getSlotSharingGroup(), inputIds);
  11.  
  12.    streamGraph.addOperator(transform.getId(),
  13.          slotSharingGroup,
  14.          transform.getCoLocationGroupKey(),
  15.          transform.getOperator(),
  16.          transform.getInputType(),
  17.          transform.getOutputType(),
  18.          transform.getName());
  19.  
  20.    if (transform.getStateKeySelector() != null) {
  21.       TypeSerializer<?> keySerializer = transform.getStateKeyType().createSerializer(env.getConfig());
  22.       streamGraph.setOneInputStateKey(transform.getId(), transform.getStateKeySelector(), keySerializer);
  23.    }
  24.  
  25.    streamGraph.setParallelism(transform.getId(), transform.getParallelism());
  26.    streamGraph.setMaxParallelism(transform.getId(), transform.getMaxParallelism());
  27.  
  28.    for (Integer inputId: inputIds) {
  29.       streamGraph.addEdge(inputId, transform.getId(), 0);
  30.    }
  31.  
  32.    return Collections.singleton(transform.getId());
  33. }

其中，在一开始，会得到该transformation的输入，尝试先于当前节点之前先转换其输入节点，由此，会一路往上回溯到source节点开始转换。

当其上游节点全部转换为streamGraph中的节点之后，就会开始当前节点的转换，而后，会根据determineSlotSharingGroup()方法确定该节点的slot共享名。

  1. private String determineSlotSharingGroup(String specifiedGroup, Collection<Integer> inputIds) {
  2.    if (specifiedGroup != null) {
  3.       return specifiedGroup;
  4.    } else {
  5.       String inputGroup = null;
  6.       for (int id: inputIds) {
  7.          String inputGroupCandidate = streamGraph.getSlotSharingGroup(id);
  8.          if (inputGroup == null) {
  9.             inputGroup = inputGroupCandidate;
  10.          } else if (!inputGroup.equals(inputGroupCandidate)) {
  11.             return "default";
  12.          }
  13.       }
  14.       return inputGroup == null ? "default" : inputGroup;
  15.    }
  16. }

在这里，如果没有专门制定相应的共享名，那么则会根据其上游输入节点进行确定，如果上游不一致，则会采用默认的default来指定。

 

在确定完共享名之后，streamGraph就会通过addOperaor()方法来将该节点加入到streamGraph中。在该方法中首先根据节点类型加入到stramGraph的streanNodes数组中，之后根据输入输出的类型得到相应的类型序列化对象设置到相应的节点中。

 

之后，会遍历当前节点的所有输入节点，调用addEdge()方法依次生成节点的边streamEdge。

  1. public void addEdge(Integer upStreamVertexID, Integer downStreamVertexID, int typeNumber) {
  2.    addEdgeInternal(upStreamVertexID,
  3.          downStreamVertexID,
  4.          typeNumber,
  5.          null,
  6.          new ArrayList<String>(),
  7.          null);
  8.  
  9. }
  10.  
  11. private void addEdgeInternal(Integer upStreamVertexID,
  12.       Integer downStreamVertexID,
  13.       int typeNumber,
  14.       StreamPartitioner<?> partitioner,
  15.       List<String> outputNames,
  16.       OutputTag outputTag) {
  17.  
  18.    if (virtualSideOutputNodes.containsKey(upStreamVertexID)) {
  19.       int virtualId = upStreamVertexID;
  20.       upStreamVertexID = virtualSideOutputNodes.get(virtualId).f0;
  21.       if (outputTag == null) {
  22.          outputTag = virtualSideOutputNodes.get(virtualId).f1;
  23.       }
  24.       addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, null, outputTag);
  25.    } else if (virtualSelectNodes.containsKey(upStreamVertexID)) {
  26.       int virtualId = upStreamVertexID;
  27.       upStreamVertexID = virtualSelectNodes.get(virtualId).f0;
  28.       if (outputNames.isEmpty()) {
  29.          // selections that happen downstream override earlier selections
  30.          outputNames = virtualSelectNodes.get(virtualId).f1;
  31.       }
  32.       addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, outputNames, outputTag);
  33.    } else if (virtualPartitionNodes.containsKey(upStreamVertexID)) {
  34.       int virtualId = upStreamVertexID;
  35.       upStreamVertexID = virtualPartitionNodes.get(virtualId).f0;
  36.       if (partitioner == null) {
  37.          partitioner = virtualPartitionNodes.get(virtualId).f1;
  38.       }
  39.       addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, outputNames, outputTag);
  40.    } else {
  41.       StreamNode upstreamNode = getStreamNode(upStreamVertexID);
  42.       StreamNode downstreamNode = getStreamNode(downStreamVertexID);
  43.  
  44.       // If no partitioner was specified and the parallelism of upstream and downstream
  45.       // operator matches use forward partitioning, use rebalance otherwise.
  46.       if (partitioner == null && upstreamNode.getParallelism() == downstreamNode.getParallelism()) {
  47.          partitioner = new ForwardPartitioner<Object>();
  48.       } else if (partitioner == null) {
  49.          partitioner = new RebalancePartitioner<Object>();
  50.       }
  51.  
  52.       if (partitioner instanceof ForwardPartitioner) {
  53.          if (upstreamNode.getParallelism() != downstreamNode.getParallelism()) {
  54.             throw new UnsupportedOperationException("Forward partitioning does not allow " +
  55.                   "change of parallelism. Upstream operation: " + upstreamNode + " parallelism: " + upstreamNode.getParallelism() +
  56.                   ", downstream operation: " + downstreamNode + " parallelism: " + downstreamNode.getParallelism() +
  57.                   " You must use another partitioning strategy, such as broadcast, rebalance, shuffle or global.");
  58.          }
  59.       }
  60.  
  61.       StreamEdge edge = new StreamEdge(upstreamNode, downstreamNode, typeNumber, outputNames, partitioner, outputTag);
  62.  
  63.       getStreamNode(edge.getSourceId()).addOutEdge(edge);
  64.       getStreamNode(edge.getTargetId()).addInEdge(edge);
  65.    }
  66. }

此处，可以看到一条StreamEdge的组成主要有以下几个要素，起始节点，终点节点，输出分区类型，output选择器选择的字段名。

在生成了相应的streamEdge会分别加在起始节点的 out边和终点节点的input边上。

至此，一个节点被加入到streamGraph的流程结束。

 

之后以spilt,select类型的transform为例子。

  1. private <T> Collection<Integer> transformSplit(SplitTransformation<T> split) {
  2.  
  3.    StreamTransformation<T> input = split.getInput();
  4.    Collection<Integer> resultIds = transform(input);
  5.  
  6.    // the recursive transform call might have transformed this already
  7.    if (alreadyTransformed.containsKey(split)) {
  8.       return alreadyTransformed.get(split);
  9.    }
  10.  
  11.    for (int inputId : resultIds) {
  12.       streamGraph.addOutputSelector(inputId, split.getOutputSelector());
  13.    }
  14.  
  15.    return resultIds;
  16. }

spilt类型的transformation不会加入到streamGraph中，而是会将split中的outputSelector加入到上游节点中。

  1. private <T> Collection<Integer> transformSelect(SelectTransformation<T> select) {
  2.    StreamTransformation<T> input = select.getInput();
  3.    Collection<Integer> resultIds = transform(input);
  4.  
  5.    // the recursive transform might have already transformed this
  6.    if (alreadyTransformed.containsKey(select)) {
  7.       return alreadyTransformed.get(select);
  8.    }
  9.  
  10.    List<Integer> virtualResultIds = new ArrayList<>();
  11.  
  12.    for (int inputId : resultIds) {
  13.       int virtualId = StreamTransformation.getNewNodeId();
  14.       streamGraph.addVirtualSelectNode(inputId, virtualId, select.getSelectedNames());
  15.       virtualResultIds.add(virtualId);
  16.    }
  17.    return virtualResultIds;
  18. }

而select类型的节点则会加入到streamGraph中的虚拟节点中，在之前的edge构造中，就将根据实际节点的id得到虚拟select节点，将相应的outputName加入到edge之中。

分区策略也将类似select的形式生成虚拟节点。

  1. private <T> Collection<Integer> transformPartition(PartitionTransformation<T> partition) {
  2.    StreamTransformation<T> input = partition.getInput();
  3.    List<Integer> resultIds = new ArrayList<>();
  4.  
  5.    Collection<Integer> transformedIds = transform(input);
  6.    for (Integer transformedId: transformedIds) {
  7.       int virtualId = StreamTransformation.getNewNodeId();
  8.       streamGraph.addVirtualPartitionNode(transformedId, virtualId, partition.getPartitioner());
  9.       resultIds.add(virtualId);
  10.    }
  11.  
  12.    return resultIds;
  13. }

类似的union也不会引起新的节点生成，只是会统计所有输入节点id，交由统一在下一个实际节点中一次获取。

 

 

最后生成的streamGraph可以转化为下方类似的json。

{"nodes":[{"id":1,"type":"Source: Collection Source","pact":"Data Source","contents":"Source: Collection Source","parallelism":1},{"id":3,"type":"Map","pact":"Operator","contents":"Map","parallelism":8,"predecessors":[{"id":1,"ship_strategy":"REBALANCE","side":"second"}]},{"id":7,"type":"Map","pact":"Operator","contents":"Map","parallelism":8,"predecessors":[{"id":3,"ship_strategy":"BROADCAST","side":"second"}]},{"id":9,"type":"Map","pact":"Operator","contents":"Map","parallelism":8,"predecessors":[{"id":3,"ship_strategy":"FORWARD","side":"second"}]},{"id":13,"type":"Map","pact":"Operator","contents":"Map","parallelism":8,"predecessors":[{"id":3,"ship_strategy":"FORWARD","side":"second"}]},{"id":17,"type":"Map","pact":"Operator","contents":"Map","parallelism":8,"predecessors":[{"id":3,"ship_strategy":"FORWARD","side":"second"}]},{"id":24,"type":"Map","pact":"Operator","contents":"Map","parallelism":8,"predecessors":[{"id":9,"ship_strategy":"BROADCAST","side":"second"},{"id":13,"ship_strategy":"GLOBAL","side":"second"},{"id":17,"ship_strategy":"SHUFFLE","side":"second"}]},{"id":8,"type":"Sink: Unnamed","pact":"Data Sink","contents":"Sink: Unnamed","parallelism":8,"predecessors":[{"id":7,"ship_strategy":"FORWARD","side":"second"}]},{"id":25,"type":"Sink: Unnamed","pact":"Data Sink","contents":"Sink: Unnamed","parallelism":8,"predecessors":[{"id":24,"ship_strategy":"FORWARD","side":"second"}]}]}

在flink给出的https://flink.apache.org/visualizer/ 上可以转化为可视化图。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/你好赵伟/article/detail/794413
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号