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