@Test public void testTopologicalSort3() { try { JobVertex source = new JobVertex("source"); JobVertex op1 = new JobVertex("op4"); JobVertex op2 = new JobVertex("op2"); JobVertex op3 = new JobVertex("op3"); op1.connectNewDataSetAsInput(source, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED); op2.connectNewDataSetAsInput(op1, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED); op2.connectNewDataSetAsInput(source, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED); op3.connectNewDataSetAsInput(op2, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED); JobGraph graph = new JobGraph("TestGraph", source, op1, op2, op3); List<JobVertex> sorted = graph.getVerticesSortedTopologicallyFromSources(); assertEquals(4, sorted.size()); assertBefore(source, op1, sorted); assertBefore(source, op2, sorted); assertBefore(op1, op2, sorted); assertBefore(op2, op3, sorted); } catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } }
public List<JobVertex> getVerticesSortedTopologicallyFromSources() throws InvalidProgramException { if (this.taskVertices.isEmpty()) { return Collections.emptyList(); } List<JobVertex> sorted = new ArrayList<JobVertex>(this.taskVertices.size()); Set<JobVertex> remaining = new LinkedHashSet<JobVertex>(this.taskVertices.values()); { Iterator<JobVertex> iter = remaining.iterator(); while (iter.hasNext()) { JobVertex vertex = iter.next(); if (vertex.hasNoConnectedInputs()) { sorted.add(vertex); iter.remove(); } } } int startNodePos = 0; while (!remaining.isEmpty()) { if (startNodePos >= sorted.size()) { throw new InvalidProgramException("The job graph is cyclic."); } JobVertex current = sorted.get(startNodePos++); addNodesThatHaveNoNewPredecessors(current, sorted, remaining); } return sorted; }
JobGraph implements Serializable { public List<JobVertex> getVerticesSortedTopologicallyFromSources() throws InvalidProgramException { if (this.taskVertices.isEmpty()) { return Collections.emptyList(); } List<JobVertex> sorted = new ArrayList<JobVertex>(this.taskVertices.size()); Set<JobVertex> remaining = new LinkedHashSet<JobVertex>(this.taskVertices.values()); { Iterator<JobVertex> iter = remaining.iterator(); while (iter.hasNext()) { JobVertex vertex = iter.next(); if (vertex.hasNoConnectedInputs()) { sorted.add(vertex); iter.remove(); } } } int startNodePos = 0; while (!remaining.isEmpty()) { if (startNodePos >= sorted.size()) { throw new InvalidProgramException("The job graph is cyclic."); } JobVertex current = sorted.get(startNodePos++); addNodesThatHaveNoNewPredecessors(current, sorted, remaining); } return sorted; } }
JobGraph implements Serializable { public List<JobVertex> getVerticesSortedTopologicallyFromSources() throws InvalidProgramException { if (this.taskVertices.isEmpty()) { return Collections.emptyList(); } List<JobVertex> sorted = new ArrayList<JobVertex>(this.taskVertices.size()); Set<JobVertex> remaining = new LinkedHashSet<JobVertex>(this.taskVertices.values()); { Iterator<JobVertex> iter = remaining.iterator(); while (iter.hasNext()) { JobVertex vertex = iter.next(); if (vertex.hasNoConnectedInputs()) { sorted.add(vertex); iter.remove(); } } } int startNodePos = 0; while (!remaining.isEmpty()) { if (startNodePos >= sorted.size()) { throw new InvalidProgramException("The job graph is cyclic."); } JobVertex current = sorted.get(startNodePos++); addNodesThatHaveNoNewPredecessors(current, sorted, remaining); } return sorted; } JobGraph(String jobName);  JobGraph(JobID jobId, String jobName);  JobGraph(JobVertex... vertices);  JobGraph(String jobName, JobVertex... vertices);  JobGraph(JobID jobId, String jobName, JobVertex... vertices); }
JobGraph implements Serializable { public List<JobVertex> getVerticesSortedTopologicallyFromSources() throws InvalidProgramException { if (this.taskVertices.isEmpty()) { return Collections.emptyList(); } List<JobVertex> sorted = new ArrayList<JobVertex>(this.taskVertices.size()); Set<JobVertex> remaining = new LinkedHashSet<JobVertex>(this.taskVertices.values()); { Iterator<JobVertex> iter = remaining.iterator(); while (iter.hasNext()) { JobVertex vertex = iter.next(); if (vertex.hasNoConnectedInputs()) { sorted.add(vertex); iter.remove(); } } } int startNodePos = 0; while (!remaining.isEmpty()) { if (startNodePos >= sorted.size()) { throw new InvalidProgramException("The job graph is cyclic."); } JobVertex current = sorted.get(startNodePos++); addNodesThatHaveNoNewPredecessors(current, sorted, remaining); } return sorted; } JobGraph(String jobName);  JobGraph(JobID jobId, String jobName);  JobGraph(JobVertex... vertices);  JobGraph(String jobName, JobVertex... vertices);  JobGraph(JobID jobId, String jobName, JobVertex... vertices); JobID getJobID(); String getName(); Configuration getJobConfiguration(); SerializedValue<ExecutionConfig> getSerializedExecutionConfig(); long getSessionTimeout(); void setSessionTimeout(long sessionTimeout); void setAllowQueuedScheduling(boolean allowQueuedScheduling); boolean getAllowQueuedScheduling(); void setScheduleMode(ScheduleMode scheduleMode); ScheduleMode getScheduleMode(); void setSavepointRestoreSettings(SavepointRestoreSettings settings); SavepointRestoreSettings getSavepointRestoreSettings(); void setExecutionConfig(ExecutionConfig executionConfig); void addVertex(JobVertex vertex); Iterable<JobVertex> getVertices(); JobVertex[] getVerticesAsArray(); int getNumberOfVertices(); void setSnapshotSettings(JobCheckpointingSettings settings); JobCheckpointingSettings getCheckpointingSettings(); boolean isCheckpointingEnabled(); JobVertex findVertexByID(JobVertexID id); void setClasspaths(List<URL> paths); List<URL> getClasspaths(); int getMaximumParallelism(); List<JobVertex> getVerticesSortedTopologicallyFromSources(); void addJar(Path jar); List<Path> getUserJars(); void addUserArtifact(String name, DistributedCache.DistributedCacheEntry file); Map<String, DistributedCache.DistributedCacheEntry> getUserArtifacts(); void addUserJarBlobKey(PermanentBlobKey key); boolean hasUsercodeJarFiles(); List<PermanentBlobKey> getUserJarBlobKeys(); @Override String toString(); void setUserArtifactBlobKey(String entryName, PermanentBlobKey blobKey); void writeUserArtifactEntriesToConfiguration(); }
JobGraph implements Serializable { public List<JobVertex> getVerticesSortedTopologicallyFromSources() throws InvalidProgramException { if (this.taskVertices.isEmpty()) { return Collections.emptyList(); } List<JobVertex> sorted = new ArrayList<JobVertex>(this.taskVertices.size()); Set<JobVertex> remaining = new LinkedHashSet<JobVertex>(this.taskVertices.values()); { Iterator<JobVertex> iter = remaining.iterator(); while (iter.hasNext()) { JobVertex vertex = iter.next(); if (vertex.hasNoConnectedInputs()) { sorted.add(vertex); iter.remove(); } } } int startNodePos = 0; while (!remaining.isEmpty()) { if (startNodePos >= sorted.size()) { throw new InvalidProgramException("The job graph is cyclic."); } JobVertex current = sorted.get(startNodePos++); addNodesThatHaveNoNewPredecessors(current, sorted, remaining); } return sorted; } JobGraph(String jobName);  JobGraph(JobID jobId, String jobName);  JobGraph(JobVertex... vertices);  JobGraph(String jobName, JobVertex... vertices);  JobGraph(JobID jobId, String jobName, JobVertex... vertices); JobID getJobID(); String getName(); Configuration getJobConfiguration(); SerializedValue<ExecutionConfig> getSerializedExecutionConfig(); long getSessionTimeout(); void setSessionTimeout(long sessionTimeout); void setAllowQueuedScheduling(boolean allowQueuedScheduling); boolean getAllowQueuedScheduling(); void setScheduleMode(ScheduleMode scheduleMode); ScheduleMode getScheduleMode(); void setSavepointRestoreSettings(SavepointRestoreSettings settings); SavepointRestoreSettings getSavepointRestoreSettings(); void setExecutionConfig(ExecutionConfig executionConfig); void addVertex(JobVertex vertex); Iterable<JobVertex> getVertices(); JobVertex[] getVerticesAsArray(); int getNumberOfVertices(); void setSnapshotSettings(JobCheckpointingSettings settings); JobCheckpointingSettings getCheckpointingSettings(); boolean isCheckpointingEnabled(); JobVertex findVertexByID(JobVertexID id); void setClasspaths(List<URL> paths); List<URL> getClasspaths(); int getMaximumParallelism(); List<JobVertex> getVerticesSortedTopologicallyFromSources(); void addJar(Path jar); List<Path> getUserJars(); void addUserArtifact(String name, DistributedCache.DistributedCacheEntry file); Map<String, DistributedCache.DistributedCacheEntry> getUserArtifacts(); void addUserJarBlobKey(PermanentBlobKey key); boolean hasUsercodeJarFiles(); List<PermanentBlobKey> getUserJarBlobKeys(); @Override String toString(); void setUserArtifactBlobKey(String entryName, PermanentBlobKey blobKey); void writeUserArtifactEntriesToConfiguration();  }