Operator Chains的生成

Chaining operators together into tasks is a useful optimization: it reduces the overhead of thread-to-thread handover and buffering, and increases overall throughput while decreasing latency.

上面是官方对Operator Chains的解释，以及示意图，那么看到这个图的时候会产生一个疑问，什么场景什么样的算子会产生Operator Chains？本文将详细解答这个疑问。

Flink job执行作业过程

下面是一段样例代码，传送门

public class TaskDemo {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
        env.setParallelism(1);

        env.addSource(new DataSource())
                .map(new MyMapFunction())
                .keyBy(0)
                .process(new MyKeyedProcessFunction())
                .addSink(new DataSink()).setParallelism(1).name("Custom Sink");

        env.execute();
    }
}

从这样一段Program到flink真正的执行，会经过如下一个流程
Program-> StreamGraph-> JobGraph-> ExecutionGraph，那么生成Operator Chains就是在StreamGraph-> JobGraph这个阶段。

Operator Chains生成规则

StreamingJobGraphGenerator这个类，是负责将StreamGraph转成JobGraph。其中isChainable方法就是用来判断operator之间是否可以构成chain。

public static boolean isChainable(StreamEdge edge, StreamGraph streamGraph) {
  StreamNode upStreamVertex = streamGraph.getSourceVertex(edge);
  StreamNode downStreamVertex = streamGraph.getTargetVertex(edge);

  StreamOperatorFactory<?> headOperator = upStreamVertex.getOperatorFactory();
  StreamOperatorFactory<?> outOperator = downStreamVertex.getOperatorFactory();

  return downStreamVertex.getInEdges().size() == 1
    && outOperator != null
    && headOperator != null
    && upStreamVertex.isSameSlotSharingGroup(downStreamVertex)
    && outOperator.getChainingStrategy() == ChainingStrategy.ALWAYS
    && (headOperator.getChainingStrategy() == ChainingStrategy.HEAD ||
        headOperator.getChainingStrategy() == ChainingStrategy.ALWAYS)
    && (edge.getPartitioner() instanceof ForwardPartitioner)
    && edge.getShuffleMode() != ShuffleMode.BATCH
    && upStreamVertex.getParallelism() == downStreamVertex.getParallelism()
    && streamGraph.isChainingEnabled();
}

可以看到是否可以组成chain判断了10个条件，有任何一个不满足，两个operator就不能构成一个chain。在分析这10个条件之前，读者需要明白，在flink中把程序转化成了一个有向图，这个图的顶点就是每个operator，边作为operator之间的连接关系有很多的属性，比如outputPartitioner等。
1. downStreamVertex.getInEdges().size() == 1下游算子的入边只有一个，也就是说它的上游的算子只能有一个。比如上面的样例程序，map的上游算子是source，而且只有一个source，就满足这个条件。如果是下面的程序，map的上游有两个source，即downStreamVertex.getInEdges()=2，这样source和map两个算子就不能构成一个chain。

public class UnionDemo {
    public static void main(String[] args) throws Exception{
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
        env.setParallelism(1);

        DataStream<Tuple2<String, Integer>> orangeStream = env.addSource(new DataSource("orangeStream"));
        DataStream<Tuple2<String, Integer>> greenStream = env.addSource(new DataSource("greenStream"));

        orangeStream.union(greenStream).map(new MapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(Tuple2<String, Integer> value) throws Exception {
                return value;
            }
        }).print("union");
        env.execute("Union Demo");
    }
}

2.outOperator != null表示下游顶点的算子不能为null，例如TaskDemo样例中map顶点的算子就是StreamMap。

3.headOperator != null表示上游顶点的算子不能为null，例如TaskDemo样例中source顶点的算子就是StreamSource。

4.upStreamVertex.isSameSlotSharingGroup(downStreamVertex)这个条件的意思是两个顶点在同一个槽位共享组中。在StreamGraphGenerator#determineSlotSharingGroup中确定了槽位共享组

/**
     * Determines the slot sharing group for an operation based on the slot sharing group set by
     * the user and the slot sharing groups of the inputs.
     *
     * <p>If the user specifies a group name, this is taken as is. If nothing is specified and
     * the input operations all have the same group name then this name is taken. Otherwise the
     * default group is chosen.
     *
     * @param specifiedGroup The group specified by the user.
     * @param inputIds The IDs of the input operations.
     */
private String determineSlotSharingGroup(String specifiedGroup, Collection<Integer> inputIds) {
  if (!isSlotSharingEnabled) {
    return null;
  }

  if (specifiedGroup != null) {
    return specifiedGroup;
  } else {
    String inputGroup = null;
    for (int id: inputIds) {
      String inputGroupCandidate = streamGraph.getSlotSharingGroup(id);
      if (inputGroup == null) {
        inputGroup = inputGroupCandidate;
      } else if (!inputGroup.equals(inputGroupCandidate)) {
        return DEFAULT_SLOT_SHARING_GROUP;
      }
    }
    return inputGroup == null ? DEFAULT_SLOT_SHARING_GROUP : inputGroup;
  }
}

官方对槽位共享组的描述如下：

Set the slot sharing group of an operation. Flink will put operations with the same slot sharing group into the same slot while keeping operations that don't have the slot sharing group in other slots. This can be used to isolate slots. The slot sharing group is inherited from input operations if all input operations are in the same slot sharing group. The name of the default slot sharing group is "default", operations can explicitly be put into this group by calling slotSharingGroup("default").

5.outOperator.getChainingStrategy() == ChainingStrategy.ALWAYS下游顶点的算子连接策略是ALWAYS，默认是ALWAYS，如果算子调用了disableChaining()，会设置为NEVER；如果算子调用了startNewChain()，会设置为HEAD。

6.(headOperator.getChainingStrategy()==ChainingStrategy.HEAD || headOperator.getChainingStrategy()==ChainingStrategy.ALWAYS)，和上一条一样，上游顶点的算子连接策略必须是HEAD或者ALWAYS。

7.(edge.getPartitioner() instanceof ForwardPartitioner)这个表示边的Partitioner必须是ForwardPartitioner，这个是在生成StreamGraph的时候，在StreamGraph#addEdgeInternal中确定的，详细的过程可以参见这个方法。

8.edge.getShuffleMode() != ShuffleMode.BATCH，shuffle模式不能是BATCH，一共有3种模式，shuffle模式的确定也是在StreamGraph#addEdgeInternal中。

/**
 * The shuffle mode defines the data exchange mode between operators.
 */
@PublicEvolving
public enum ShuffleMode {
    /**
     * Producer and consumer are online at the same time.
     * Produced data is received by consumer immediately.
     */
    PIPELINED,

    /**
     * The producer first produces its entire result and finishes.
     * After that, the consumer is started and may consume the data.
     */
    BATCH,

    /**
     * The shuffle mode is undefined. It leaves it up to the framework to decide the shuffle mode.
     * The framework will pick one of {@link ShuffleMode#BATCH} or {@link ShuffleMode#PIPELINED} in
     * the end.
     */
    UNDEFINED
}

9.upStreamVertex.getParallelism() == downStreamVertex.getParallelism()上游顶点的并行度和下游的要一样。例如上面TaskDemo代码中source和map的并行度都是1，所以可以构成一个chain，如果将map设置为2，那么他们就不能构成一个chain。

10.streamGraph.isChainingEnabled()默认这个值是true，如果调用了StreamExecutionEnvironment.disableOperatorChaining()那么streamGraph.isChainingEnabled()返回值就是false。

总结

本文详细的分析了算子之间可以生成operator chain的条件，对于Partitioner和ShuffleMode并没有展开说明，后续文章会对这部分进行补充。