final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<Tuple2<Long, Long>> stream = env.addSource(...);
stream
    .map(new MapFunction<Integer, Integer>() {...})
    .addSink(new SinkFunction<Tuple2<Long, Long>>() {...});

env.execute();

    public <OUT> DataStreamSource<OUT> addSource(SourceFunction<OUT> function, String sourceName, TypeInformation<OUT> typeInfo) {

        if(typeInfo == null) { //如果没有指定typeInfo，做类型推断
            if (function instanceof ResultTypeQueryable) {
                typeInfo = ((ResultTypeQueryable<OUT>) function).getProducedType();
            } else {
                try {
                    typeInfo = TypeExtractor.createTypeInfo(
                            SourceFunction.class,
                            function.getClass(), 0, null, null);
                } catch (final InvalidTypesException e) {
                    typeInfo = (TypeInformation<OUT>) new MissingTypeInfo(sourceName, e);
                }
            }
        }

        boolean isParallel = function instanceof ParallelSourceFunction;

        clean(function);
        StreamSource<OUT, ?> sourceOperator;
        if (function instanceof StoppableFunction) {
            sourceOperator = new StoppableStreamSource<>(cast2StoppableSourceFunction(function));
        } else {
            sourceOperator = new StreamSource<>(function); //将SourceFunction封装成StreamSource
        }

        return new DataStreamSource<>(this, typeInfo, sourceOperator, isParallel, sourceName); //将StreamSource封装成DataStreamSource
    }

public class StreamSource<OUT, SRC extends SourceFunction<OUT>> 
        extends AbstractUdfStreamOperator<OUT, SRC> implements StreamOperator<OUT> {
    
    private transient SourceFunction.SourceContext<OUT> ctx; //用于collect output

    private transient volatile boolean canceledOrStopped = false;
    
    
    public StreamSource(SRC sourceFunction) {
        super(sourceFunction);

        this.chainingStrategy = ChainingStrategy.HEAD; //Source只能做Chaining Head
    }
    
    public void run(final Object lockingObject, final Output<StreamRecord<OUT>> collector) throws Exception {
        final TimeCharacteristic timeCharacteristic = getOperatorConfig().getTimeCharacteristic();

        LatencyMarksEmitter latencyEmitter = null;  //latencyMarker的相关逻辑
        if(getExecutionConfig().isLatencyTrackingEnabled()) {
            latencyEmitter = new LatencyMarksEmitter<>(
                getProcessingTimeService(),
                collector,
                getExecutionConfig().getLatencyTrackingInterval(),
                getOperatorConfig().getVertexID(),
                getRuntimeContext().getIndexOfThisSubtask());
        }
        
        final long watermarkInterval = getRuntimeContext().getExecutionConfig().getAutoWatermarkInterval();

        this.ctx = StreamSourceContexts.getSourceContext(
            timeCharacteristic, getProcessingTimeService(), lockingObject, collector, watermarkInterval);

        try {
            userFunction.run(ctx); //调用souceFunction执行用户逻辑，source应该不停的发送，该函数不会结束

            // if we get here, then the user function either exited after being done (finite source)
            // or the function was canceled or stopped. For the finite source case, we should emit
            // a final watermark that indicates that we reached the end of event-time
            if (!isCanceledOrStopped()) {
                ctx.emitWatermark(Watermark.MAX_WATERMARK); //发出最大的waterMarker
            }
        } finally {

        }
    }

  public class DataStreamSource<T> extends SingleOutputStreamOperator<T> {

    boolean isParallel;

    public DataStreamSource(StreamExecutionEnvironment environment,
            TypeInformation<T> outTypeInfo, StreamSource<T, ?> operator,
            boolean isParallel, String sourceName) {
        super(environment, new SourceTransformation<>(sourceName, operator, outTypeInfo, environment.getParallelism()));

        this.isParallel = isParallel;
        if (!isParallel) {
            setParallelism(1);
        }
    }

  public class SingleOutputStreamOperator<T> extends DataStream<T> {

    protected SingleOutputStreamOperator(StreamExecutionEnvironment environment, StreamTransformation<T> transformation) {
        super(environment, transformation);
    }

    public <R> SingleOutputStreamOperator<R> map(MapFunction<T, R> mapper) {

        TypeInformation<R> outType = TypeExtractor.getMapReturnTypes(clean(mapper), getType(),
                Utils.getCallLocationName(), true);

        return transform("Map", outType, new StreamMap<>(clean(mapper)));
    }

public class StreamMap<IN, OUT>
        extends AbstractUdfStreamOperator<OUT, MapFunction<IN, OUT>>
        implements OneInputStreamOperator<IN, OUT> {

    public StreamMap(MapFunction<IN, OUT> mapper) {
        super(mapper);
        chainingStrategy = ChainingStrategy.ALWAYS; //对于map而已，永远是可以chain的
    }

    @Override
    public void processElement(StreamRecord<IN> element) throws Exception {
        output.collect(element.replace(userFunction.map(element.getValue()))); //map的逻辑就执行mapFunc，并替换原有的element
    }
}

public <R> SingleOutputStreamOperator<R> transform(String operatorName, TypeInformation<R> outTypeInfo, OneInputStreamOperator<T, R> operator) {

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

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

        getExecutionEnvironment().addOperator(resultTransform);

        return returnStream;
    }

    protected final List<StreamTransformation<?>> transformations = new ArrayList<>();

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

    public DataStreamSink<T> addSink(SinkFunction<T> sinkFunction) {

        // configure the type if needed
        if (sinkFunction instanceof InputTypeConfigurable) {
            ((InputTypeConfigurable) sinkFunction).setInputType(getType(), getExecutionConfig() );
        }

        StreamSink<T> sinkOperator = new StreamSink<>(clean(sinkFunction));

        DataStreamSink<T> sink = new DataStreamSink<>(this, sinkOperator);

        getExecutionEnvironment().addOperator(sink.getTransformation());
        return sink;
    }

public class StreamSink<IN> extends AbstractUdfStreamOperator<Object, SinkFunction<IN>>
        implements OneInputStreamOperator<IN, Object> {

    public StreamSink(SinkFunction<IN> sinkFunction) {
        super(sinkFunction);
        chainingStrategy = ChainingStrategy.ALWAYS; //对于sink也是永远可以chain的
    }

    @Override
    public void processElement(StreamRecord<IN> element) throws Exception {
        userFunction.invoke(element.getValue());
    }

    @Override
    protected void reportOrForwardLatencyMarker(LatencyMarker maker) {
        // all operators are tracking latencies
        this.latencyGauge.reportLatency(maker, true);

        // sinks don't forward latency markers
    }
}

  public class DataStreamSink<T> {

    SinkTransformation<T> transformation;

    @SuppressWarnings("unchecked")
    protected DataStreamSink(DataStream<T> inputStream, StreamSink<T> operator) {
        this.transformation = new SinkTransformation<T>(inputStream.getTransformation(), "Unnamed", operator, inputStream.getExecutionEnvironment().getParallelism());
    }

public JobExecutionResult execute(String jobName) throws ProgramInvocationException {
    StreamGraph streamGraph = getStreamGraph();
    streamGraph.setJobName(jobName);
    transformations.clear();
    return executeRemotely(streamGraph);
}

public StreamGraph getStreamGraph() {
    if (transformations.size() <= 0) {
        throw new IllegalStateException("No operators defined in streaming topology. Cannot execute.");
    }
    return StreamGraphGenerator.generate(this, transformations);
}

public class StreamGraphGenerator {

    // The StreamGraph that is being built, this is initialized at the beginning.
    private StreamGraph streamGraph;

    private final StreamExecutionEnvironment env;

    // Keep track of which Transforms we have already transformed, this is necessary because
    // we have loops, i.e. feedback edges.
    private Map<StreamTransformation<?>, Collection<Integer>> alreadyTransformed; //防止环，所以把transformed过的记下来


    /**
     * Private constructor. The generator should only be invoked using {@link #generate}.
     */
    private StreamGraphGenerator(StreamExecutionEnvironment env) {
        this.streamGraph = new StreamGraph(env);
        this.streamGraph.setChaining(env.isChainingEnabled());
        this.streamGraph.setStateBackend(env.getStateBackend());
        this.env = env;
        this.alreadyTransformed = new HashMap<>();
    }

    /**
     * Generates a {@code StreamGraph} by traversing the graph of {@code StreamTransformations}
     * starting from the given transformations.
     *
     * @param env The {@code StreamExecutionEnvironment} that is used to set some parameters of the
     *            job
     * @param transformations The transformations starting from which to transform the graph
     *
     * @return The generated {@code StreamGraph}
     */
    public static StreamGraph generate(StreamExecutionEnvironment env, List<StreamTransformation<?>> transformations) {
        return new StreamGraphGenerator(env).generateInternal(transformations);
    }

    /**
     * This starts the actual transformation, beginning from the sinks.
     */
    private StreamGraph generateInternal(List<StreamTransformation<?>> transformations) {
        for (StreamTransformation<?> transformation: transformations) {
            transform(transformation);
        }
        return streamGraph;
    }

    private Collection<Integer> transform(StreamTransformation<?> transform) {

        if (alreadyTransformed.containsKey(transform)) {
            return alreadyTransformed.get(transform); //如果transform过，就直接返回
        }

        Collection<Integer> transformedIds;
        if (transform instanceof OneInputTransformation<?, ?>) {
            transformedIds = transformOnInputTransform((OneInputTransformation<?, ?>) transform);
        } else if (transform instanceof TwoInputTransformation<?, ?, ?>) {
            transformedIds = transformTwoInputTransform((TwoInputTransformation<?, ?, ?>) transform);
        } else if (transform instanceof SourceTransformation<?>) {
            transformedIds = transformSource((SourceTransformation<?>) transform);
        } else if (transform instanceof SinkTransformation<?>) {
            transformedIds = transformSink((SinkTransformation<?>) transform);
        } else if (transform instanceof UnionTransformation<?>) {
            transformedIds = transformUnion((UnionTransformation<?>) transform);
        } else if (transform instanceof SplitTransformation<?>) {
            transformedIds = transformSplit((SplitTransformation<?>) transform);
        } else if (transform instanceof SelectTransformation<?>) {
            transformedIds = transformSelect((SelectTransformation<?>) transform);
        } else if (transform instanceof FeedbackTransformation<?>) {
            transformedIds = transformFeedback((FeedbackTransformation<?>) transform);
        } else if (transform instanceof CoFeedbackTransformation<?>) {
            transformedIds = transformCoFeedback((CoFeedbackTransformation<?>) transform);
        } else if (transform instanceof PartitionTransformation<?>) {
            transformedIds = transformPartition((PartitionTransformation<?>) transform);
        }

        return transformedIds;
    }

/**
     * Transforms a {@code OneInputTransformation}.
     *
     * <p>
     * This recusively transforms the inputs, creates a new {@code StreamNode} in the graph and
     * wired the inputs to this new node.
     */
    private <IN, OUT> Collection<Integer> transformOnInputTransform(OneInputTransformation<IN, OUT> transform) {

        Collection<Integer> inputIds = transform(transform.getInput()); //递归调用transform，所以前面source没有加到transformations，因为这里会递归到

        // the recursive call might have already transformed this
        if (alreadyTransformed.containsKey(transform)) {
            return alreadyTransformed.get(transform); //如果已经transform过，直接返回
        }

        String slotSharingGroup = determineSlotSharingGroup(transform.getSlotSharingGroup(), inputIds); //产生slotSharingGroup

        streamGraph.addOperator(transform.getId(), //addOperator
                slotSharingGroup,
                transform.getOperator(),
                transform.getInputType(),
                transform.getOutputType(),
                transform.getName());

        if (transform.getStateKeySelector() != null) {
            TypeSerializer<?> keySerializer = transform.getStateKeyType().createSerializer(env.getConfig());
            streamGraph.setOneInputStateKey(transform.getId(), transform.getStateKeySelector(), keySerializer);
        }

        streamGraph.setParallelism(transform.getId(), transform.getParallelism());
        streamGraph.setMaxParallelism(transform.getId(), transform.getMaxParallelism());

        for (Integer inputId: inputIds) {
            streamGraph.addEdge(inputId, transform.getId(), 0); //addEdge
        }

        return Collections.singleton(transform.getId());
    }

public abstract class StreamTransformation<T> {

    // This is used to assign a unique ID to every StreamTransformation
    protected static Integer idCounter = 0;

    public static int getNewNodeId() {
        idCounter++;
        return idCounter;
    }


    protected final int id;
    
    public StreamTransformation(String name, TypeInformation<T> outputType, int parallelism) {
        this.id = getNewNodeId();

public abstract class StreamTransformation<T> {
    private String slotSharingGroup;
    public StreamTransformation(String name, TypeInformation<T> outputType, int parallelism) {
        this.slotSharingGroup = null;

   /**
     * Sets the slot sharing group of this operation. Parallel instances of
     * operations that are in the same slot sharing group will be co-located in the same
     * TaskManager slot, if possible.
     *
     * <p>Operations inherit the slot sharing group of input operations if all input operations
     * are in the same slot sharing group and no slot sharing group was explicitly specified.
     *
     * <p>Initially an operation is in the default slot sharing group. An operation can be put into
     * the default group explicitly by setting the slot sharing group to {@code "default"}.
     *
     * @param slotSharingGroup The slot sharing group name.
     */
    @PublicEvolving
    public DataStreamSink<T> slotSharingGroup(String slotSharingGroup) {
        transformation.setSlotSharingGroup(slotSharingGroup);
        return this;
    }

   /**
     * 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 choosen.
     *
     * @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 (specifiedGroup != null) { //如果用户指定，以用户指定为准
            return specifiedGroup;
        } else {
            String inputGroup = null;
            for (int id: inputIds) { //根据输入的SlotSharingGroup进行推断
                String inputGroupCandidate = streamGraph.getSlotSharingGroup(id);
                if (inputGroup == null) {
                    inputGroup = inputGroupCandidate; //初始化
                } else if (!inputGroup.equals(inputGroupCandidate)) { //逻辑如果所有input的SlotSharingGroup都相同，就用；否则就用“default”
                    return "default";
                }
            }
            return inputGroup == null ? "default" : inputGroup; //默认用default
        }
    }

    public <IN, OUT> void addOperator(
            Integer vertexID,
            String slotSharingGroup,
            StreamOperator<OUT> operatorObject,
            TypeInformation<IN> inTypeInfo,
            TypeInformation<OUT> outTypeInfo,
            String operatorName) {

        if (operatorObject instanceof StoppableStreamSource) {
            addNode(vertexID, slotSharingGroup, StoppableSourceStreamTask.class, operatorObject, operatorName);
        } else if (operatorObject instanceof StreamSource) {
            addNode(vertexID, slotSharingGroup, SourceStreamTask.class, operatorObject, operatorName);
        } else {
            addNode(vertexID, slotSharingGroup, OneInputStreamTask.class, operatorObject, operatorName);
        }

    protected StreamNode addNode(Integer vertexID,
        String slotSharingGroup,
        Class<? extends AbstractInvokable> vertexClass,
        StreamOperator<?> operatorObject,
        String operatorName) {

        if (streamNodes.containsKey(vertexID)) { //如果已经有vertexId
            throw new RuntimeException("Duplicate vertexID " + vertexID);
        }

        StreamNode vertex = new StreamNode(environment,
            vertexID,
            slotSharingGroup,
            operatorObject,
            operatorName,
            new ArrayList<OutputSelector<?>>(),
            vertexClass);

        streamNodes.put(vertexID, vertex);

        return vertex;
    }

    public void addEdge(Integer upStreamVertexID, Integer downStreamVertexID, int typeNumber) {
        addEdgeInternal(upStreamVertexID,
                downStreamVertexID,
                typeNumber,
                null,
                new ArrayList<String>());

    }

private void addEdgeInternal(Integer upStreamVertexID,
            Integer downStreamVertexID,
            int typeNumber,
            StreamPartitioner<?> partitioner,
            List<String> outputNames) {


        if (virtualSelectNodes.containsKey(upStreamVertexID)) { //如果是虚拟select节点
            int virtualId = upStreamVertexID;
            upStreamVertexID = virtualSelectNodes.get(virtualId).f0; //由于不是真实节点，所以以虚拟节点的父节点为父节点
            if (outputNames.isEmpty()) {
                // selections that happen downstream override earlier selections
                outputNames = virtualSelectNodes.get(virtualId).f1; //将select虚拟节点，转换为outputNames
            }
            addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, outputNames);//递归的调用addEdgeInternal
        } else if (virtualPartitionNodes.containsKey(upStreamVertexID)) {
            int virtualId = upStreamVertexID;
            upStreamVertexID = virtualPartitionNodes.get(virtualId).f0;
            if (partitioner == null) {
                partitioner = virtualPartitionNodes.get(virtualId).f1; //对于partition虚拟节点，转换为partitioner
            }
            addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, outputNames);//递归的调用addEdgeInternal
        } 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
                partitioner = new ForwardPartitioner<Object>(); //如果并发度相同则是forward
            } else if (partitioner == null) {
                partitioner = new RebalancePartitioner<Object>(); //如果并发度不同则是Rebalance
            }

            if (partitioner instanceof ForwardPartitioner) { //判断如果用户指定forward，而并发度不同，抛异常
                if (upstreamNode.getParallelism() != downstreamNode.getParallelism()) {
                    throw new UnsupportedOperationException("Forward partitioning does not allow " +
                            "change of parallelism. Upstream operation: " + upstreamNode + " parallelism: " + upstreamNode.getParallelism() +
                            ", downstream operation: " + downstreamNode + " parallelism: " + downstreamNode.getParallelism() +
                            " You must use another partitioning strategy, such as broadcast, rebalance, shuffle or global.");
                }
            }

            StreamEdge edge = new StreamEdge(upstreamNode, downstreamNode, typeNumber, outputNames, partitioner); //创建StreamEdge

            getStreamNode(edge.getSourceId()).addOutEdge(edge); //将上下游StreamNode用StreamEdge相连
            getStreamNode(edge.getTargetId()).addInEdge(edge);
        }
    }

/** 
 * The following graph of {@code StreamTransformations}:
 *
 * <pre>{@code
 *   Source              Source        
 *      +                        +           
 *      |                        |           
 *      v                        v           
 *  Rebalance          HashPartition    
 *      +                        +           
 *      |                        |           
 *      |                        |           
 *      +------>Union<------+           
 *                +                     
 *                |                     
 *                v                     
 *              Split                   
 *                +                     
 *                |                     
 *                v                     
 *              Select                  
 *                +                     
 *                v                     
 *               Map                    
 *                +                     
 *                |                     
 *                v                     
 *              Sink 
 * }</pre>
 *
 * Would result in this graph of operations at runtime:
 *
 * <pre>{@code
 *  Source              Source
 *    +                   +
 *    |                   |
 *    |                   |
 *    +------->Map<-------+
 *              +
 *              |
 *              v
 *             Sink
 * /

    private <T> Collection<Integer> transformPartition(PartitionTransformation<T> partition) {
        StreamTransformation<T> input = partition.getInput();
        List<Integer> resultIds = new ArrayList<>();

        Collection<Integer> transformedIds = transform(input); //递归transform父节点，并得到他们的id
        for (Integer transformedId: transformedIds) {
            int virtualId = StreamTransformation.getNewNodeId(); //产生自己的id
            streamGraph.addVirtualPartitionNode(transformedId, virtualId, partition.getPartitioner()); //只是注册到VirtualPartitionNode，而没有真正产生StreamNode
            resultIds.add(virtualId);
        }

        return resultIds;
    }

  private <T> Collection<Integer> transformUnion(UnionTransformation<T> union) {
        List<StreamTransformation<T>> inputs = union.getInputs();
        List<Integer> resultIds = new ArrayList<>();

        for (StreamTransformation<T> input: inputs) {
            resultIds.addAll(transform(input)); //递归
        }

        return resultIds;
    }

public JobExecutionResult execute(String jobName) throws ProgramInvocationException {
    StreamGraph streamGraph = getStreamGraph();
    streamGraph.setJobName(jobName);
    transformations.clear();
    return executeRemotely(streamGraph);
}

    protected JobExecutionResult executeRemotely(StreamGraph streamGraph, List<URL> jarFiles) throws ProgramInvocationException {
        ClusterClient client;
        try {
            return client.run(streamGraph, jarFiles, globalClasspaths, usercodeClassLoader).getJobExecutionResult();
        }
    }

  public JobSubmissionResult run(FlinkPlan compiledPlan,
            List<URL> libraries, List<URL> classpaths, ClassLoader classLoader, SavepointRestoreSettings savepointSettings)
        throws ProgramInvocationException
    {
        JobGraph job = getJobGraph(compiledPlan, libraries, classpaths, savepointSettings);
        return submitJob(job, classLoader);
    }

    private JobGraph getJobGraph(FlinkPlan optPlan, List<URL> jarFiles, List<URL> classpaths, SavepointRestoreSettings savepointSettings) {
        JobGraph job;
        if (optPlan instanceof StreamingPlan) { //如果是流job plan
            job = ((StreamingPlan) optPlan).getJobGraph();
            job.setSavepointRestoreSettings(savepointSettings);
        } else { //如果是batch
            JobGraphGenerator gen = new JobGraphGenerator(this.flinkConfig);
            job = gen.compileJobGraph((OptimizedPlan) optPlan);
        }

        for (URL jar : jarFiles) {
            try {
                job.addJar(new Path(jar.toURI())); //加入jar
            } catch (URISyntaxException e) {
                throw new RuntimeException("URL is invalid. This should not happen.", e);
            }
        }
 
        job.setClasspaths(classpaths); //加上classpath

        return job;
    }

    public JobGraph getJobGraph() {
        StreamingJobGraphGenerator jobgraphGenerator = new StreamingJobGraphGenerator(this);

        return jobgraphGenerator.createJobGraph();
    }

    public JobGraph createJobGraph() {

        jobGraph = new JobGraph(streamGraph.getJobName()); //创建JobGraph

        // make sure that all vertices start immediately
        jobGraph.setScheduleMode(ScheduleMode.EAGER); //对于流所有vertices需要立即启动，相对的模式，LAZY_FROM_SOURCES，task只有在input ready时，才会创建

        init(); //简单的结构new，初始化

        // Generate deterministic hashes for the nodes in order to identify them across
        // submission iff they didn't change.
        Map<Integer, byte[]> hashes = defaultStreamGraphHasher.traverseStreamGraphAndGenerateHashes(streamGraph); //为每个node创建唯一的hashid，这样多次提交时能够定位到，最终返回node id和hash id的对应


         setChaining(hashes, legacyHashes); //核心逻辑，创建JobVertex，JobEdge

        setPhysicalEdges(); //只是将每个vertex的入边信息，写入该vertex所对应的StreamConfig里面

        setSlotSharing();

        configureCheckpointing();

        // set the ExecutionConfig last when it has been finalized
        jobGraph.setExecutionConfig(streamGraph.getExecutionConfig());

        return jobGraph;
    }

    private void setChaining(Map<Integer, byte[]> hashes, List<Map<Integer, byte[]>> legacyHashes) {
        for (Integer sourceNodeId : streamGraph.getSourceIDs()) {
            createChain(sourceNodeId, sourceNodeId, hashes, legacyHashes, 0);
        }
    }

    private List<StreamEdge> createChain(
            Integer startNodeId,
            Integer currentNodeId,
            Map<Integer, byte[]> hashes,
            List<Map<Integer, byte[]>> legacyHashes,
            int chainIndex) {

        if (!builtVertices.contains(startNodeId)) {

            List<StreamEdge> transitiveOutEdges = new ArrayList<StreamEdge>();//最终要生成JobEdge的StreamingEdge

            List<StreamEdge> chainableOutputs = new ArrayList<StreamEdge>();
            List<StreamEdge> nonChainableOutputs = new ArrayList<StreamEdge>();

            for (StreamEdge outEdge : streamGraph.getStreamNode(currentNodeId).getOutEdges()) { //遍历当前Node的所有出边
                if (isChainable(outEdge, streamGraph)) { //判断是否可以chain，核心逻辑
                    chainableOutputs.add(outEdge);
                } else {
                    nonChainableOutputs.add(outEdge);
                }
            }

            for (StreamEdge chainable : chainableOutputs) { //对于chainable，递归调用下去
                transitiveOutEdges.addAll(
                        createChain(startNodeId, chainable.getTargetId(), hashes, legacyHashes, chainIndex + 1)); //currentNodeId设为targetNode的id，同时chainIndex加1
            }

            for (StreamEdge nonChainable : nonChainableOutputs) { //对于nonChainable
                transitiveOutEdges.add(nonChainable); //既然不是chained，就需要产生真正的JobEdge，所以放到transitiveOutEdges
                createChain(nonChainable.getTargetId(), nonChainable.getTargetId(), hashes, legacyHashes, 0); //继续，但注意这里startNodeId和currentNodeId都设为TargetId，因为当前的非chained，下一个需要开始新的chain
            }

            chainedNames.put(currentNodeId, createChainedName(currentNodeId, chainableOutputs)); //为每个chain生成name

            StreamConfig config = currentNodeId.equals(startNodeId)
                    ? createJobVertex(startNodeId, hashes, legacyHashes) //只有为chain中的startNode创建JobVertex，其他的只是创建空StreamConfig
                    : new StreamConfig(new Configuration());

            setVertexConfig(currentNodeId, config, chainableOutputs, nonChainableOutputs); //将StreamNode中的配置放到StreamConfig中

            if (currentNodeId.equals(startNodeId)) { //如果是startNode

                config.setChainStart();
                config.setChainIndex(0);
                config.setOperatorName(streamGraph.getStreamNode(currentNodeId).getOperatorName());
                config.setOutEdgesInOrder(transitiveOutEdges);
                config.setOutEdges(streamGraph.getStreamNode(currentNodeId).getOutEdges());

                for (StreamEdge edge : transitiveOutEdges) {
                    connect(startNodeId, edge); //只要startNode需要connect edge
                }

                config.setTransitiveChainedTaskConfigs(chainedConfigs.get(startNodeId));

            } else {

                Map<Integer, StreamConfig> chainedConfs = chainedConfigs.get(startNodeId);

                if (chainedConfs == null) {
                    chainedConfigs.put(startNodeId, new HashMap<Integer, StreamConfig>());
                }
                config.setChainIndex(chainIndex);
                config.setOperatorName(streamGraph.getStreamNode(currentNodeId).getOperatorName());
                chainedConfigs.get(startNodeId).put(currentNodeId, config);
            }
            if (chainableOutputs.isEmpty()) {
                config.setChainEnd();
            }

            return transitiveOutEdges;

        } else {
            return new ArrayList<>();
        }
    }

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

        StreamOperator<?> headOperator = upStreamVertex.getOperator();
        StreamOperator<?> outOperator = downStreamVertex.getOperator();

        return downStreamVertex.getInEdges().size() == 1 //终点的入边为1，如果多个输入，需要等其他输入，无法chain执行
                && outOperator != null
                && headOperator != null
                && upStreamVertex.isSameSlotSharingGroup(downStreamVertex) //在同一个SlotSharingGroup
                && outOperator.getChainingStrategy() == ChainingStrategy.ALWAYS //终点ChainingStrategy是Always
                && (headOperator.getChainingStrategy() == ChainingStrategy.HEAD ||
                    headOperator.getChainingStrategy() == ChainingStrategy.ALWAYS) //启动ChainingStrategy是Head或Always
                && (edge.getPartitioner() instanceof ForwardPartitioner) //Edge是ForwardPartitioner
                && upStreamVertex.getParallelism() == downStreamVertex.getParallelism() //起点和终点的并发度相同
                && streamGraph.isChainingEnabled(); //允许chain
    }

    private StreamConfig createJobVertex(
            Integer streamNodeId,
            Map<Integer, byte[]> hashes,
            List<Map<Integer, byte[]>> legacyHashes) {

        JobVertex jobVertex;
        StreamNode streamNode = streamGraph.getStreamNode(streamNodeId);

        byte[] hash = hashes.get(streamNodeId); //取出streamNode对应的唯一id

        JobVertexID jobVertexId = new JobVertexID(hash); //生成JobVertexID

        if (streamNode.getInputFormat() != null) {
            jobVertex = new InputFormatVertex(
                    chainedNames.get(streamNodeId),
                    jobVertexId,
                    legacyJobVertexIds);
            TaskConfig taskConfig = new TaskConfig(jobVertex.getConfiguration());
            taskConfig.setStubWrapper(new UserCodeObjectWrapper<Object>(streamNode.getInputFormat()));
        } else {
            jobVertex = new JobVertex(
                    chainedNames.get(streamNodeId),
                    jobVertexId,
                    legacyJobVertexIds);
        }

        jobVertex.setInvokableClass(streamNode.getJobVertexClass());

        int parallelism = streamNode.getParallelism();

        if (parallelism > 0) {
            jobVertex.setParallelism(parallelism); //设置并发度
        } else {
            parallelism = jobVertex.getParallelism();
        }

        jobVertex.setMaxParallelism(streamNode.getMaxParallelism());

        jobVertices.put(streamNodeId, jobVertex); //将jobVertex加到相应的结构中去
        builtVertices.add(streamNodeId);
        jobGraph.addVertex(jobVertex);

        return new StreamConfig(jobVertex.getConfiguration());
    }

    private void connect(Integer headOfChain, StreamEdge edge) {

        physicalEdgesInOrder.add(edge);//connect都是物理边，即会产生JobEdge

        Integer downStreamvertexID = edge.getTargetId();

        JobVertex headVertex = jobVertices.get(headOfChain);
        JobVertex downStreamVertex = jobVertices.get(downStreamvertexID);

        StreamConfig downStreamConfig = new StreamConfig(downStreamVertex.getConfiguration());

        downStreamConfig.setNumberOfInputs(downStreamConfig.getNumberOfInputs() + 1); //多一个入边，inputs + 1

        StreamPartitioner<?> partitioner = edge.getPartitioner();
        JobEdge jobEdge = null;
        if (partitioner instanceof ForwardPartitioner) {
            jobEdge = downStreamVertex.connectNewDataSetAsInput(
                headVertex,
                DistributionPattern.POINTWISE,
                ResultPartitionType.PIPELINED); //Streaming都是pipelining，即一有结果，consumer就会来拖
        } else if (partitioner instanceof RescalePartitioner){
            jobEdge = downStreamVertex.connectNewDataSetAsInput(
                headVertex,
                DistributionPattern.POINTWISE, //produer的subtask可以对应一个或多个consumer的tasks
                ResultPartitionType.PIPELINED);
        } else {
            jobEdge = downStreamVertex.connectNewDataSetAsInput(
                    headVertex,
                    DistributionPattern.ALL_TO_ALL, //producer和consumer的subtask，一对一
                    ResultPartitionType.PIPELINED);
        }
        // set strategy name so that web interface can show it.
        jobEdge.setShipStrategyName(partitioner.toString());
    }

public JobEdge connectNewDataSetAsInput(
            JobVertex input,
            DistributionPattern distPattern,
            ResultPartitionType partitionType) {

        IntermediateDataSet dataSet = input.createAndAddResultDataSet(partitionType); //创建IntermediateDataSet，并注册到inputVertex

        JobEdge edge = new JobEdge(dataSet, this, distPattern); //创建JobEdge
        this.inputs.add(edge); //把edge作为当前vertex的input
        dataSet.addConsumer(edge); //edge从IntermediateDataSet去数据
        return edge;
    }

    private void setSlotSharing() {

        Map<String, SlotSharingGroup> slotSharingGroups = new HashMap<>();

        for (Entry<Integer, JobVertex> entry : jobVertices.entrySet()) { //遍历每个JobVertex

            String slotSharingGroup = streamGraph.getStreamNode(entry.getKey()).getSlotSharingGroup();

            SlotSharingGroup group = slotSharingGroups.get(slotSharingGroup);
            if (group == null) {
                group = new SlotSharingGroup(); //初始化SlotSharingGroup
                slotSharingGroups.put(slotSharingGroup, group);
            }
            entry.getValue().setSlotSharingGroup(group); //把节点加入SlotSharingGroup
        }

        for (Tuple2<StreamNode, StreamNode> pair : streamGraph.getIterationSourceSinkPairs()) { //对于Iteration要创建CoLocationGroup

            CoLocationGroup ccg = new CoLocationGroup();

            JobVertex source = jobVertices.get(pair.f0.getId());
            JobVertex sink = jobVertices.get(pair.f1.getId());

            ccg.addVertex(source);
            ccg.addVertex(sink);
            source.updateCoLocationGroup(ccg);
            sink.updateCoLocationGroup(ccg);
        }

    }

    private void configureCheckpointing() {
        CheckpointConfig cfg = streamGraph.getCheckpointConfig();

        long interval = cfg.getCheckpointInterval();
        if (interval > 0) {  //只要设置过CheckpointInterval，默认设为fixedDelayRestart策略
            // check if a restart strategy has been set, if not then set the FixedDelayRestartStrategy
            if (streamGraph.getExecutionConfig().getRestartStrategy() == null) {
                // if the user enabled checkpointing, the default number of exec retries is infinite.
                streamGraph.getExecutionConfig().setRestartStrategy(
                    RestartStrategies.fixedDelayRestart(Integer.MAX_VALUE, DEFAULT_RESTART_DELAY));
            }
        } else {
            // interval of max value means disable periodic checkpoint
            interval = Long.MAX_VALUE;
        }

        // collect the vertices that receive "trigger checkpoint" messages.
        // currently, these are all the sources
        List<JobVertexID> triggerVertices = new ArrayList<>();

        // collect the vertices that need to acknowledge the checkpoint
        // currently, these are all vertices
        List<JobVertexID> ackVertices = new ArrayList<>(jobVertices.size()); //所以JobVertex都需要ack

        // collect the vertices that receive "commit checkpoint" messages
        // currently, these are all vertices
        List<JobVertexID> commitVertices = new ArrayList<>();

        for (JobVertex vertex : jobVertices.values()) {
            if (vertex.isInputVertex()) { //没有输入的Vertex
                triggerVertices.add(vertex.getID()); //加入triggerVertex
            }
            commitVertices.add(vertex.getID());
            ackVertices.add(vertex.getID());
        }

        CheckpointingMode mode = cfg.getCheckpointingMode();

        boolean isExactlyOnce;
        if (mode == CheckpointingMode.EXACTLY_ONCE) { //Checkpoint模式
            isExactlyOnce = true;
        } else if (mode == CheckpointingMode.AT_LEAST_ONCE) {
            isExactlyOnce = false;
        } else {
            throw new IllegalStateException("Unexpected checkpointing mode. " +
                "Did not expect there to be another checkpointing mode besides " +
                "exactly-once or at-least-once.");
        }

        JobSnapshottingSettings settings = new JobSnapshottingSettings(
                triggerVertices, ackVertices, commitVertices, interval,
                cfg.getCheckpointTimeout(), cfg.getMinPauseBetweenCheckpoints(),
                cfg.getMaxConcurrentCheckpoints(),
                externalizedCheckpointSettings,
                isExactlyOnce);

        jobGraph.setSnapshotSettings(settings);
    }