本文共 66331 字，大约阅读时间需要 221 分钟。
在Flink中的有以下几种图，即StreamGraph，它用来生成JobGraph，然后再由分发器将其分发生成ExecutionGraph并进而形成Task任务执行的任务图（这个图就只是一个逻辑概念了）。JobGraph是非常重要的一环，其它的图以后再详细分析。学过图的都知道，图和其它数据结构明显不同的是，图有节点和边的概念。
那么看一下Flink中这个图的定义：public class JobGraph implements Serializable { private static final long serialVersionUID = 1L; // --- job and configuration --- /** List of task vertices included in this job graph. */ private final Map taskVertices = new LinkedHashMap (); /** The job configuration attached to this job. */ private final Configuration jobConfiguration = new Configuration(); /** ID of this job. May be set if specific job id is desired (e.g. session management) */ private final JobID jobID; /** Name of this job. */ private final String jobName; /** The number of seconds after which the corresponding ExecutionGraph is removed at the * job manager after it has been executed. */ private long sessionTimeout = 0; /** flag to enable queued scheduling */ private boolean allowQueuedScheduling; /** The mode in which the job is scheduled */ private ScheduleMode scheduleMode = ScheduleMode.LAZY_FROM_SOURCES; // --- checkpointing --- /** Job specific execution config */ private SerializedValue serializedExecutionConfig; /** The settings for the job checkpoints */ private JobCheckpointingSettings snapshotSettings; /** Savepoint restore settings. */ private SavepointRestoreSettings savepointRestoreSettings = SavepointRestoreSettings.none(); // --- attached resources --- /** Set of JAR files required to run this job. */ private final List userJars = new ArrayList (); /** Set of custom files required to run this job. */ private final Map userArtifacts = new HashMap<>(); /** Set of blob keys identifying the JAR files required to run this job. */ private final List userJarBlobKeys = new ArrayList<>(); /** List of classpaths required to run this job. */ private List classpaths = Collections.emptyList(); // -------------------------------------------------------------------------------------------- /** * Constructs a new job graph with the given name, the given {@link ExecutionConfig}, * and a random job ID. The ExecutionConfig will be serialized and can't be modified afterwards. * * @param jobName The name of the job. */ public JobGraph(String jobName) { this(null, jobName); } /** * Constructs a new job graph with the given job ID (or a random ID, if {@code null} is passed), * the given name and the given execution configuration (see {@link ExecutionConfig}). * The ExecutionConfig will be serialized and can't be modified afterwards. * * @param jobId The id of the job. A random ID is generated, if {@code null} is passed. * @param jobName The name of the job. */ public JobGraph(JobID jobId, String jobName) { this.jobID = jobId == null ? new JobID() : jobId; this.jobName = jobName == null ? "(unnamed job)" : jobName; try { setExecutionConfig(new ExecutionConfig()); } catch (IOException e) { // this should never happen, since an empty execution config is always serializable throw new RuntimeException("bug, empty execution config is not serializable"); } } /** * Constructs a new job graph with no name, a random job ID, the given {@link ExecutionConfig}, and * the given job vertices. The ExecutionConfig will be serialized and can't be modified afterwards. * * @param vertices The vertices to add to the graph. */ public JobGraph(JobVertex... vertices) { this(null, vertices); } /** * Constructs a new job graph with the given name, the given {@link ExecutionConfig}, a random job ID, * and the given job vertices. The ExecutionConfig will be serialized and can't be modified afterwards. * * @param jobName The name of the job. * @param vertices The vertices to add to the graph. */ public JobGraph(String jobName, JobVertex... vertices) { this(null, jobName, vertices); } /** * Constructs a new job graph with the given name, the given {@link ExecutionConfig}, * the given jobId or a random one if null supplied, and the given job vertices. * The ExecutionConfig will be serialized and can't be modified afterwards. * * @param jobId The id of the job. A random ID is generated, if {@code null} is passed. * @param jobName The name of the job. * @param vertices The vertices to add to the graph. */ public JobGraph(JobID jobId, String jobName, JobVertex... vertices) { this(jobId, jobName); for (JobVertex vertex : vertices) { addVertex(vertex); } }...... /** * Sets the savepoint restore settings. * @param settings The savepoint restore settings. */ public void setSavepointRestoreSettings(SavepointRestoreSettings settings) { this.savepointRestoreSettings = checkNotNull(settings, "Savepoint restore settings"); } /** * Returns the configured savepoint restore setting. * @return The configured savepoint restore settings. */ public SavepointRestoreSettings getSavepointRestoreSettings() { return savepointRestoreSettings; } /** * Sets the execution config. This method eagerly serialized the ExecutionConfig for future RPC * transport. Further modification of the referenced ExecutionConfig object will not affect * this serialized copy. * * @param executionConfig The ExecutionConfig to be serialized. * @throws IOException Thrown if the serialization of the ExecutionConfig fails */ public void setExecutionConfig(ExecutionConfig executionConfig) throws IOException { checkNotNull(executionConfig, "ExecutionConfig must not be null."); this.serializedExecutionConfig = new SerializedValue<>(executionConfig); } /** * Adds a new task vertex to the job graph if it is not already included. * * @param vertex * the new task vertex to be added */ public void addVertex(JobVertex vertex) { final JobVertexID id = vertex.getID(); JobVertex previous = taskVertices.put(id, vertex); // if we had a prior association, restore and throw an exception if (previous != null) { taskVertices.put(id, previous); throw new IllegalArgumentException("The JobGraph already contains a vertex with that id."); } } /** * Returns an Iterable to iterate all vertices registered with the job graph. * * @return an Iterable to iterate all vertices registered with the job graph */ public Iterable getVertices() { return this.taskVertices.values(); } /** * Returns an array of all job vertices that are registered with the job graph. The order in which the vertices * appear in the list is not defined. * * @return an array of all job vertices that are registered with the job graph */ public JobVertex[] getVerticesAsArray() { return this.taskVertices.values().toArray(new JobVertex[this.taskVertices.size()]); } /** * Returns the number of all vertices. * * @return The number of all vertices. */ public int getNumberOfVertices() { return this.taskVertices.size(); } /** * Sets the settings for asynchronous snapshots. A value of {@code null} means that * snapshotting is not enabled. * * @param settings The snapshot settings */ public void setSnapshotSettings(JobCheckpointingSettings settings) { this.snapshotSettings = settings; } /** * Gets the settings for asynchronous snapshots. This method returns null, when * checkpointing is not enabled. * * @return The snapshot settings */ public JobCheckpointingSettings getCheckpointingSettings() { return snapshotSettings; } /** * Checks if the checkpointing was enabled for this job graph * * @return true if checkpointing enabled */ public boolean isCheckpointingEnabled() { if (snapshotSettings == null) { return false; } long checkpointInterval = snapshotSettings.getCheckpointCoordinatorConfiguration().getCheckpointInterval(); return checkpointInterval > 0 && checkpointInterval < Long.MAX_VALUE; } /** * Searches for a vertex with a matching ID and returns it. * * @param id * the ID of the vertex to search for * @return the vertex with the matching ID or null if no vertex with such ID could be found */ public JobVertex findVertexByID(JobVertexID id) { return this.taskVertices.get(id); } /** * Sets the classpaths required to run the job on a task manager. * * @param paths paths of the directories/JAR files required to run the job on a task manager */ public void setClasspaths(List paths) { classpaths = paths; } public List getClasspaths() { return classpaths; } /** * Gets the maximum parallelism of all operations in this job graph. * * @return The maximum parallelism of this job graph */ public int getMaximumParallelism() { int maxParallelism = -1; for (JobVertex vertex : taskVertices.values()) { maxParallelism = Math.max(vertex.getParallelism(), maxParallelism); } return maxParallelism; } // -------------------------------------------------------------------------------------------- // Topological Graph Access // -------------------------------------------------------------------------------------------- public List getVerticesSortedTopologicallyFromSources() throws InvalidProgramException { // early out on empty lists if (this.taskVertices.isEmpty()) { return Collections.emptyList(); } List sorted = new ArrayList (this.taskVertices.size()); Set remaining = new LinkedHashSet (this.taskVertices.values()); // start by finding the vertices with no input edges // and the ones with disconnected inputs (that refer to some standalone data set) { Iterator iter = remaining.iterator(); while (iter.hasNext()) { JobVertex vertex = iter.next(); if (vertex.hasNoConnectedInputs()) { sorted.add(vertex); iter.remove(); } } } int startNodePos = 0; // traverse from the nodes that were added until we found all elements while (!remaining.isEmpty()) { // first check if we have more candidates to start traversing from. if not, then the // graph is cyclic, which is not permitted if (startNodePos >= sorted.size()) { throw new InvalidProgramException("The job graph is cyclic."); } JobVertex current = sorted.get(startNodePos++); addNodesThatHaveNoNewPredecessors(current, sorted, remaining); } return sorted; } private void addNodesThatHaveNoNewPredecessors(JobVertex start, List target, Set remaining) { // forward traverse over all produced data sets and all their consumers for (IntermediateDataSet dataSet : start.getProducedDataSets()) { for (JobEdge edge : dataSet.getConsumers()) { // a vertex can be added, if it has no predecessors that are still in the 'remaining' set JobVertex v = edge.getTarget(); if (!remaining.contains(v)) { continue; } boolean hasNewPredecessors = false; for (JobEdge e : v.getInputs()) { // skip the edge through which we came if (e == edge) { continue; } IntermediateDataSet source = e.getSource(); if (remaining.contains(source.getProducer())) { hasNewPredecessors = true; break; } } if (!hasNewPredecessors) { target.add(v); remaining.remove(v); addNodesThatHaveNoNewPredecessors(v, target, remaining); } } } } // -------------------------------------------------------------------------------------------- // Handling of attached JAR files // -------------------------------------------------------------------------------------------- /** * Adds the path of a JAR file required to run the job on a task manager. * * @param jar * path of the JAR file required to run the job on a task manager */ public void addJar(Path jar) { if (jar == null) { throw new IllegalArgumentException(); } if (!userJars.contains(jar)) { userJars.add(jar); } } /** * Gets the list of assigned user jar paths. * * @return The list of assigned user jar paths */ public List getUserJars() { return userJars; } /** * Adds the path of a custom file required to run the job on a task manager. * * @param name a name under which this artifact will be accessible through {@link DistributedCache} * @param file path of a custom file required to run the job on a task manager */ public void addUserArtifact(String name, DistributedCache.DistributedCacheEntry file) { if (file == null) { throw new IllegalArgumentException(); } userArtifacts.putIfAbsent(name, file); } /** * Gets the list of assigned user jar paths. * * @return The list of assigned user jar paths */ public Map getUserArtifacts() { return userArtifacts; } /** * Adds the BLOB referenced by the key to the JobGraph's dependencies. * * @param key * path of the JAR file required to run the job on a task manager */ public void addUserJarBlobKey(PermanentBlobKey key) { if (key == null) { throw new IllegalArgumentException(); } if (!userJarBlobKeys.contains(key)) { userJarBlobKeys.add(key); } } /** * Checks whether the JobGraph has user code JAR files attached. * * @return True, if the JobGraph has user code JAR files attached, false otherwise. */ public boolean hasUsercodeJarFiles() { return this.userJars.size() > 0; } /** * Returns a set of BLOB keys referring to the JAR files required to run this job. * * @return set of BLOB keys referring to the JAR files required to run this job */ public List getUserJarBlobKeys() { return this.userJarBlobKeys; } @Override public String toString() { return "JobGraph(jobId: " + jobID + ")"; } public void setUserArtifactBlobKey(String entryName, PermanentBlobKey blobKey) throws IOException { byte[] serializedBlobKey; serializedBlobKey = InstantiationUtil.serializeObject(blobKey); userArtifacts.computeIfPresent(entryName, (key, originalEntry) -> new DistributedCache.DistributedCacheEntry( originalEntry.filePath, originalEntry.isExecutable, serializedBlobKey, originalEntry.isZipped )); } public void writeUserArtifactEntriesToConfiguration() { for (Map.Entry userArtifact : userArtifacts.entrySet()) { DistributedCache.writeFileInfoToConfig( userArtifact.getKey(), userArtifact.getValue(), jobConfiguration ); } }} 作业图是是以DAG（有向无环图）来组织的，是通过顶点和中间态的连接形成的，来看一下这个类的定义中Map<JobVertexID, JobVertex>类型的映射变量，ScheduleMode调度模式和检查点（快照），而在函数addNodesThatHaveNoNewPredecessors中有JobEdge（做为JobVertex的数据输入通道，也就是图的连接边），那看一下JobVertex的定义：
/** * The base class for job vertexes. */public class JobVertex implements java.io.Serializable { private static final long serialVersionUID = 1L; private static final String DEFAULT_NAME = "(unnamed vertex)"; // -------------------------------------------------------------------------------------------- // Members that define the structure / topology of the graph // -------------------------------------------------------------------------------------------- /** The ID of the vertex. */ private final JobVertexID id; /** The alternative IDs of the vertex. */ private final ArrayList idAlternatives = new ArrayList<>(); /** The IDs of all operators contained in this vertex. */ private final ArrayList operatorIDs = new ArrayList<>(); /** The alternative IDs of all operators contained in this vertex. */ private final ArrayList operatorIdsAlternatives = new ArrayList<>(); /** List of produced data sets, one per writer */ private final ArrayList results = new ArrayList (); /** List of edges with incoming data. One per Reader. */ private final ArrayList inputs = new ArrayList (); /** Number of subtasks to split this task into at runtime.*/ private int parallelism = ExecutionConfig.PARALLELISM_DEFAULT; /** Maximum number of subtasks to split this task into a runtime. */ private int maxParallelism = -1; /** The minimum resource of the vertex */ private ResourceSpec minResources = ResourceSpec.DEFAULT; /** The preferred resource of the vertex */ private ResourceSpec preferredResources = ResourceSpec.DEFAULT; /** Custom configuration passed to the assigned task at runtime. */ private Configuration configuration; /** The class of the invokable. */ private String invokableClassName; /** Indicates of this job vertex is stoppable or not. */ private boolean isStoppable = false; /** Optionally, a source of input splits */ private InputSplitSource inputSplitSource; /** The name of the vertex. This will be shown in runtime logs and will be in the runtime environment */ private String name; /** Optionally, a sharing group that allows subtasks from different job vertices to run concurrently in one slot */ private SlotSharingGroup slotSharingGroup; /** The group inside which the vertex subtasks share slots */ private CoLocationGroup coLocationGroup; /** Optional, the name of the operator, such as 'Flat Map' or 'Join', to be included in the JSON plan */ private String operatorName; /** Optional, the description of the operator, like 'Hash Join', or 'Sorted Group Reduce', * to be included in the JSON plan */ private String operatorDescription; /** Optional, pretty name of the operator, to be displayed in the JSON plan */ private String operatorPrettyName; /** Optional, the JSON for the optimizer properties of the operator result, * to be included in the JSON plan */ private String resultOptimizerProperties; /** The input dependency constraint to schedule this vertex. */ private InputDependencyConstraint inputDependencyConstraint = InputDependencyConstraint.ANY; // -------------------------------------------------------------------------------------------- /** * Constructs a new job vertex and assigns it with the given name. * * @param name The name of the new job vertex. */ public JobVertex(String name) { this(name, null); } /** * Constructs a new job vertex and assigns it with the given name. * * @param name The name of the new job vertex. * @param id The id of the job vertex. */ public JobVertex(String name, JobVertexID id) { this.name = name == null ? DEFAULT_NAME : name; this.id = id == null ? new JobVertexID() : id; // the id lists must have the same size this.operatorIDs.add(OperatorID.fromJobVertexID(this.id)); this.operatorIdsAlternatives.add(null); } /** * Constructs a new job vertex and assigns it with the given name. * * @param name The name of the new job vertex. * @param primaryId The id of the job vertex. * @param alternativeIds The alternative ids of the job vertex. * @param operatorIds The ids of all operators contained in this job vertex. * @param alternativeOperatorIds The alternative ids of all operators contained in this job vertex- */ public JobVertex(String name, JobVertexID primaryId, List alternativeIds, List operatorIds, List alternativeOperatorIds) { Preconditions.checkArgument(operatorIds.size() == alternativeOperatorIds.size()); this.name = name == null ? DEFAULT_NAME : name; this.id = primaryId == null ? new JobVertexID() : primaryId; this.idAlternatives.addAll(alternativeIds); this.operatorIDs.addAll(operatorIds); this.operatorIdsAlternatives.addAll(alternativeOperatorIds); }...... /** * Returns the number of inputs. * * @return The number of inputs. */ public int getNumberOfInputs() { return this.inputs.size(); } public List getOperatorIDs() { return operatorIDs; } public List getUserDefinedOperatorIDs() { return operatorIdsAlternatives; } /** * Returns the vertex's configuration object which can be used to pass custom settings to the task at runtime. * * @return the vertex's configuration object */ public Configuration getConfiguration() { if (this.configuration == null) { this.configuration = new Configuration(); } return this.configuration; } public void setInvokableClass(Class invokable) { Preconditions.checkNotNull(invokable); this.invokableClassName = invokable.getName(); this.isStoppable = StoppableTask.class.isAssignableFrom(invokable); } /** * Returns the name of the invokable class which represents the task of this vertex. * * @return The name of the invokable class, null if not set. */ public String getInvokableClassName() { return this.invokableClassName; } /** * Returns the invokable class which represents the task of this vertex * * @param cl The classloader used to resolve user-defined classes * @return The invokable class, null if it is not set */ public Class getInvokableClass(ClassLoader cl) { if (cl == null) { throw new NullPointerException("The classloader must not be null."); } if (invokableClassName == null) { return null; } try { return Class.forName(invokableClassName, true, cl).asSubclass(AbstractInvokable.class); } catch (ClassNotFoundException e) { throw new RuntimeException("The user-code class could not be resolved.", e); } catch (ClassCastException e) { throw new RuntimeException("The user-code class is no subclass of " + AbstractInvokable.class.getName(), e); } } /** * Gets the parallelism of the task. * * @return The parallelism of the task. */ public int getParallelism() { return parallelism; } /** * Sets the parallelism for the task. * * @param parallelism The parallelism for the task. */ public void setParallelism(int parallelism) { if (parallelism < 1) { throw new IllegalArgumentException("The parallelism must be at least one."); } this.parallelism = parallelism; }...... /** * Sets the minimum and preferred resources for the task. * * @param minResources The minimum resource for the task. * @param preferredResources The preferred resource for the task. */ public void setResources(ResourceSpec minResources, ResourceSpec preferredResources) { this.minResources = checkNotNull(minResources); this.preferredResources = checkNotNull(preferredResources); } public InputSplitSource getInputSplitSource() { return inputSplitSource; } public void setInputSplitSource(InputSplitSource inputSplitSource) { this.inputSplitSource = inputSplitSource; } public List getProducedDataSets() { return this.results; } public List getInputs() { return this.inputs; } /** * Associates this vertex with a slot sharing group for scheduling. Different vertices in the same * slot sharing group can run one subtask each in the same slot. * * @param grp The slot sharing group to associate the vertex with. */ public void setSlotSharingGroup(SlotSharingGroup grp) { if (this.slotSharingGroup != null) { this.slotSharingGroup.removeVertexFromGroup(id); } this.slotSharingGroup = grp; if (grp != null) { grp.addVertexToGroup(id); } } /** * Gets the slot sharing group that this vertex is associated with. Different vertices in the same * slot sharing group can run one subtask each in the same slot. If the vertex is not associated with * a slot sharing group, this method returns {@code null}. * * @return The slot sharing group to associate the vertex with, or {@code null}, if not associated with one. */ public SlotSharingGroup getSlotSharingGroup() { return slotSharingGroup; }...... public void setStrictlyCoLocatedWith(JobVertex strictlyCoLocatedWith) { if (this.slotSharingGroup == null || this.slotSharingGroup != strictlyCoLocatedWith.slotSharingGroup) { throw new IllegalArgumentException("Strict co-location requires that both vertices are in the same slot sharing group."); } CoLocationGroup thisGroup = this.coLocationGroup; CoLocationGroup otherGroup = strictlyCoLocatedWith.coLocationGroup; if (otherGroup == null) { if (thisGroup == null) { CoLocationGroup group = new CoLocationGroup(this, strictlyCoLocatedWith); this.coLocationGroup = group; strictlyCoLocatedWith.coLocationGroup = group; } else { thisGroup.addVertex(strictlyCoLocatedWith); strictlyCoLocatedWith.coLocationGroup = thisGroup; } } else { if (thisGroup == null) { otherGroup.addVertex(this); this.coLocationGroup = otherGroup; } else { // both had yet distinct groups, we need to merge them thisGroup.mergeInto(otherGroup); } } } public CoLocationGroup getCoLocationGroup() { return coLocationGroup; } public void updateCoLocationGroup(CoLocationGroup group) { this.coLocationGroup = group; } // -------------------------------------------------------------------------------------------- public IntermediateDataSet createAndAddResultDataSet(ResultPartitionType partitionType) { return createAndAddResultDataSet(new IntermediateDataSetID(), partitionType); } public IntermediateDataSet createAndAddResultDataSet( IntermediateDataSetID id, ResultPartitionType partitionType) { IntermediateDataSet result = new IntermediateDataSet(id, partitionType, this); this.results.add(result); return result; } public JobEdge connectDataSetAsInput(IntermediateDataSet dataSet, DistributionPattern distPattern) { JobEdge edge = new JobEdge(dataSet, this, distPattern); this.inputs.add(edge); dataSet.addConsumer(edge); return edge; } public JobEdge connectNewDataSetAsInput( JobVertex input, DistributionPattern distPattern, ResultPartitionType partitionType) { IntermediateDataSet dataSet = input.createAndAddResultDataSet(partitionType); JobEdge edge = new JobEdge(dataSet, this, distPattern); this.inputs.add(edge); dataSet.addConsumer(edge); return edge; } public void connectIdInput(IntermediateDataSetID dataSetId, DistributionPattern distPattern) { JobEdge edge = new JobEdge(dataSetId, this, distPattern); this.inputs.add(edge); } // -------------------------------------------------------------------------------------------- //能过下面两个函数，可以看出此数据结构的输入和输出分别为JobEdge和IntermediateDataSet public boolean isInputVertex() { return this.inputs.isEmpty(); } public boolean isOutputVertex() { return this.results.isEmpty(); } public boolean hasNoConnectedInputs() { for (JobEdge edge : inputs) { if (!edge.isIdReference()) { return false; } } return true; }......} 通过分析代码可以看出，这个图的节点的输入输出分别为：JobEdge和IntermediateDataSet两个数据结构。内部还有OperatorID的数据结构，用来管理对数据流的操作形态。涉及到Slot的SlotSharingGroup和CoLocationGroup。再看一下JobEdge和IntermediateDataSet两个数据结构的定义：
public class JobEdge implements java.io.Serializable { private static final long serialVersionUID = 1L; /** The vertex connected to this edge. */ private final JobVertex target; /** The distribution pattern that should be used for this job edge. */ private final DistributionPattern distributionPattern; /** The data set at the source of the edge, may be null if the edge is not yet connected*/ private IntermediateDataSet source; /** The id of the source intermediate data set */ private IntermediateDataSetID sourceId; /** Optional name for the data shipping strategy (forward, partition hash, rebalance, ...), * to be displayed in the JSON plan */ private String shipStrategyName; /** Optional name for the pre-processing operation (sort, combining sort, ...), * to be displayed in the JSON plan */ private String preProcessingOperationName; /** Optional description of the caching inside an operator, to be displayed in the JSON plan */ private String operatorLevelCachingDescription; /** * Constructs a new job edge, that connects an intermediate result to a consumer task. * * @param source The data set that is at the source of this edge. * @param target The operation that is at the target of this edge. * @param distributionPattern The pattern that defines how the connection behaves in parallel. */ public JobEdge(IntermediateDataSet source, JobVertex target, DistributionPattern distributionPattern) { if (source == null || target == null || distributionPattern == null) { throw new NullPointerException(); } this.target = target; this.distributionPattern = distributionPattern; this.source = source; this.sourceId = source.getId(); } /** * Constructs a new job edge that refers to an intermediate result via the Id, rather than directly through * the intermediate data set structure. * * @param sourceId The id of the data set that is at the source of this edge. * @param target The operation that is at the target of this edge. * @param distributionPattern The pattern that defines how the connection behaves in parallel. */ public JobEdge(IntermediateDataSetID sourceId, JobVertex target, DistributionPattern distributionPattern) { if (sourceId == null || target == null || distributionPattern == null) { throw new NullPointerException(); } this.target = target; this.distributionPattern = distributionPattern; this.sourceId = sourceId; } /** * Returns the data set at the source of the edge. May be null, if the edge refers to the source via an ID * and has not been connected. * * @return The data set at the source of the edge */ public IntermediateDataSet getSource() { return source; } /** * Returns the vertex connected to this edge. * * @return The vertex connected to this edge. */ public JobVertex getTarget() { return target; } /** * Returns the distribution pattern used for this edge. * * @return The distribution pattern used for this edge. */ public DistributionPattern getDistributionPattern(){ return this.distributionPattern; } /** * Gets the ID of the consumed data set. * * @return The ID of the consumed data set. */ public IntermediateDataSetID getSourceId() { return sourceId; } public boolean isIdReference() { return this.source == null; } // -------------------------------------------------------------------------------------------- public void connecDataSet(IntermediateDataSet dataSet) { if (dataSet == null) { throw new NullPointerException(); } if (this.source != null) { throw new IllegalStateException("The edge is already connected."); } if (!dataSet.getId().equals(sourceId)) { throw new IllegalArgumentException("The data set to connect does not match the sourceId."); } this.source = dataSet; }......}public class IntermediateDataSet implements java.io.Serializable { private static final long serialVersionUID = 1L; private final IntermediateDataSetID id; // the identifier private final JobVertex producer; // the operation that produced this data set private final List consumers = new ArrayList (); // The type of partition to use at runtime private final ResultPartitionType resultType; // -------------------------------------------------------------------------------------------- public IntermediateDataSet(IntermediateDataSetID id, ResultPartitionType resultType, JobVertex producer) { this.id = checkNotNull(id); this.producer = checkNotNull(producer); this.resultType = checkNotNull(resultType); } // -------------------------------------------------------------------------------------------- public IntermediateDataSetID getId() { return id; } public JobVertex getProducer() { return producer; } public List getConsumers() { return this.consumers; } public ResultPartitionType getResultType() { return resultType; } // -------------------------------------------------------------------------------------------- public void addConsumer(JobEdge edge) { this.consumers.add(edge); } // -------------------------------------------------------------------------------------------- @Override public String toString() { return "Intermediate Data Set (" + id + ")"; }} 是不是觉得这几个类都是互相勾连的，而且它们都继承了Serializable这个串行化的类，说明它们都是对数据进行操作的。不然继承这个玩意儿有啥意义。
在Standalone的模式下，看一下，分发器的创建：
创建的顺序：StandaloneSessionClusterEntrypoint->SessionDispatcherResourceManagerComponentFactory->createDispatcherResourceManagerComponent->createDispatcher(StandaloneDispatcher)public class StandaloneDispatcher extends Dispatcher { public StandaloneDispatcher( RpcService rpcService, String endpointId, Configuration configuration, HighAvailabilityServices highAvailabilityServices, GatewayRetriever resourceManagerGatewayRetriever, BlobServer blobServer, HeartbeatServices heartbeatServices, JobManagerMetricGroup jobManagerMetricGroup, @Nullable String metricQueryServicePath, ArchivedExecutionGraphStore archivedExecutionGraphStore, JobManagerRunnerFactory jobManagerRunnerFactory, FatalErrorHandler fatalErrorHandler, HistoryServerArchivist historyServerArchivist) throws Exception { super( rpcService, endpointId, configuration, highAvailabilityServices, highAvailabilityServices.getSubmittedJobGraphStore(),//注意这个函数 resourceManagerGatewayRetriever, blobServer, heartbeatServices, jobManagerMetricGroup, metricQueryServicePath, archivedExecutionGraphStore, jobManagerRunnerFactory, fatalErrorHandler, historyServerArchivist); }} getSubmittedJobGraphStore可以用来拿到实现JobGraph的实例的句柄，看一下这个接口的定义：
public interface SubmittedJobGraphStore { /** * Starts the {@link SubmittedJobGraphStore} service. */ void start(SubmittedJobGraphListener jobGraphListener) throws Exception; /** * Stops the {@link SubmittedJobGraphStore} service. */ void stop() throws Exception; /** * Returns the {@link SubmittedJobGraph} with the given {@link JobID} or * {@code null} if no job was registered. */ @Nullable SubmittedJobGraph recoverJobGraph(JobID jobId) throws Exception; /** * Adds the {@link SubmittedJobGraph} instance. * * If a job graph with the same {@link JobID} exists, it is replaced. */ void putJobGraph(SubmittedJobGraph jobGraph) throws Exception; /** * Removes the {@link SubmittedJobGraph} with the given {@link JobID} if it exists. */ void removeJobGraph(JobID jobId) throws Exception; /** * Releases the locks on the specified {@link JobGraph}. * * Releasing the locks allows that another instance can delete the job from * the {@link SubmittedJobGraphStore}. * * @param jobId specifying the job to release the locks for * @throws Exception if the locks cannot be released */ void releaseJobGraph(JobID jobId) throws Exception; /** * Get all job ids of submitted job graphs to the submitted job graph store. * * @return Collection of submitted job ids * @throws Exception if the operation fails */ Collection
getJobIds() throws Exception; /** * A listener for {@link SubmittedJobGraph} instances. This is used to react to races between * multiple running {@link SubmittedJobGraphStore} instances (on multiple job managers). */ interface SubmittedJobGraphListener { /** * Callback for {@link SubmittedJobGraph} instances added by a different {@link * SubmittedJobGraphStore} instance. * * Important: It is possible to get false positives and be notified * about a job graph, which was added by this instance. * * @param jobId The {@link JobID} of the added job graph */ void onAddedJobGraph(JobID jobId); /** * Callback for {@link SubmittedJobGraph} instances removed by a different {@link * SubmittedJobGraphStore} instance. * * @param jobId The {@link JobID} of the removed job graph */ void onRemovedJobGraph(JobID jobId); }}
也就是在前面分析HA服务创建的时候儿，会创建这些个服务，这些服务中会有一个启动对JobGraph的管理的任务，通过它拿到作业图的控制权。
那么工作图是如何分发的呢？这里可以看看所有的Dispatcher的抽象类的定义：