入口：

org.apache.flink.streaming.api.environment.StreamExecutionEnvironment#getStreamGraph(java.lang.String)

StreamGraph介绍

  • streamGraph的图片

image.png

  • image.png
  • streamGraph由哪些部分组成

  • image.png

    • addSource 返回dataStreamSource
    • addSinck 返回dataStreamSink

      源码分析

  • 所有的transformation都保存在 transformations 里，transformations是转换算子的集合，但不包含source，构建任务时进一步转换为DataStream

    示例任务

    ```java DataStream text = env.fromElements(WordCountData.WORDS);

DataStream> counts = // split up the lines in pairs (2-tuples) containing: (word,1) text.flatMap(new Tokenizer()) // group by the tuple field “0” and sum up tuple field “1” .keyBy(value -> value.f0).sum(1); counts.print();

  2. 此任务source是一个collection，sink则是print输出，中间有2个transformation，分别为 flatMap以及keyBy
  6. org.apache.flink.streaming.api.environment.StreamExecutionEnvironment#getStreamGraph(java.lang.String, boolean)
  7. ```java
  8.     public StreamGraph getStreamGraph(String jobName, boolean clearTransformations) {
  9.         StreamGraph streamGraph = getStreamGraphGenerator().setJobName(jobName).generate();
  10.         if (clearTransformations) {
  11.             this.transformations.clear();
  12.         }
  13.         return streamGraph;
  14.     }

获取 StreamGraphGenerator ，由 StreamGraphGenerator构建StreamGraph

StreamGraphGenerator 组成

  1.  private StreamGraphGenerator getStreamGraphGenerator() {
  2.         if (transformations.size() <= 0) {
  3.             throw new IllegalStateException(
  4.                     "No operators defined in streaming topology. Cannot execute.");
  5.         }
  7.         final RuntimeExecutionMode executionMode = configuration.get(ExecutionOptions.RUNTIME_MODE);
  9.         return new StreamGraphGenerator(transformations, config, checkpointCfg, getConfiguration())
  10.                 .setRuntimeExecutionMode(executionMode)
  11.                 .setStateBackend(defaultStateBackend)
  12.                 .setChaining(isChainingEnabled)
  13.                 .setUserArtifacts(cacheFile)
  14.                 .setTimeCharacteristic(timeCharacteristic)
  15.                 .setDefaultBufferTimeout(bufferTimeout);
  16.     }

StreamGraph 构建

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

StreamNode生成

  2.         for (Transformation<?> transformation: transformations) {
  3.             transform(transformation);
  4.         }

将transformation转为StreamNode
示例的 transformation 为
image.png
一共是三个transformation，分别为flatmap，keyBy，print对应的transformation，source则是后面会对flatmap转换的时候，递归获取其上游得到source的

  1.     private Collection<Integer> transform(Transformation<?> transform) {
  3.          transform.getOutputType();
  6.         final TransformationTranslator<?, Transformation<?>> translator =
  7.                 (TransformationTranslator<?, Transformation<?>>) translatorMap.get(transform.getClass());
  9.         Collection<Integer> transformedIds;
  10.         if (translator != null) {
  11.             transformedIds = translate(translator, transform);
  12.         } else {
  13.             transformedIds = legacyTransform(transform);
  14.         }
  17.         return transformedIds;
  18.     }

根据transform的class在translatorMap中获取对应的TransformationTranslator

  1.     static {
  2.         @SuppressWarnings("rawtypes")
  3.         Map<Class<? extends Transformation>, TransformationTranslator<?, ? extends Transformation>> tmp = new HashMap<>();
  4.         tmp.put(OneInputTransformation.class, new OneInputTransformationTranslator<>());
  5.         tmp.put(TwoInputTransformation.class, new TwoInputTransformationTranslator<>());
  6.         tmp.put(MultipleInputTransformation.class, new MultiInputTransformationTranslator<>());
  7.         tmp.put(KeyedMultipleInputTransformation.class, new MultiInputTransformationTranslator<>());
  8.         tmp.put(SourceTransformation.class, new SourceTransformationTranslator<>());
  9.         tmp.put(SinkTransformation.class, new SinkTransformationTranslator<>());
  10.         tmp.put(LegacySinkTransformation.class, new LegacySinkTransformationTranslator<>());
  11.         tmp.put(LegacySourceTransformation.class, new LegacySourceTransformationTranslator<>());
  12.         tmp.put(UnionTransformation.class, new UnionTransformationTranslator<>());
  13.         tmp.put(PartitionTransformation.class, new PartitionTransformationTranslator<>());
  14.         tmp.put(SideOutputTransformation.class, new SideOutputTransformationTranslator<>());
  15.         tmp.put(ReduceTransformation.class, new ReduceTransformationTranslator<>());
  16.         tmp.put(TimestampsAndWatermarksTransformation.class, new TimestampsAndWatermarksTransformationTranslator<>());
  17.         tmp.put(BroadcastStateTransformation.class, new BroadcastStateTransformationTranslator<>());
  18.         translatorMap = Collections.unmodifiableMap(tmp);
  19.     }
  • 如果获取到 TransformationTranslator

org.apache.flink.streaming.api.graph.StreamGraphGenerator#translate




  • 如果未获取到 TransformationTranslator 则代表是虚拟分区节点？

org.apache.flink.streaming.runtime.translators.AbstractOneInputTransformationTranslator#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);
  13.         final StreamGraph streamGraph = context.getStreamGraph();
  14.         final String slotSharingGroup = context.getSlotSharingGroup();
  15.         final int transformationId = transformation.getId();
  16.         final ExecutionConfig executionConfig = streamGraph.getExecutionConfig();
  18.         //添加StreamNode
  19.         streamGraph.addOperator(
  20.                 transformationId,
  21.                 slotSharingGroup,
  22.                 transformation.getCoLocationGroupKey(),
  23.                 operatorFactory,
  24.                 inputType,
  25.                 transformation.getOutputType(),
  26.                 transformation.getName());
  28.         if (stateKeySelector != null) {
  29.             TypeSerializer<?> keySerializer = stateKeyType.createSerializer(executionConfig);
  30.             streamGraph.setOneInputStateKey(transformationId, stateKeySelector, keySerializer);
  31.         }
  33.         int parallelism =
  34.                 transformation.getParallelism() != ExecutionConfig.PARALLELISM_DEFAULT
  35.                         ? transformation.getParallelism()
  36.                         : executionConfig.getParallelism();
  37.         streamGraph.setParallelism(transformationId, parallelism);
  38.         streamGraph.setMaxParallelism(transformationId, transformation.getMaxParallelism());
  40.         final List<Transformation<?>> parentTransformations = transformation.getInputs();
  41.         checkState(
  42.                 parentTransformations.size() == 1,
  43.                 "Expected exactly one input transformation but found "
  44.                         + parentTransformations.size());
  46.         for (Integer inputId : context.getStreamNodeIds(parentTransformations.get(0))) {
  47.             //添加StreamEdge 按照类型添加边 
  48.             streamGraph.addEdge(inputId, transformationId, 0);
  49.         }
  51.         return Collections.singleton(transformationId);
  52.     }

如果是keyby union这种节点会生成虚拟分区节点