devbox
菜鸟追梦旅行
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Flink源码系列(生成StreamGraph)-第三期_flink 生成stream ing graph

作者：菜鸟追梦旅行 | 2024-06-01 23:33:14

赞

踩

flink 生成stream ing graph

上一期指路：

第二期

上一期我们分析完了用户代码，其核心是把相关算子加入transformations中，这个对于生成流图很重要。我们在编写用户代码时，最后一步肯定会写env.execute这一步，如果不写的话，那么程序其实并没有真正意义上的运行。于是我们接着从execute函数开始进行源码分析

1.StreamExecutionEnvironment#execute

  1. 	public JobExecutionResult execute(String jobName) throws Exception {
  2. 		Preconditions.checkNotNull(jobName, "Streaming Job name should not be null.");
  3.  
  4. 		return execute(getStreamGraph(jobName));
  5. 	}

由于我们这一期只分析生成流图，所以这一期主要分析getStreamGraph函数，下一期就分析获取到流图后继续执行execute的过程。

2.StreamExecutionEnvironment#getStreamGraph->StreamExecutionEnvironment#getStreamGraph

  1. 	public StreamGraph getStreamGraph(String jobName, boolean clearTransformations) {
  2. 		StreamGraph streamGraph = getStreamGraphGenerator().setJobName(jobName).generate();
  3. 		if (clearTransformations) {
  4. 			this.transformations.clear();
  5. 		}
  6. 		return streamGraph;
  7. 	}

3.StreamGraphGenerator#generate

  1. 	public StreamGraph generate() {
  2. 		streamGraph = new StreamGraph(executionConfig, checkpointConfig, savepointRestoreSettings);
  3. 		shouldExecuteInBatchMode = shouldExecuteInBatchMode(runtimeExecutionMode);
  4. 		configureStreamGraph(streamGraph);
  5.  
  6. 		alreadyTransformed = new HashMap<>();
  7.  
  8. 		for (Transformation<?> transformation: transformations) {
  9. 			transform(transformation);
  10. 		}
  11.  
  12. 		final StreamGraph builtStreamGraph = streamGraph;
  13.  
  14. 		alreadyTransformed.clear();
  15. 		alreadyTransformed = null;
  16. 		streamGraph = null;
  17.  
  18. 		return builtStreamGraph;
  19. 	}

 重点关注transform(transformation)，点击进入

  1. 	private Collection<Integer> transform(Transformation<?> transform) {
  2. 		if (alreadyTransformed.containsKey(transform)) {
  3. 			return alreadyTransformed.get(transform);
  4. 		}
  5.  
  6. 		LOG.debug("Transforming " + transform);
  7.  
  8. 		if (transform.getMaxParallelism() <= 0) {
  9.  
  10. 			// if the max parallelism hasn't been set, then first use the job wide max parallelism
  11. 			// from the ExecutionConfig.
  12. 			int globalMaxParallelismFromConfig = executionConfig.getMaxParallelism();
  13. 			if (globalMaxParallelismFromConfig > 0) {
  14. 				transform.setMaxParallelism(globalMaxParallelismFromConfig);
  15. 			}
  16. 		}
  17.  
  18. 		// call at least once to trigger exceptions about MissingTypeInfo
  19. 		transform.getOutputType();
  20.  
  21. 		@SuppressWarnings("unchecked")
  22. 		final TransformationTranslator<?, Transformation<?>> translator =
  23. 				(TransformationTranslator<?, Transformation<?>>) translatorMap.get(transform.getClass());
  24.  
  25. 		Collection<Integer> transformedIds;
  26. 		if (translator != null) {
  27. 			transformedIds = translate(translator, transform);
  28. 		} else {
  29. 			transformedIds = legacyTransform(transform);
  30. 		}
  31.  
  32. 		// need this check because the iterate transformation adds itself before
  33. 		// transforming the feedback edges
  34. 		if (!alreadyTransformed.containsKey(transform)) {
  35. 			alreadyTransformed.put(transform, transformedIds);
  36. 		}
  37.  
  38. 		return transformedIds;
  39. 	}

遍历所有transformations并转换,转换成 StreamGraph 中的 StreamNode 和 StreamEdge；返回值为该 transform 的 id 集合。

关注translate(translator, transform)，点击并进入

  1. 	private Collection<Integer> translate(
  2. 			final TransformationTranslator<?, Transformation<?>> translator,
  3. 			final Transformation<?> transform) {
  4. 		checkNotNull(translator);
  5. 		checkNotNull(transform);
  6.  
  7. 		final List<Collection<Integer>> allInputIds = getParentInputIds(transform.getInputs());
  8.  
  9. 		// the recursive call might have already transformed this
  10. 		if (alreadyTransformed.containsKey(transform)) {
  11. 			return alreadyTransformed.get(transform);
  12. 		}
  13.  
  14. 		final String slotSharingGroup = determineSlotSharingGroup(
  15. 				transform.getSlotSharingGroup(),
  16. 				allInputIds.stream()
  17. 						.flatMap(Collection::stream)
  18. 						.collect(Collectors.toList()));
  19.  
  20. 		final TransformationTranslator.Context context = new ContextImpl(
  21. 				this, streamGraph, slotSharingGroup, configuration);
  22.  
  23. 		return shouldExecuteInBatchMode
  24. 				? translator.translateForBatch(transform, context)
  25. 				: translator.translateForStreaming(transform, context);
  26. 	}

我们直接看最后几行，我们是流模式，所以执行translator.translateForStreaming(transform, context)，点击进入

  1. 	public Collection<Integer> translateForStreaming(final T transformation, final Context context) {
  2. 		checkNotNull(transformation);
  3. 		checkNotNull(context);
  4.  
  5. 		final Collection<Integer> transformedIds =
  6. 				translateForStreamingInternal(transformation, context);
  7. 		configure(transformation, context);
  8.  
  9. 		return transformedIds;
  10. 	}

点击translateForStreamingInternal，要分情况，加入是keyBy的话，会执行PartitionTransformationTranslator中的translateForStreamingInternal，如果是flatMap，会执行OneInputTransformationTranslator中的translateForStreamingInternal。

4.PartitionTransformationTranslator#translateForStreamingInternal

  1. 	protected Collection<Integer> translateForStreamingInternal(
  2. 			final PartitionTransformation<OUT> transformation,
  3. 			final Context context) {
  4. 		return translateInternal(transformation, context);
  5. 	}

 点击translateInternal进入

  1. 	private Collection<Integer> translateInternal(
  2. 			final PartitionTransformation<OUT> transformation,
  3. 			final Context context) {
  4. 		checkNotNull(transformation);
  5. 		checkNotNull(context);
  6.  
  7. 		final StreamGraph streamGraph = context.getStreamGraph();
  8.  
  9. 		final List<Transformation<?>> parentTransformations = transformation.getInputs();
  10. 		checkState(
  11. 				parentTransformations.size() == 1,
  12. 				"Expected exactly one input transformation but found " + parentTransformations.size());
  13. 		final Transformation<?> input = parentTransformations.get(0);
  14.  
  15. 		List<Integer> resultIds = new ArrayList<>();
  16.  
  17. 		for (Integer inputId: context.getStreamNodeIds(input)) {
  18. 			final int virtualId = Transformation.getNewNodeId();
  19. 			streamGraph.addVirtualPartitionNode(
  20. 					inputId,
  21. 					virtualId,
  22. 					transformation.getPartitioner(),
  23. 					transformation.getShuffleMode());
  24. 			resultIds.add(virtualId);
  25. 		}
  26. 		return resultIds;
  27. 	}

5.OneInputTransformationTranslator#translateForStreamingInternal

  1. 	public Collection<Integer> translateForStreamingInternal(
  2. 			final OneInputTransformation<IN, OUT> transformation,
  3. 			final Context context) {
  4. 		return translateInternal(transformation,
  5. 			transformation.getOperatorFactory(),
  6. 			transformation.getInputType(),
  7. 			transformation.getStateKeySelector(),
  8. 			transformation.getStateKeyType(),
  9. 			context
  10. 		);
  11. 	}

点击translateInternal进入

  1. 	protected Collection<Integer> translateInternal(
  2. 			final Transformation<OUT> transformation,
  3. 			final StreamOperatorFactory<OUT> operatorFactory,
  4. 			final TypeInformation<IN> inputType,
  5. 			@Nullable final KeySelector<IN, ?> stateKeySelector,
  6. 			@Nullable final TypeInformation<?> stateKeyType,
  7. 			final Context context) {
  8. 		checkNotNull(transformation);
  9. 		checkNotNull(operatorFactory);
  10. 		checkNotNull(inputType);
  11. 		checkNotNull(context);
  12.  
  13. 		final StreamGraph streamGraph = context.getStreamGraph();
  14. 		final String slotSharingGroup = context.getSlotSharingGroup();
  15. 		final int transformationId = transformation.getId();
  16. 		final ExecutionConfig executionConfig = streamGraph.getExecutionConfig();
  17.  
  18. 		streamGraph.addOperator(
  19. 			transformationId,
  20. 			slotSharingGroup,
  21. 			transformation.getCoLocationGroupKey(),
  22. 			operatorFactory,
  23. 			inputType,
  24. 			transformation.getOutputType(),
  25. 			transformation.getName());
  26.  
  27. 		if (stateKeySelector != null) {
  28. 			TypeSerializer<?> keySerializer = stateKeyType.createSerializer(executionConfig);
  29. 			streamGraph.setOneInputStateKey(transformationId, stateKeySelector, keySerializer);
  30. 		}
  31.  
  32. 		int parallelism = transformation.getParallelism() != ExecutionConfig.PARALLELISM_DEFAULT
  33. 			? transformation.getParallelism()
  34. 			: executionConfig.getParallelism();
  35. 		streamGraph.setParallelism(transformationId, parallelism);
  36. 		streamGraph.setMaxParallelism(transformationId, transformation.getMaxParallelism());
  37.  
  38. 		final List<Transformation<?>> parentTransformations = transformation.getInputs();
  39. 		checkState(
  40. 			parentTransformations.size() == 1,
  41. 			"Expected exactly one input transformation but found " + parentTransformations.size());
  42.  
  43. 		for (Integer inputId: context.getStreamNodeIds(parentTransformations.get(0))) {
  44. 			streamGraph.addEdge(inputId, transformationId, 0);
  45. 		}
  46.  
  47. 		return Collections.singleton(transformationId);
  48. 	}

6.StreamGraph#addEdge->StreamGraph#addEdgeInternal

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

7.最终StreamGraph效果

Nodes的数据如下：

  1. {
  2.     "nodes": [
  3.         {
  4.             "id": 1,
  5.             "type": "Source: Socket Stream",
  6.             "pact": "Data Source",
  7.             "contents": "Source: Socket Stream",
  8.             "parallelism": 1
  9.         },
  10.         {
  11.             "id": 2,
  12.             "type": "Flat Map",
  13.             "pact": "Operator",
  14.             "contents": "Flat Map",
  15.             "parallelism": 4,
  16.             "predecessors": [
  17.                 {
  18.                     "id": 1,
  19.                     "ship_strategy": "REBALANCE",
  20.                     "side": "second"
  21.                 }
  22.             ]
  23.         },
  24.         {
  25.             "id": 4,
  26.             "type": "Window(TumblingProcessingTimeWindows(5000), ProcessingTimeTrigger, ReduceFunction$1, PassThroughWindowFunction)",
  27.             "pact": "Operator",
  28.             "contents": "Window(TumblingProcessingTimeWindows(5000), ProcessingTimeTrigger, ReduceFunction$1, PassThroughWindowFunction)",
  29.             "parallelism": 4,
  30.             "predecessors": [
  31.                 {
  32.                     "id": 2,
  33.                     "ship_strategy": "HASH",
  34.                     "side": "second"
  35.                 }
  36.             ]
  37.         },
  38.         {
  39.             "id": 5,
  40.             "type": "Sink: Print to Std. Out",
  41.             "pact": "Data Sink",
  42.             "contents": "Sink: Print to Std. Out",
  43.             "parallelism": 1,
  44.             "predecessors": [
  45.                 {
  46.                     "id": 4,
  47.                     "ship_strategy": "REBALANCE",
  48.                     "side": "second"
  49.                 }
  50.             ]
  51.         }
  52.     ]
  53. }

画图如下

总览

本期涉及到的代码部分流程图总览如下：

我们下期见！

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

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

  

闽ICP备14008679号