1.概述

转载：Flink源码分析——批处理模式JobGraph的创建 仅供自己学习。

Flink不管是流处理还是批处理都是将我们的程序编译成JobGraph进行提交的，之前我们分析过流处理模式下的JobGraph创建，现在我们来分析一下批处理模式下的JobGraph创建。

本文以本地模式为例，分析JobGraph的创建

我们仍然以WordCount为例子来分析JobGraph的创建过程，WordCount代码

val env = ExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
//    env.getConfig.setExecutionMode(ExecutionMode.BATCH)
    val text = env.fromElements(
      "Who's there?",
      "I think I hear them. Stand, ho! Who's there?",
      "li wen tao li wen tao li wen tao"
    )
    text.flatMap { _.toLowerCase.split("\\W+").filter{ _.nonEmpty } }
      .map { (_, 1) }
      .groupBy(0)
      .sum(1)
      .writeAsText("D:\\IDEASPARK\\flink\\wordcount", WriteMode.OVERWRITE)
    env.execute()

这个WordCount执行之后生成的DataSet关系图如下所示：

DataSource ——> FlatMapOperator ——> MapOperator ——> ScalaAggregateOperator ——> DataSink

注意这里的Operator并非是指算子层面的operator，而是在数据集层面的operator，这些operator也还是DataSet的子类型（DataSink除外）

首先看一下执行入口，在本地模式下，会执行LocalEnvironment.execute()方法，先创建执行计划Plan，再开始执行这个计划

//LocalEnvironment
public JobExecutionResult execute(String jobName) throws Exception {
   if (executor == null) {
      startNewSession();
   }
   Plan p = createProgramPlan(jobName);
   // Session management is disabled, revert this commit to enable
   //p.setJobId(jobID);
   //p.setSessionTimeout(sessionTimeout);
   JobExecutionResult result = executor.executePlan(p);
   this.lastJobExecutionResult = result;
   return result;
}

这个执行计划Plan很简单，里面只包含了一些sinks，先创建执行计划的过程就是将WordCount代码中创建的每个DataSet转换成对应算子层面的operator。

2.创建执行计划Plan

首先我们来看看createProgramPlan()源码实现

//ExecutionEnvironment
public Plan createProgramPlan(String jobName, boolean clearSinks) {
   ...
   //创建一个translator转换器，从sink开始转换
   OperatorTranslation translator = new OperatorTranslation();
   Plan plan = translator.translateToPlan(this.sinks, jobName);
   ...
   return plan;
}

//OperatorTranslation
public Plan translateToPlan(List<DataSink<?>> sinks, String jobName) {
   List<GenericDataSinkBase<?>> planSinks = new ArrayList<>();
   //从sink开始进行向上的深度优先遍历
   for (DataSink<?> sink : sinks) {
      planSinks.add(translate(sink));
   }
   Plan p = new Plan(planSinks);
   p.setJobName(jobName);
   return p;
}
private <T> GenericDataSinkBase<T> translate(DataSink<T> sink) {
   // translate the input recursively
   //从sink开始递归的向上去进行转换
   Operator<T> input = translate(sink.getDataSet());
   // translate the sink itself and connect it to the input
   GenericDataSinkBase<T> translatedSink = sink.translateToDataFlow(input);
   translatedSink.setResources(sink.getMinResources(), sink.getPreferredResources());
   return translatedSink;
}
private <T> Operator<T> translate(DataSet<T> dataSet) {
   while (dataSet instanceof NoOpOperator) {
      dataSet = ((NoOpOperator<T>) dataSet).getInput();
   }
   // check if we have already translated that data set (operation or source)
   Operator<?> previous = this.translated.get(dataSet);
   if (previous != null) {
      ... //已经转换过了
   }
   Operator<T> dataFlowOp;
   if (dataSet instanceof DataSource) {
      DataSource<T> dataSource = (DataSource<T>) dataSet;
      dataFlowOp = dataSource.translateToDataFlow();
      dataFlowOp.setResources(dataSource.getMinResources(), dataSource.getPreferredResources());
   }
   else if (dataSet instanceof SingleInputOperator) {
      SingleInputOperator<?, ?, ?> singleInputOperator = (SingleInputOperator<?, ?, ?>) dataSet;
      dataFlowOp = translateSingleInputOperator(singleInputOperator);
      dataFlowOp.setResources(singleInputOperator.getMinResources(), singleInputOperator.getPreferredResources());
   }
   else if (dataSet instanceof TwoInputOperator) {
      TwoInputOperator<?, ?, ?, ?> twoInputOperator = (TwoInputOperator<?, ?, ?, ?>) dataSet;
      dataFlowOp = translateTwoInputOperator(twoInputOperator);
      dataFlowOp.setResources(twoInputOperator.getMinResources(), twoInputOperator.getPreferredResources());
   }else if
   ...
   this.translated.put(dataSet, dataFlowOp);
   // take care of broadcast variables
   translateBcVariables(dataSet, dataFlowOp);
   return dataFlowOp;
}
private <I, O> org.apache.flink.api.common.operators.Operator<O> translateSingleInputOperator(SingleInputOperator<?, ?, ?> op) {
   @SuppressWarnings("unchecked")
   SingleInputOperator<I, O, ?> typedOp = (SingleInputOperator<I, O, ?>) op;
   @SuppressWarnings("unchecked")
   DataSet<I> typedInput = (DataSet<I>) op.getInput();
    //在遇到SingleInputOperator节点是继续向上递归，那么整个的递归过程就是从sink后续遍历，先转换source，再依次向下进行转换
   Operator<I> input = translate(typedInput);
   org.apache.flink.api.common.operators.Operator<O> dataFlowOp = typedOp.translateToDataFlow(input);
   ...
   return dataFlowOp;
}

大致实现就是从sink开始进行向上递归的转换，整个的递归过程就是从sink进行深度优化遍历，先转换source，再依次向下进行转换，转换的方法就是调用每个DataSet（或者DataSink）的translateToDataFlow()方法，将DataSet转换成算子层面的Operator，然后将上一级转换后的Operator当做输入input。

下面看一下每种DataSet(或DataSink)的translateToDataFlow()方法

//
protected GenericDataSourceBase<OUT, ?> translateToDataFlow() {
   String name = this.name != null ? this.name : "at " + dataSourceLocationName + " (" + inputFormat.getClass().getName() + ")";
   if (name.length() > 150) {
      name = name.substring(0, 150);
   }
   @SuppressWarnings({"unchecked", "rawtypes"})
   GenericDataSourceBase<OUT, ?> source = new GenericDataSourceBase(this.inputFormat,
      new OperatorInformation<OUT>(getType()), name);
   source.setParallelism(parallelism);
   if (this.parameters != null) {
      source.getParameters().addAll(this.parameters);
   }
   if (this.splitDataProperties != null) {
      source.setSplitDataProperties(this.splitDataProperties);
   }
   return source;
}
//MapOperator
protected MapOperatorBase<IN, OUT, MapFunction<IN, OUT>> translateToDataFlow(Operator<IN> input) {
   String name = getName() != null ? getName() : "Map at " + defaultName;
   // create operator
   MapOperatorBase<IN, OUT, MapFunction<IN, OUT>> po = new MapOperatorBase<IN, OUT, MapFunction<IN, OUT>>(function,
         new UnaryOperatorInformation<IN, OUT>(getInputType(), getResultType()), name);
   // set input
   po.setInput(input);
   // set parallelism
   if (this.getParallelism() > 0) {
      // use specified parallelism
      po.setParallelism(this.getParallelism());
   } else {
      // if no parallelism has been specified, use parallelism of input operator to enable chaining
      po.setParallelism(input.getParallelism());
   }
   return po;
}
//ScalaAggregateOperator
protected org.apache.flink.api.common.operators.base.GroupReduceOperatorBase<IN, IN, GroupReduceFunction<IN, IN>> translateToDataFlow(Operator<IN> input) {
   // sanity check
   if (this.aggregationFunctions.isEmpty() || this.aggregationFunctions.size() != this.fields.size()) {
      throw new IllegalStateException();
   }
   // construct the aggregation function
   AggregationFunction<Object>[] aggFunctions = new AggregationFunction[this.aggregationFunctions.size()];
   int[] fields = new int[this.fields.size()];
   StringBuilder genName = new StringBuilder();
   for (int i = 0; i < fields.length; i++) {
      aggFunctions[i] = (AggregationFunction<Object>) this.aggregationFunctions.get(i);
      fields[i] = this.fields.get(i);
      genName.append(aggFunctions[i].toString()).append('(').append(fields[i]).append(')').append(',');
   }
   genName.setLength(genName.length() - 1);
   @SuppressWarnings("rawtypes")
   RichGroupReduceFunction<IN, IN> function = new AggregatingUdf(getInputType(), aggFunctions, fields);
   String name = getName() != null ? getName() : genName.toString();
   // distinguish between grouped reduce and non-grouped reduce
   //这种是针对未分组的reduce
   if (this.grouping == null) {
      // non grouped aggregation
      UnaryOperatorInformation<IN, IN> operatorInfo = new UnaryOperatorInformation<>(getInputType(), getResultType());
      GroupReduceOperatorBase<IN, IN, GroupReduceFunction<IN, IN>> po =
            new GroupReduceOperatorBase<IN, IN, GroupReduceFunction<IN, IN>>(function, operatorInfo, new int[0], name);
      po.setCombinable(true);
      // set input
      po.setInput(input);
      // set parallelism
      po.setParallelism(this.getParallelism());
      return po;
   }
   //这种是针对的是分组的reduce，我们的WordCount代码走这里
   if (this.grouping.getKeys() instanceof Keys.ExpressionKeys) {
      // grouped aggregation
      int[] logicalKeyPositions = this.grouping.getKeys().computeLogicalKeyPositions();
      UnaryOperatorInformation<IN, IN> operatorInfo = new UnaryOperatorInformation<>(getInputType(), getResultType());
      GroupReduceOperatorBase<IN, IN, GroupReduceFunction<IN, IN>> po =
            new GroupReduceOperatorBase<IN, IN, GroupReduceFunction<IN, IN>>(function, operatorInfo, logicalKeyPositions, name);
      //默认就开启combiner了，数据预先进行聚合，减少数据传输
      po.setCombinable(true);
      // set input
      po.setInput(input);
      // set parallelism
      po.setParallelism(this.getParallelism());
      SingleInputSemanticProperties props = new SingleInputSemanticProperties();
      for (int keyField : logicalKeyPositions) {
         boolean keyFieldUsedInAgg = false;
         for (int aggField : fields) {
            if (keyField == aggField) {
               keyFieldUsedInAgg = true;
               break;
            }
         }
         if (!keyFieldUsedInAgg) {
            props.addForwardedField(keyField, keyField);
         }
      }
      po.setSemanticProperties(props);
      po.setCustomPartitioner(grouping.getCustomPartitioner());
      return po;
   }
   else if (this.grouping.getKeys() instanceof Keys.SelectorFunctionKeys) {
      throw new UnsupportedOperationException("Aggregate does not support grouping with KeySelector functions, yet.");
   }
   else {
      throw new UnsupportedOperationException("Unrecognized key type.");
   }
}
//DataSink
protected GenericDataSinkBase<T> translateToDataFlow(Operator<T> input) {
   // select the name (or create a default one)
   String name = this.name != null ? this.name : this.format.toString();
   GenericDataSinkBase<T> sink = new GenericDataSinkBase<>(this.format, new UnaryOperatorInformation<>(this.type, new NothingTypeInfo()), name);
   // set input
   sink.setInput(input);
   // set parameters
   if (this.parameters != null) {
      sink.getParameters().addAll(this.parameters);
   }
   // set parallelism
   if (this.parallelism > 0) {
      // use specified parallelism
      sink.setParallelism(this.parallelism);
   } else {
      // if no parallelism has been specified, use parallelism of input operator to enable chaining
      sink.setParallelism(input.getParallelism());
   }
   if (this.sortKeyPositions != null) {
      // configure output sorting
      Ordering ordering = new Ordering();
      for (int i = 0; i < this.sortKeyPositions.length; i++) {
         ordering.appendOrdering(this.sortKeyPositions[i], null, this.sortOrders[i]);
      }
      sink.setLocalOrder(ordering);
   }
   return sink;
}

经过转换，上述WordCount转换成的算子层面的Operator就如下所示：

GenericDataSourceBase --> FlatMapOperatorBase --> MapOperatorBase --> GroupReduceOperatorBase --> GenericDataSinkBase

上级operator作为下级operator的input，这样一级一级的进行链接起来。

3.编译成OptimizedPlan

接下来就到了执行这个计划的代码了，也就是executor.executePlan§，关于JobGraph的实现大致如下：

1. 创建一个优化器，对Plan进行优化，编译成OptimizedPlan
2. 创建JobGraph生成器，再对OptimizedPlan进行编译成JobGraph

public JobExecutionResult executePlan(Plan plan) throws Exception {
   if (plan == null) {
      throw new IllegalArgumentException("The plan may not be null.");
   }
   synchronized (this.lock) {
   ... //启动本地集群环境
      try {
         // TODO: Set job's default parallelism to max number of slots
         final int slotsPerTaskManager = jobExecutorServiceConfiguration.getInteger(TaskManagerOptions.NUM_TASK_SLOTS, taskManagerNumSlots);
         final int numTaskManagers = jobExecutorServiceConfiguration.getInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 1);
         plan.setDefaultParallelism(slotsPerTaskManager * numTaskManagers);
         //下面几行代码是JobGraph创建的关键过程
         Optimizer pc = new Optimizer(new DataStatistics(), jobExecutorServiceConfiguration);
         OptimizedPlan op = pc.compile(plan);
         JobGraphGenerator jgg = new JobGraphGenerator(jobExecutorServiceConfiguration);
         JobGraph jobGraph = jgg.compileJobGraph(op, plan.getJobId());
         return jobExecutorService.executeJobBlocking(jobGraph);
      }
      finally {
         if (shutDownAtEnd) {
            stop();
         }
      }
   }
}

4.优化器Optimizer

优化器Optimizer对原始计划进行编译，编译的过程大致实现如下：

1. 创建GraphCreatingVisitor，对原始的Plan进行优化，将每个operator优化成OptimizerNode，OptimizerNode之间通过DagConnection相连，DagConnection相当于一个边模型，有source和target，可以表示OptimizerNode的输入和输出
2. 对OptimizerNode再进行优化，将每个OptimizerNode优化成PlanNode，PlanNode之间通过Channel相连，Channel也相当于是一个边模型，可以表示PlanNode的输入和输出。这个过程会做很多优化，比如对GroupReduceNode会增加combiner的节点，对Channel会设置ShipStrategyType和DataExchangeMode,ShipStrategyType表示的两个节点之间数据的传输策略，比如进行hash分区，范围分区等，DataExchangeMode表示的是两个节点间数据交换的模式，有PIPELINED和BATCH

//Optimizer类
public OptimizedPlan compile(Plan program) throws CompilerException {
   final OptimizerPostPass postPasser = getPostPassFromPlan(program);
   return compile(program, postPasser);
}
private OptimizedPlan compile(Plan program, OptimizerPostPass postPasser) throws CompilerException {
    ...
   final ExecutionMode defaultDataExchangeMode = program.getExecutionConfig().getExecutionMode();
   final int defaultParallelism = program.getDefaultParallelism() > 0 ?
      program.getDefaultParallelism() : this.defaultParallelism;
   ...
   //对原始的Plan进行优化，将每个operator优化成OptimizerNode
   GraphCreatingVisitor graphCreator = new GraphCreatingVisitor(defaultParallelism, defaultDataExchangeMode);
   program.accept(graphCreator);
   // if we have a plan with multiple data sinks, add logical optimizer nodes that have two data-sinks as children
   // each until we have only a single root node. This allows to transparently deal with the nodes with
   // multiple outputs
   OptimizerNode rootNode;
   if (graphCreator.getSinks().size() == 1) {
      rootNode = graphCreator.getSinks().get(0);
   }
   else if (graphCreator.getSinks().size() > 1) {
      Iterator<DataSinkNode> iter = graphCreator.getSinks().iterator();
      rootNode = iter.next();
      while (iter.hasNext()) {
         rootNode = new SinkJoiner(rootNode, iter.next());
      }
   }
   else {
      throw new CompilerException("Bug: The optimizer plan representation has no sinks.");
   }
   // now that we have all nodes created and recorded which ones consume memory, tell the nodes their minimal
   // guaranteed memory, for further cost estimations. We assume an equal distribution of memory among consumer tasks
   rootNode.accept(new IdAndEstimatesVisitor(this.statistics));
   // We need to enforce that union nodes always forward their output to their successor.
   // Any partitioning must be either pushed before or done after the union, but not on the union's output.
   UnionParallelismAndForwardEnforcer unionEnforcer = new UnionParallelismAndForwardEnforcer();
   rootNode.accept(unionEnforcer);
   // We are dealing with operator DAGs, rather than operator trees.
   // That requires us to deviate at some points from the classical DB optimizer algorithms.
   // This step builds auxiliary structures to help track branches and joins in the DAG
   BranchesVisitor branchingVisitor = new BranchesVisitor();
   rootNode.accept(branchingVisitor);
   // Propagate the interesting properties top-down through the graph
   InterestingPropertyVisitor propsVisitor = new InterestingPropertyVisitor(this.costEstimator);
   rootNode.accept(propsVisitor);
   
   // perform a sanity check: the root may not have any unclosed branches
   if (rootNode.getOpenBranches() != null && rootNode.getOpenBranches().size() > 0) {
      throw new CompilerException("Bug: Logic for branching plans (non-tree plans) has an error, and does not " +
            "track the re-joining of branches correctly.");
   }
   // the final step is now to generate the actual plan alternatives
   //对OptimizerNode再进行优化，对每个OptimizerNode优化成PlanNode
   List<PlanNode> bestPlan = rootNode.getAlternativePlans(this.costEstimator);
   if (bestPlan.size() != 1) {
      throw new CompilerException("Error in compiler: more than one best plan was created!");
   }
   // check if the best plan's root is a data sink (single sink plan)
   // if so, directly take it. if it is a sink joiner node, get its contained sinks
   PlanNode bestPlanRoot = bestPlan.get(0);
   List<SinkPlanNode> bestPlanSinks = new ArrayList<SinkPlanNode>(4);
   if (bestPlanRoot instanceof SinkPlanNode) {
      bestPlanSinks.add((SinkPlanNode) bestPlanRoot);
   } else if (bestPlanRoot instanceof SinkJoinerPlanNode) {
      ((SinkJoinerPlanNode) bestPlanRoot).getDataSinks(bestPlanSinks);
   }
   // finalize the plan
   //创建最终的优化过的计划OptimizedPlan 
   OptimizedPlan plan = new PlanFinalizer().createFinalPlan(bestPlanSinks, program.getJobName(), program);
   
   plan.accept(new BinaryUnionReplacer());
   plan.accept(new RangePartitionRewriter(plan));
   // post pass the plan. this is the phase where the serialization and comparator code is set
   postPasser.postPass(plan);
   
   return plan;
}

5.将Operator转换成OptimizerNode

GraphCreatingVisitor对原始Plan进行优化成OptimizerNode

首先我们来看看原始Plan进行优化成OptimizerNode的过程，代码实现在program.accept(graphCreator)

//Plan
public void accept(Visitor<Operator<?>> visitor) {
   for (GenericDataSinkBase<?> sink : this.sinks) {
      sink.accept(visitor);
   }
}
//GenericDataSinkBase
public void accept(Visitor<Operator<?>> visitor) {
   boolean descend = visitor.preVisit(this);
   if (descend) {
      this.input.accept(visitor);
      visitor.postVisit(this);
   }
}
//SingleInputOperator
public void accept(Visitor<Operator<?>> visitor) {
   if (visitor.preVisit(this)) {
      this.input.accept(visitor);
      for (Operator<?> c : this.broadcastInputs.values()) {
         c.accept(visitor);
      }
      visitor.postVisit(this);
   }
}
//GenericDataSourceBase
public void accept(Visitor<Operator<?>> visitor) {
   if (visitor.preVisit(this)) {
      visitor.postVisit(this);
   }
}

从代码中可以看到，整个accept()过程就是一个递归遍历的过程，有点类似于中序遍历的过程。先从sink（GenericDataSinkBase）开始，由下至上对每个operator执行visitor.preVisit()方法，再由上至下对每个operator执行visitor.postVisit()。

既然核心方法在visitor.preVisit()和visitor.postVisit()，那我们就来看看GraphCreatingVisitor的这两个方法。

preVisit()

public boolean preVisit(Operator<?> c) {
   // check if we have been here before
   if (this.con2node.containsKey(c)) {
      return false;
   }
   final OptimizerNode n;
   // create a node for the operator (or sink or source) if we have not been here before
   if (c instanceof GenericDataSinkBase) {
      DataSinkNode dsn = new DataSinkNode((GenericDataSinkBase<?>) c);
      this.sinks.add(dsn);
      n = dsn;
   }
   else if (c instanceof GenericDataSourceBase) {
      n = new DataSourceNode((GenericDataSourceBase<?, ?>) c);
   }
   else if (c instanceof MapOperatorBase) {
      n = new MapNode((MapOperatorBase<?, ?, ?>) c);
   }
   else if (c instanceof MapPartitionOperatorBase) {
      n = new MapPartitionNode((MapPartitionOperatorBase<?, ?, ?>) c);
   }
   else if (c instanceof FlatMapOperatorBase) {
      n = new FlatMapNode((FlatMapOperatorBase<?, ?, ?>) c);
   }
   else if (c instanceof FilterOperatorBase) {
      n = new FilterNode((FilterOperatorBase<?, ?>) c);
   }
   else if (c instanceof ReduceOperatorBase) {
      n = new ReduceNode((ReduceOperatorBase<?, ?>) c);
   }
   else if (c instanceof GroupCombineOperatorBase) {
      n = new GroupCombineNode((GroupCombineOperatorBase<?, ?, ?>) c);
   }
   else if (c instanceof GroupReduceOperatorBase) {
      n = new GroupReduceNode((GroupReduceOperatorBase<?, ?, ?>) c);
   }
   else if (c instanceof InnerJoinOperatorBase) {
      n = new JoinNode((InnerJoinOperatorBase<?, ?, ?, ?>) c);
   }
   else if (c instanceof OuterJoinOperatorBase) {
      n = new OuterJoinNode((OuterJoinOperatorBase<?, ?, ?, ?>) c);
   }
   else if (c instanceof CoGroupOperatorBase) {
      n = new CoGroupNode((CoGroupOperatorBase<?, ?, ?, ?>) c);
   }
   else if (c instanceof CoGroupRawOperatorBase) {
      n = new CoGroupRawNode((CoGroupRawOperatorBase<?, ?, ?, ?>) c);
   }
   else if (c instanceof CrossOperatorBase) {
      n = new CrossNode((CrossOperatorBase<?, ?, ?, ?>) c);
   }
   else if (c instanceof BulkIterationBase) {
      n = new BulkIterationNode((BulkIterationBase<?>) c);
   }
   else if (c instanceof DeltaIterationBase) {
      n = new WorksetIterationNode((DeltaIterationBase<?, ?>) c);
   }
   else if (c instanceof Union){
      n = new BinaryUnionNode((Union<?>) c);
   }
   else if (c instanceof PartitionOperatorBase) {
      n = new PartitionNode((PartitionOperatorBase<?>) c);
   }
   else if (c instanceof SortPartitionOperatorBase) {
      n = new SortPartitionNode((SortPartitionOperatorBase<?>) c);
   }
   else if (c instanceof BulkIterationBase.PartialSolutionPlaceHolder) {
      ...
   }
   else if (c instanceof DeltaIterationBase.WorksetPlaceHolder) {
      ...
   }
   else if (c instanceof DeltaIterationBase.SolutionSetPlaceHolder) {
      ...
   }
   else {
      throw new IllegalArgumentException("Unknown operator type: " + c);
   }
   this.con2node.put(c, n);
   // set the parallelism only if it has not been set before. some nodes have a fixed parallelism, such as the
   // key-less reducer (all-reduce)
   if (n.getParallelism() < 1) {
      // set the parallelism
      int par = c.getParallelism();
      if (n instanceof BinaryUnionNode) {
         // Keep parallelism of union undefined for now.
         // It will be determined based on the parallelism of its successor.
         par = -1;
      } else if (par > 0) {
         if (this.forceParallelism && par != this.defaultParallelism) {
            par = this.defaultParallelism;
            Optimizer.LOG.warn("The parallelism of nested dataflows (such as step functions in iterations) is " +
               "currently fixed to the parallelism of the surrounding operator (the iteration).");
         }
      } else {
         par = this.defaultParallelism;
      }
      n.setParallelism(par);
   }
   return true;
}

preVisit()方法非常简单，仅仅是判断输入Operator的类型，来创建对应的OptimizerNode，然后设置并行度

postVisit()

public void postVisit(Operator<?> c) {
   OptimizerNode n = this.con2node.get(c);
   // first connect to the predecessors
   n.setInput(this.con2node, this.defaultDataExchangeMode);
   n.setBroadcastInputs(this.con2node, this.defaultDataExchangeMode);
   // if the node represents a bulk iteration, we recursively translate the data flow now
   if (n instanceof BulkIterationNode) {
     ...
   }
   else if (n instanceof WorksetIterationNode) {
      ...
   }
}

postVisit()方法也很简单，就是对每个Operator对应的OptimizerNode设置input。defaultDataExchangeMode在这里默认就是ExecutionMode.PIPELINED，也可以通过env.getConfig.setExecutionMode(ExecutionMode.BATCH)来进行设置默认的ExecutionMode。ExecutionMode表示的是两个节点间数据交换的模式，有PIPELINED和BATCH，

1. PIPELINED模式数据像流水线一样的进行传输，上游任务和下游任务能够同时进行生产和消费数据；
2. BATCH模式需要等上游的任务数据全部处理完之后才会开始下游的任务，中间数据会spill到磁盘上。

下面看看每种OptimizerNode的setInput()方法

//DataSourceNode
public void setInput(Map<Operator<?>, OptimizerNode> contractToNode, ExecutionMode defaultDataExchangeMode) {}
//SingleInputNode
public void setInput(Map<Operator<?>, OptimizerNode> contractToNode, ExecutionMode defaultExchangeMode)
      throws CompilerException
{
   // see if an internal hint dictates the strategy to use
   final Configuration conf = getOperator().getParameters();
   final String shipStrategy = conf.getString(Optimizer.HINT_SHIP_STRATEGY, null);
   final ShipStrategyType preSet;
   //默认情况下这里都是null
   if (shipStrategy != null) {
      if (shipStrategy.equalsIgnoreCase(Optimizer.HINT_SHIP_STRATEGY_REPARTITION_HASH)) {
         preSet = ShipStrategyType.PARTITION_HASH;
      } else if (shipStrategy.equalsIgnoreCase(Optimizer.HINT_SHIP_STRATEGY_REPARTITION_RANGE)) {
         preSet = ShipStrategyType.PARTITION_RANGE;
      } else if (shipStrategy.equalsIgnoreCase(Optimizer.HINT_SHIP_STRATEGY_FORWARD)) {
         preSet = ShipStrategyType.FORWARD;
      } else if (shipStrategy.equalsIgnoreCase(Optimizer.HINT_SHIP_STRATEGY_REPARTITION)) {
         preSet = ShipStrategyType.PARTITION_RANDOM;
      } else {
         throw new CompilerException("Unrecognized ship strategy hint: " + shipStrategy);
      }
   } else {
      preSet = null;
   }
   
   // get the predecessor node
   Operator<?> children = ((SingleInputOperator<?, ?, ?>) getOperator()).getInput();
   
   OptimizerNode pred;
   DagConnection conn;
   if (children == null) {
      throw new CompilerException("Error: Node for '" + getOperator().getName() + "' has no input.");
   } else {
      pred = contractToNode.get(children);
      conn = new DagConnection(pred, this, defaultExchangeMode);
      if (preSet != null) {
         conn.setShipStrategy(preSet);
      }
   }
   
   // create the connection and add it
   setIncomingConnection(conn);
   pred.addOutgoingConnection(conn);
}
//DataSinkNode
public void setInput(Map<Operator<?>, OptimizerNode> contractToNode, ExecutionMode defaultExchangeMode) {
   Operator<?> children = getOperator().getInput();
   final OptimizerNode pred;
   final DagConnection conn;
   
   pred = contractToNode.get(children);
   conn = new DagConnection(pred, this, defaultExchangeMode);
      
   // create the connection and add it
   this.input = conn;
   pred.addOutgoingConnection(conn);
}

setInput()方法就是创建了DagConnection把OptimizerNode连接在了一起。这个DagConnection就是一个边的模型，作为下游节点OptimizerNode的输入，同时作为上游节点OptimizerNode的输出。这里的DagConnection里的ShipStrategy和ExecutionMode还都是默认情况下的，不是最终的状态。

简单看一下DagConnection的结构

public class DagConnection implements EstimateProvider, DumpableConnection<OptimizerNode> {
   
   private final OptimizerNode source; // The source node of the connection
   private final OptimizerNode target; // The target node of the connection.
   private final ExecutionMode dataExchangeMode; // defines whether to use batch or pipelined data exchange
   private InterestingProperties interestingProps; // local properties that succeeding nodes are interested in
   private ShipStrategyType shipStrategy; // The data shipping strategy, if predefined.
   
   private TempMode materializationMode = TempMode.NONE; // the materialization mode
   
   private int maxDepth = -1;
   private boolean breakPipeline;  // whet

这样，经过优化器初步的优化，WordCount整个计划变成了如下的拓扑结构：

DataSourceNode --> FlatMapNode --> MapNode --> GroupReduceNode --> DataSinkNode

每个OptimizerNode之间通过DagConnection进行连接

6.将OptimizerNode进一步优化成PlanNode

接下来是进一步的优化，将OptimizerNode优化成PlanNode，PlanNode是最终的优化节点类型，它包含了节点的更多属性，节点之间通过Channel进行连接，Channel也是一种边模型，同时确定了节点之间的数据交换方式ShipStrategyType和DataExchangeMode，ShipStrategyType表示的两个节点之间数据的传输策略，比如是否进行数据分区，进行hash分区，范围分区等; DataExchangeMode表示的是两个节点间数据交换的模式，有PIPELINED和BATCH，和ExecutionMode是一样的，ExecutionMode决定了DataExchangeMode。

代码实现在rootNode.getAlternativePlans()，这个rootNode也就是DataSinkNode

public List<PlanNode> getAlternativePlans(CostEstimator estimator) {
   // check if we have a cached version
   if (this.cachedPlans != null) {
      return this.cachedPlans;
   }
   
   // calculate alternative sub-plans for predecessor
   //递归的向上创建PlanNode，再创建当前节点，后序遍历
   List<? extends PlanNode> subPlans = getPredecessorNode().getAlternativePlans(estimator);
   List<PlanNode> outputPlans = new ArrayList<PlanNode>();
   
   final int parallelism = getParallelism();
   final int inDop = getPredecessorNode().getParallelism();
   final ExecutionMode executionMode = this.input.getDataExchangeMode();
   final boolean dopChange = parallelism != inDop;
   final boolean breakPipeline = this.input.isBreakingPipeline();
   InterestingProperties ips = this.input.getInterestingProperties();
   for (PlanNode p : subPlans) {
      for (RequestedGlobalProperties gp : ips.getGlobalProperties()) {
         for (RequestedLocalProperties lp : ips.getLocalProperties()) {
             //创建Channel，并对channel进行参数化赋值
            Channel c = new Channel(p);
            gp.parameterizeChannel(c, dopChange, executionMode, breakPipeline);
            lp.parameterizeChannel(c);
            c.setRequiredLocalProps(lp);
            c.setRequiredGlobalProps(gp);
            
            // no need to check whether the created properties meet what we need in case
            // of ordering or global ordering, because the only interesting properties we have
            // are what we require
            //创建一个SinkPlanNode，channel作为SinkPlanNode的输入
            outputPlans.add(new SinkPlanNode(this, "DataSink ("+this.getOperator().getName()+")" ,c));
         }
      }
   }
   
   // cost and prune the plans
   for (PlanNode node : outputPlans) {
      estimator.costOperator(node);
   }
   prunePlanAlternatives(outputPlans);
   this.cachedPlans = outputPlans;
   return outputPlans;
}

从这个代码看到，getAlternativePlans()又是一个递归的遍历，是后续递归遍历，从上（source）到下（sink）的去创建PlanNode和Channel。getAlternativePlans()的核心就是创建PlanNode和Channel

Channel的数据结构如下，比较重要的两个参数就是ShipStrategyType和DataExchangeMode

public class Channel implements EstimateProvider, Cloneable, DumpableConnection<PlanNode> {
   
   private PlanNode source;
   
   private PlanNode target;
   private ShipStrategyType shipStrategy = ShipStrategyType.NONE;
   private DataExchangeMode dataExchangeMode;
   
   private LocalStrategy localStrategy = LocalStrategy.NONE;
   
   private FieldList shipKeys;
   
   private FieldList localKeys;
   
   private boolean[] shipSortOrder;
   
   private boolean[] localSortOrder;

我们先来分析一下sink端创建Channel，并对channel进行参数化赋值的过程，重点在RequestedGlobalProperties.parameterizeChannel()方法。parameterizeChannel()方法就是给Channel设置ShipStrategyType和DataExchangeMode

public void parameterizeChannel(Channel channel, boolean globalDopChange,
                        ExecutionMode exchangeMode, boolean breakPipeline) {
   ...
   // if we request nothing, then we need no special strategy. forward, if the number of instances remains
   // the same, randomly repartition otherwise
   //这些一般对应了MapNode、FilterNode等
   if (isTrivial() || this.partitioning == PartitioningProperty.ANY_DISTRIBUTION) {
      ShipStrategyType shipStrategy = globalDopChange ? ShipStrategyType.PARTITION_RANDOM :
                                             ShipStrategyType.FORWARD;
      DataExchangeMode em = DataExchangeMode.select(exchangeMode, shipStrategy, breakPipeline);
      channel.setShipStrategy(shipStrategy, em);
      return;
   }
   
   final GlobalProperties inGlobals = channel.getSource().getGlobalProperties();
   // if we have no global parallelism change, check if we have already compatible global properties
   //DataSinkNode、GroupCombineNode会走这里
   if (!globalDopChange && isMetBy(inGlobals)) {
      DataExchangeMode em = DataExchangeMode.select(exchangeMode, ShipStrategyType.FORWARD, breakPipeline);
      channel.setShipStrategy(ShipStrategyType.FORWARD, em);
      return;
   }
   
   // if we fall through the conditions until here, we need to re-establish
   ShipStrategyType shipType;
   FieldList partitionKeys;
   boolean[] sortDirection;
   Partitioner<?> partitioner;
   switch (this.partitioning) {
      case FULL_REPLICATION:
         shipType = ShipStrategyType.BROADCAST;
         partitionKeys = null;
         sortDirection = null;
         partitioner = null;
         break;
      case ANY_PARTITIONING:
          //如果是ANY_PARTITIONING就直接执行HASH_PARTITIONED的步骤了，GroupReduceNode会走这里
      case HASH_PARTITIONED:
         shipType = ShipStrategyType.PARTITION_HASH;
         partitionKeys = Utils.createOrderedFromSet(this.partitioningFields);
         sortDirection = null;
         partitioner = null;
         break;
      
      case RANGE_PARTITIONED:
         shipType = ShipStrategyType.PARTITION_RANGE;
         partitionKeys = this.ordering.getInvolvedIndexes();
         sortDirection = this.ordering.getFieldSortDirections();
         partitioner = null;
         if (this.dataDistribution != null) {
            channel.setDataDistribution(this.dataDistribution);
         }
         break;
      case FORCED_REBALANCED:
         shipType = ShipStrategyType.PARTITION_FORCED_REBALANCE;
         partitionKeys = null;
         sortDirection = null;
         partitioner = null;
         break;
      case CUSTOM_PARTITIONING:
         shipType = ShipStrategyType.PARTITION_CUSTOM;
         partitionKeys = Utils.createOrderedFromSet(this.partitioningFields);
         sortDirection = null;
         partitioner = this.customPartitioner;
         break;
      default:
         throw new CompilerException("Invalid partitioning to create through a data exchange: "
                              + this.partitioning.name());
   }
   DataExchangeMode exMode = DataExchangeMode.select(exchangeMode, shipType, breakPipeline);
   channel.setShipStrategy(shipType, partitionKeys, sortDirection, partitioner, exMode);
}

通过代码可以看到，Channel的ShipStrategyType和DataExchangeMode跟当前节点的partitioning属性和程序设置的ExecutionMode模式有关。对于像MapNode、FilterNode、FlatMapNode的partitioning属性为PartitioningProperty.ANY_DISTRIBUTION，GroupCombineNode、DataSinkNode它的partitioning是PartitioningProperty.RANDOM_PARTITIONED，GroupReduceNode它的partitioning是PartitioningProperty.ANY_PARTITIONING。

这种情况下MapNode、FilterNode、FlatMapNode、GroupCombineNode、DataSinkNode的ShipStrategyType都是FORWARD，GroupReduceNode的ShipStrategyType是PARTITION_HASH

具体DataExchangeMode的选择代码如下，可以看到，即使我们设置了ExecutionMode，最终的DataExchangeMode也不一定就和ExecutionMode一样，它还跟ShipStrategyType有关，比如DataSink，即使我们设置了ExecutionMode=BATCH，最终DataExchangeMode也还是PIPELINED

//DataExchangeMode
public static DataExchangeMode select(ExecutionMode executionMode, ShipStrategyType shipStrategy,
                                    boolean breakPipeline) {
   if (shipStrategy == null || shipStrategy == ShipStrategyType.NONE) {
      throw new IllegalArgumentException("shipStrategy may not be null or NONE");
   }
   if (executionMode == null) {
      throw new IllegalArgumentException("executionMode may not mbe null");
   }
   if (breakPipeline) {
      return getPipelineBreakingExchange(executionMode);
   }
   else if (shipStrategy == ShipStrategyType.FORWARD) {
      return getForForwardExchange(executionMode);
   }
   else {
      return getForShuffleOrBroadcast(executionMode);
   }
}
public static DataExchangeMode getForForwardExchange(ExecutionMode mode) {
   return FORWARD[mode.ordinal()];
}
public static DataExchangeMode getForShuffleOrBroadcast(ExecutionMode mode) {
   return SHUFFLE[mode.ordinal()];
}
public static DataExchangeMode getPipelineBreakingExchange(ExecutionMode mode) {
   return BREAKING[mode.ordinal()];
}
private static final DataExchangeMode[] FORWARD = new DataExchangeMode[ExecutionMode.values().length];
private static final DataExchangeMode[] SHUFFLE = new DataExchangeMode[ExecutionMode.values().length];
private static final DataExchangeMode[] BREAKING = new DataExchangeMode[ExecutionMode.values().length];
// initialize the map between execution modes and exchange modes in
static {
   FORWARD[ExecutionMode.PIPELINED_FORCED.ordinal()] = PIPELINED;
   SHUFFLE[ExecutionMode.PIPELINED_FORCED.ordinal()] = PIPELINED;
   BREAKING[ExecutionMode.PIPELINED_FORCED.ordinal()] = PIPELINED;
   FORWARD[ExecutionMode.PIPELINED.ordinal()] = PIPELINED;
   SHUFFLE[ExecutionMode.PIPELINED.ordinal()] = PIPELINED;
   BREAKING[ExecutionMode.PIPELINED.ordinal()] = BATCH;
   FORWARD[ExecutionMode.BATCH.ordinal()] = PIPELINED;
   SHUFFLE[ExecutionMode.BATCH.ordinal()] = BATCH;
   BREAKING[ExecutionMode.BATCH.ordinal()] = BATCH;
   FORWARD[ExecutionMode.BATCH_FORCED.ordinal()] = BATCH;
   SHUFFLE[ExecutionMode.BATCH_FORCED.ordinal()] = BATCH;
   BREAKING[ExecutionMode.BATCH_FORCED.ordinal()] = BATCH;
}

既然是递归向上调用，那我们再来看看SingleInputNode的getAlternativePlans()

public List<PlanNode> getAlternativePlans(CostEstimator estimator) {
   // check if we have a cached version
   if (this.cachedPlans != null) {
      return this.cachedPlans;
   }
   boolean childrenSkippedDueToReplicatedInput = false;
   // calculate alternative sub-plans for predecessor
   //也是向上递归的调用，先获取父节点对应的PlanNode
   final List<? extends PlanNode> subPlans = getPredecessorNode().getAlternativePlans(estimator);
   final Set<RequestedGlobalProperties> intGlobal = this.inConn.getInterestingProperties().getGlobalProperties();
   
    ...
    
   final ArrayList<PlanNode> outputPlans = new ArrayList<PlanNode>();
   final ExecutionMode executionMode = this.inConn.getDataExchangeMode();
   final int parallelism = getParallelism();
   final int inParallelism = getPredecessorNode().getParallelism();
   final boolean parallelismChange = inParallelism != parallelism;
   final boolean breaksPipeline = this.inConn.isBreakingPipeline();
   // create all candidates
   for (PlanNode child : subPlans) {
      ...
      if (this.inConn.getShipStrategy() == null) {
         // pick the strategy ourselves
         for (RequestedGlobalProperties igps: intGlobal) {
             //创建Channel并参数化
            final Channel c = new Channel(child, this.inConn.getMaterializationMode());
            igps.parameterizeChannel(c, parallelismChange, executionMode, breaksPipeline);
            
            ...
            
            for (RequestedGlobalProperties rgps: allValidGlobals) {
               if (rgps.isMetBy(c.getGlobalProperties())) {
                  c.setRequiredGlobalProps(rgps);
                  //创建当前节点对应的PlanNode，添加到outputPlans中
                  addLocalCandidates(c, broadcastPlanChannels, igps, outputPlans, estimator);
                  break;
               }
            }
         }
      } else {
        ...
      }
   }
  ...
   return outputPlans;
}

前面都一样，都是在获取到父节点的PlanNode之后，创建Channel，给Channel设置ShipStrategyType和ExecutionMode。创建PlanNode的过程在addLocalCandidates()中，addLocalCandidates()最终都会调用每个SingleInputNode中OperatorDescriptorSingle.instantiate()方法。

//MapDescriptor
public SingleInputPlanNode instantiate(Channel in, SingleInputNode node) {
   return new SingleInputPlanNode(node, "Map ("+node.getOperator().getName()+")", in, DriverStrategy.MAP);
}
//FlatMapDescriptor
public SingleInputPlanNode instantiate(Channel in, SingleInputNode node) {
   return new SingleInputPlanNode(node, "FlatMap ("+node.getOperator().getName()+")", in, DriverStrategy.FLAT_MAP);
}
//FilterDescriptor
public SingleInputPlanNode instantiate(Channel in, SingleInputNode node) {
   return new SingleInputPlanNode(node, "Filter ("+node.getOperator().getName()+")", in, DriverStrategy.FLAT_MAP);
}