Flink源码之执行图

本文请谨慎参考！！！

flink的变量名、方法名之类的命名是真的好啊

Flink 中的执行图可以分成四层：StreamGraph -> JobGraph -> ExecutionGraph -> 物理执行图。

StreamGraph：是根据用户通过 Stream API 编写的代码生成的最初的图。用来表示程序 的拓扑结构。

JobGraph：StreamGraph 经过优化后生成了 JobGraph，提交给 JobManager 的数据结构。 主要的优化为，将多个符合条件的节点 chain 在一起作为一个节点，这样可以减少数据在节 点之间流动所需要的序列化/反序列化/传输消耗。

ExecutionGraph：JobManager 根据 JobGraph 生成 ExecutionGraph。ExecutionGraph 是 JobGraph 的并行化版本，是调度层最核心的数据结构。

物 理 执 行 图 ： JobManager 根 据 ExecutionGraph 对 Job 进 行 调 度 后 ， 在 各 个 TaskManager 上部署 Task 后形成的“图”，并不是一个具体的数据结构。

在StreamExecutionEnvironment类的execute方法中

public JobExecutionResult execute(String jobName) throws Exception {
   ...
       
   return execute(getStreamGraph(jobName));
}

public StreamGraph getStreamGraph(String jobName, boolean clearTransformations) {
   StreamGraph streamGraph = getStreamGraphGenerator().setJobName(jobName).generate();  // generate()
   ...
}

点generate方法到StreamGraphGenerator类

public StreamGraph generate() {
   ...

   
   for (Transformation transformation: transformations) {
      transform(transformation);
   }

   ...
}

transformations是一个list，依次存放了 用户代码里的 算子

看看如何将算子添加到这个list里的

以map为例

去DataStream类找到map方法

public  SingleOutputStreamOperator map(MapFunction mapper) {

   ...

   return map(mapper, outType);
}

点map直到doTransform方法

protected  SingleOutputStreamOperator doTransform(
      String operatorName,
      TypeInformation outTypeInfo,
      StreamOperatorFactory operatorFactory) {

   // read the output type of the input Transform to coax out errors about MissingTypeInfo
   transformation.getOutputType();

   OneInputTransformation resultTransform = new OneInputTransformation<>(
         this.transformation,
         operatorName,
         operatorFactory,
         outTypeInfo,
         environment.getParallelism());

   @SuppressWarnings({"unchecked", "rawtypes"})
   SingleOutputStreamOperator returnStream = new SingleOutputStreamOperator(environment, resultTransform);

   getExecutionEnvironment().addOperator(resultTransform);

   return returnStream;
}

点addOperator方法到StreamExecutionEnvironment类，就在这添加的

public void addOperator(Transformation transformation) {
   Preconditions.checkNotNull(transformation, "transformation must not be null.");
   this.transformations.add(transformation);
}

回到generate方法点transform方法直到translate方法，在这判断是批还是流处理

private Collection translate(...) {
   ...

   return shouldExecuteInBatchMode
         ? translator.translateForBatch(transform, context)
         : translator.translateForStreaming(transform, context);
}

继续点，点translateForStreaming方法SimpleTransformationTranslator类

public Collection translateForStreaming(final T transformation, final Context context) {
   ...

   // 区分 map之类的转换算子（OneInput） 和 keyby值类的分区算子（partition）
   final Collection transformedIds =
         translateForStreamingInternal(transformation, context);
   ...
}

这个translateForStreamingInternal方法是个抽象类SimpleTransformationTranslator里的抽象方法，他有几个的实现类

①先看OneInputTransformationTranslator类

像map算子就走OneInputTransformationTranslator类

继续点直到translateInternal方法，本次咱就只看与streamGraph对象相关的，像addXXX，其他几个实现类也一样

protected Collection translateInternal(...) {
   ...
      
   final StreamGraph streamGraph = context.getStreamGraph();
   
   ...

   
   streamGraph.addOperator(
      transformationId,
      slotSharingGroup,
      transformation.getCoLocationGroupKey(),
      operatorFactory,
      inputType,
      transformation.getOutputType(),
      transformation.getName());
   ...
       
   
   for (Integer inputId: context.getStreamNodeIds(parentTransformations.get(0))) {
      streamGraph.addEdge(inputId, transformationId, 0);
   }

   ...
}

点addOperator方法直到StreamGraph类的addNode方法

protected StreamNode addNode(
      Integer vertexID,
      @Nullable String slotSharingGroup,
      @Nullable String coLocationGroup,
      Class vertexClass,
      StreamOperatorFactory operatorFactory,
      String operatorName) {

   if (streamNodes.containsKey(vertexID)) {
      throw new RuntimeException("Duplicate vertexID " + vertexID);
   }

   StreamNode vertex = new StreamNode(
         vertexID,
         slotSharingGroup,
         coLocationGroup,
         operatorFactory,
         operatorName,
         vertexClass);

   streamNodes.put(vertexID, vertex);

   return vertex;
}

有必要看一下StreamGraph、StreamNode、StreamEdge的部分内容

StreamGraph类里的部分成员变量

private String jobName;
private ScheduleMode scheduleMode;
private boolean chaining;

private Map streamNodes;
private Set sources;
private Set sinks;
private Map> virtualSideOutputNodes;
private Map, ShuffleMode>> virtualPartitionNodes;

protected Map vertexIDtoBrokerID;
protected Map vertexIDtoLoopTimeout;
private StateBackend stateBackend;
private Set> iterationSourceSinkPairs;
private InternalTimeServiceManager.Provider timerServiceProvider;

StreamNode类的部分成员变量和方法

private List inEdges = new ArrayList();
private List outEdges = new ArrayList();

public void addInEdge(StreamEdge inEdge) {
	...
	inEdges.add(inEdge);
}

public void addOutEdge(StreamEdge outEdge) {
	...
	outEdges.add(outEdge);	
}

StreamEdge类的部分成员变量和方法

private final String edgeId;
private final int sourceId;
private final int targetId;

public StreamEdge(
	StreamNode sourceVertex,
	StreamNode targetVertex,
	int typeNumber,
	StreamPartitioner outputPartitioner,
	OutputTag outputTag,
	ShuffleMode shuffleMode) {

	...
}

public int getSourceId() {
   return sourceId;
}

public int getTargetId() {
   return targetId;
}

回到AbstractOneInputTransformationTranslator类的translateInternal方法，点addEdge方法直到StreamGraph类的addEdgeInternal方法

private void addEdgeInternal(Integer upStreamVertexID,
      Integer downStreamVertexID,
      int typeNumber,
      StreamPartitioner partitioner,
      List outputNames,
      OutputTag outputTag,
      ShuffleMode shuffleMode) {

   
   if (virtualSideOutputNodes.containsKey(upStreamVertexID)) {
      int virtualId = upStreamVertexID;
      upStreamVertexID = virtualSideOutputNodes.get(virtualId).f0;
      if (outputTag == null) {
         outputTag = virtualSideOutputNodes.get(virtualId).f1;
      }
      addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, null, outputTag, shuffleMode);
   } else if (virtualPartitionNodes.containsKey(upStreamVertexID)) {
      
      int virtualId = upStreamVertexID;
      upStreamVertexID = virtualPartitionNodes.get(virtualId).f0;
      if (partitioner == null) {
         partitioner = virtualPartitionNodes.get(virtualId).f1;
      }
      shuffleMode = virtualPartitionNodes.get(virtualId).f2;
      addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, outputNames, outputTag, shuffleMode);
   } else {
      
      StreamNode upstreamNode = getStreamNode(upStreamVertexID);
      StreamNode downstreamNode = getStreamNode(downStreamVertexID);

      // If no partitioner was specified and the parallelism of upstream and downstream
      // operator matches use forward partitioning, use rebalance otherwise.
      
      if (partitioner == null && upstreamNode.getParallelism() == downstreamNode.getParallelism()) {
         partitioner = new ForwardPartitioner