Flink chain conditional source code analysis

flink In order to improve the efficiency of execution , Will merge some operators , Composition after merger operator chain, So the whole operator chain As a JobVertex Be dispatched to a slot Go ahead and execute , It avoids the network overhead caused by upstream and downstream operators when sending data and the performance loss of data serialization and deserialization .
flink How to judge whether an operator can form a operate chain Well ？ The answer is in the source code StreamingJobGraphGenerator.java in ：

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();
    }

1. downStreamVertex.getInEdges().size() == 1： The entry edge of the downstream operator must be 1, The downstream operator cannot be connect,union,join, The edges of these operators are greater than 1, Because he needs to access two streams
2. upStreamVertex.isSameSlotSharingGroup(downStreamVertex)： Both upstream and downstream operators are in the same SlotSharingGroup： Every operator can pass .slotSharingGroup() To specify the SlotSharingGroup, The default is not to specify Default, If you want to form upstream and downstream operators operater chain, Upstream and downstream SlotSharingGroup It has to be the same , Or none of them , Or set it to the same value
3. outOperator.getChainingStrategy() == ChainingStrategy.ALWAYS && (headOperator.getChainingStrategy() == ChainingStrategy.HEAD || headOperator.getChainingStrategy() == ChainingStrategy.ALWAYS)： Downstream operator ChainingStrategy The strategy must be ALWAYS, Upstream operator ChainingStrategy The strategy must be ALWAYS perhaps HEAD： If the operator calls disableChaining() Words , Of this operator ChainingStrategy Will become NEVER, It cannot be composed of any upstream and downstream operators operator chain; If the operator calls startNewChain(), So this operator's ChainingStrategy It becomes HEAD; Cannot be formed with upstream operators chain, It can only be composed of downstream operators chain.Source Operator defaults to HEAD, Other operators default to ALWAYS
4. edge.getPartitioner() instanceof ForwardPartitioner： The data partition mode between upstream and downstream operators is ForwardPartitioner： There are several ways to partition data ：shuffle,rebalance,rescale,broadcast,keyedgroup,forward etc. . If there is no display between upstream and downstream operators, call other partition modes （keyby,shuffle,rebalance,rescale,broadcast） And the concurrency of upstream and downstream operators is the same , The default is ForwardPartitioner Pattern
5. edge.getShuffleMode() != ShuffleMode.BATCH： If not specified, the default is PIPELINED
6. upStreamVertex.getParallelism() == downStreamVertex.getParallelism()： The concurrency of upstream and downstream operators must be the same
7. streamGraph.isChainingEnabled()：job There is no call disableChaining()

original text ：flink operater chain - Simple books