Flink 原始碼閱讀筆記（3）- ExecutionGraph 的生成

private final JobVertex jobVertex;

private final List<OperatorID> operatorIDs;
private final List<OperatorID> userDefinedOperatorIds;

//ExecutionVertex 對應一個並行的子任務
private final ExecutionVertex[] taskVertices;

private final IntermediateResult[] producedDataSets;

private final List<IntermediateResult> inputs;

private final int parallelism;

private final SlotSharingGroup slotSharingGroup;

private final CoLocationGroup coLocationGroup;

private final InputSplit[] inputSplits;

private int maxParallelism;

private final ExecutionJobVertex jobVertex;

private final Map<IntermediateResultPartitionID, IntermediateResultPartition> resultPartitions;

private final ExecutionEdge[][] inputEdges;

private final int subTaskIndex;

private final EvictingBoundedList<ArchivedExecution> priorExecutions;

private volatile CoLocationConstraint locationConstraint;

/** The current or latest execution attempt of this vertex's task. */
private volatile Execution currentExecution;// this field must never be null

//對應的IntermediateDataSet的ID
private final IntermediateDataSetID id;
//生產者
private final ExecutionJobVertex producer;

//對應ExecutionJobVertex的並行度
private final int numParallelProducers;

private final IntermediateResultPartition[] partitions = new IntermediateResultPartition[numParallelProducers];

private final ResultPartitionType resultType;

ExecutionJobVertex -->IntermediateResult

ExecutionVertex -->IntermediateResultPartition

private final IntermediateResultPartition source;

private final ExecutionVertex target;

private final int inputNum;

/** Job specific information like the job id, job name, job configuration, etc. */
private final JobInformation jobInformation;

/** The slot provider to use for allocating slots for tasks as they are needed. */
private final SlotProvider slotProvider;

/** The classloader for the user code. Needed for calls into user code classes. */
private final ClassLoader userClassLoader;

/** All job vertices that are part of this graph. */
private final ConcurrentHashMap<JobVertexID, ExecutionJobVertex> tasks;

/** All vertices, in the order in which they were created. **/
private final List<ExecutionJobVertex> verticesInCreationOrder;

/** All intermediate results that are part of this graph. */
private final ConcurrentHashMap<IntermediateDataSetID, IntermediateResult> intermediateResults;

/** Current status of the job execution. */
private volatile JobStatus state = JobStatus.CREATED;

 /** Listeners that receive messages when the entire job switches it status
 * (such as from RUNNING to FINISHED). */
private final List<JobStatusListener> jobStatusListeners;

/** Listeners that receive messages whenever a single task execution changes its status. */
private final List<ExecutionStatusListener> executionListeners;

for (JobVertex vertex : jobGraph.getVertices()) {
....
try {
vertex.initializeOnMaster(classLoader);
}
catch (Throwable t) {
throw new JobExecutionException(jobId,
"Cannot initialize task '" + vertex.getName() + "': " + t.getMessage(), t);
}
}

//topologically sort the job vertices and attach the graph to the existing one
List<JobVertex> sortedTopology = jobGraph.getVerticesSortedTopologicallyFromSources();
if (log.isDebugEnabled()) {
log.debug("Adding {} vertices from job graph {} ({}).", sortedTopology.size(), jobName, jobId);
}
executionGraph.attachJobGraph(sortedTopology);

for (JobVertex jobVertex : topologiallySorted) {

if (jobVertex.isInputVertex() && !jobVertex.isStoppable()) {
this.isStoppable = false;
}

// create the execution job vertex and attach it to the graph
//建立 ExecutionJobVertex
ExecutionJobVertex ejv = new ExecutionJobVertex(
this,
jobVertex,
1,
rpcTimeout,
globalModVersion,
createTimestamp);

//連線上游節點
ejv.connectToPredecessors(this.intermediateResults);

ExecutionJobVertex previousTask = this.tasks.putIfAbsent(jobVertex.getID(), ejv);
if (previousTask != null) {
throw new JobException(String.format("Encountered two job vertices with ID %s : previous=[%s] / new=[%s]",
jobVertex.getID(), ejv, previousTask));
}

for (IntermediateResult res : ejv.getProducedDataSets()) {
IntermediateResult previousDataSet = this.intermediateResults.putIfAbsent(res.getId(), res);
if (previousDataSet != null) {
throw new JobException(String.format("Encountered two intermediate data set with ID %s : previous=[%s] / new=[%s]",
res.getId(), res, previousDataSet));
}
}

this.verticesInCreationOrder.add(ejv);
this.numVerticesTotal += ejv.getParallelism();
newExecJobVertices.add(ejv);
}