@Test public void testIteration() { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream<Integer> source = env.fromElements(1, 2, 3).name("source"); IterativeStream<Integer> iteration = source.iterate(3000); iteration.name("iteration").setParallelism(2); DataStream<Integer> map = iteration.map(x -> x + 1).name("map").setParallelism(2); DataStream<Integer> filter = map.filter((x) -> false).name("filter").setParallelism(2); iteration.closeWith(filter).print(); JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(env.getStreamGraph()); SlotSharingGroup slotSharingGroup = jobGraph.getVerticesAsArray()[0].getSlotSharingGroup(); assertNotNull(slotSharingGroup); CoLocationGroup iterationSourceCoLocationGroup = null; CoLocationGroup iterationSinkCoLocationGroup = null; for (JobVertex jobVertex : jobGraph.getVertices()) { assertEquals(slotSharingGroup, jobVertex.getSlotSharingGroup()); if (jobVertex.getName().startsWith(StreamGraph.ITERATION_SOURCE_NAME_PREFIX)) { iterationSourceCoLocationGroup = jobVertex.getCoLocationGroup(); assertTrue(iterationSourceCoLocationGroup.getVertices().contains(jobVertex)); } else if (jobVertex.getName().startsWith(StreamGraph.ITERATION_SINK_NAME_PREFIX)) { iterationSinkCoLocationGroup = jobVertex.getCoLocationGroup(); assertTrue(iterationSinkCoLocationGroup.getVertices().contains(jobVertex)); } else { assertNull(jobVertex.getCoLocationGroup()); } } assertNotNull(iterationSourceCoLocationGroup); assertNotNull(iterationSinkCoLocationGroup); assertEquals(iterationSourceCoLocationGroup, iterationSinkCoLocationGroup); }
public static JobGraph createJobGraph(StreamGraph streamGraph) { return createJobGraph(streamGraph, null); }
StreamingJobGraphGenerator { public static JobGraph createJobGraph(StreamGraph streamGraph) { return createJobGraph(streamGraph, null); } }
StreamingJobGraphGenerator { public static JobGraph createJobGraph(StreamGraph streamGraph) { return createJobGraph(streamGraph, null); } private  StreamingJobGraphGenerator(StreamGraph streamGraph); private  StreamingJobGraphGenerator(StreamGraph streamGraph, @Nullable JobID jobID); }
StreamingJobGraphGenerator { public static JobGraph createJobGraph(StreamGraph streamGraph) { return createJobGraph(streamGraph, null); } private  StreamingJobGraphGenerator(StreamGraph streamGraph); private  StreamingJobGraphGenerator(StreamGraph streamGraph, @Nullable JobID jobID); static JobGraph createJobGraph(StreamGraph streamGraph); static JobGraph createJobGraph(StreamGraph streamGraph, @Nullable JobID jobID); static boolean isChainable(StreamEdge edge, StreamGraph streamGraph); }
StreamingJobGraphGenerator { public static JobGraph createJobGraph(StreamGraph streamGraph) { return createJobGraph(streamGraph, null); } private  StreamingJobGraphGenerator(StreamGraph streamGraph); private  StreamingJobGraphGenerator(StreamGraph streamGraph, @Nullable JobID jobID); static JobGraph createJobGraph(StreamGraph streamGraph); static JobGraph createJobGraph(StreamGraph streamGraph, @Nullable JobID jobID); static boolean isChainable(StreamEdge edge, StreamGraph streamGraph);  }