StreamingJobGraphGenerator { public static JobGraph createJobGraph(StreamGraph streamGraph) { return new StreamingJobGraphGenerator(streamGraph).createJobGraph(); } private  StreamingJobGraphGenerator(StreamGraph streamGraph); static JobGraph createJobGraph(StreamGraph streamGraph); static boolean isChainable(StreamEdge edge, StreamGraph streamGraph);  }
@Test public void testDisabledCheckpointing() throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); StreamGraph streamGraph = new StreamGraph(env); assertFalse("Checkpointing enabled", streamGraph.getCheckpointConfig().isCheckpointingEnabled()); JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph); JobCheckpointingSettings snapshottingSettings = jobGraph.getCheckpointingSettings(); assertEquals(Long.MAX_VALUE, snapshottingSettings.getCheckpointCoordinatorConfiguration().getCheckpointInterval()); }
@Test public void testChainStartEndSetting() throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.fromElements(1, 2, 3) .map(new MapFunction<Integer, Integer>() { @Override public Integer map(Integer value) throws Exception { return value; } }) .print(); JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(env.getStreamGraph()); List<JobVertex> verticesSorted = jobGraph.getVerticesSortedTopologicallyFromSources(); JobVertex sourceVertex = verticesSorted.get(0); JobVertex mapPrintVertex = verticesSorted.get(1); assertEquals(ResultPartitionType.PIPELINED_BOUNDED, sourceVertex.getProducedDataSets().get(0).getResultType()); assertEquals(ResultPartitionType.PIPELINED_BOUNDED, mapPrintVertex.getInputs().get(0).getSource().getResultType()); StreamConfig sourceConfig = new StreamConfig(sourceVertex.getConfiguration()); StreamConfig mapConfig = new StreamConfig(mapPrintVertex.getConfiguration()); Map<Integer, StreamConfig> chainedConfigs = mapConfig.getTransitiveChainedTaskConfigs(getClass().getClassLoader()); StreamConfig printConfig = chainedConfigs.values().iterator().next(); assertTrue(sourceConfig.isChainStart()); assertTrue(sourceConfig.isChainEnd()); assertTrue(mapConfig.isChainStart()); assertFalse(mapConfig.isChainEnd()); assertFalse(printConfig.isChainStart()); assertTrue(printConfig.isChainEnd()); }
@Test public void testResourcesForChainedSourceSink() throws Exception { ResourceSpec resource1 = new ResourceSpec(0.1, 100); ResourceSpec resource2 = new ResourceSpec(0.2, 200); ResourceSpec resource3 = new ResourceSpec(0.3, 300); ResourceSpec resource4 = new ResourceSpec(0.4, 400); ResourceSpec resource5 = new ResourceSpec(0.5, 500); Method opMethod = SingleOutputStreamOperator.class.getDeclaredMethod("setResources", ResourceSpec.class); opMethod.setAccessible(true); Method sinkMethod = DataStreamSink.class.getDeclaredMethod("setResources", ResourceSpec.class); sinkMethod.setAccessible(true); StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream<Tuple2<Integer, Integer>> source = env.addSource(new ParallelSourceFunction<Tuple2<Integer, Integer>>() { @Override public void run(SourceContext<Tuple2<Integer, Integer>> ctx) throws Exception { } @Override public void cancel() { } }); opMethod.invoke(source, resource1); DataStream<Tuple2<Integer, Integer>> map = source.map(new MapFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>>() { @Override public Tuple2<Integer, Integer> map(Tuple2<Integer, Integer> value) throws Exception { return value; } }); opMethod.invoke(map, resource2); DataStream<Tuple2<Integer, Integer>> filter = map.filter(new FilterFunction<Tuple2<Integer, Integer>>() { @Override public boolean filter(Tuple2<Integer, Integer> value) throws Exception { return false; } }); opMethod.invoke(filter, resource3); DataStream<Tuple2<Integer, Integer>> reduce = filter.keyBy(0).reduce(new ReduceFunction<Tuple2<Integer, Integer>>() { @Override public Tuple2<Integer, Integer> reduce(Tuple2<Integer, Integer> value1, Tuple2<Integer, Integer> value2) throws Exception { return new Tuple2<>(value1.f0, value1.f1 + value2.f1); } }); opMethod.invoke(reduce, resource4); DataStreamSink<Tuple2<Integer, Integer>> sink = reduce.addSink(new SinkFunction<Tuple2<Integer, Integer>>() { @Override public void invoke(Tuple2<Integer, Integer> value) throws Exception { } }); sinkMethod.invoke(sink, resource5); JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(env.getStreamGraph()); JobVertex sourceMapFilterVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(0); JobVertex reduceSinkVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(1); assertTrue(sourceMapFilterVertex.getMinResources().equals(resource1.merge(resource2).merge(resource3))); assertTrue(reduceSinkVertex.getPreferredResources().equals(resource4.merge(resource5))); }
@Test public void testResourcesForIteration() throws Exception { ResourceSpec resource1 = new ResourceSpec(0.1, 100); ResourceSpec resource2 = new ResourceSpec(0.2, 200); ResourceSpec resource3 = new ResourceSpec(0.3, 300); ResourceSpec resource4 = new ResourceSpec(0.4, 400); ResourceSpec resource5 = new ResourceSpec(0.5, 500); Method opMethod = SingleOutputStreamOperator.class.getDeclaredMethod("setResources", ResourceSpec.class); opMethod.setAccessible(true); Method sinkMethod = DataStreamSink.class.getDeclaredMethod("setResources", ResourceSpec.class); sinkMethod.setAccessible(true); StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream<Integer> source = env.addSource(new ParallelSourceFunction<Integer>() { @Override public void run(SourceContext<Integer> ctx) throws Exception { } @Override public void cancel() { } }).name("test_source"); opMethod.invoke(source, resource1); IterativeStream<Integer> iteration = source.iterate(3000); opMethod.invoke(iteration, resource2); DataStream<Integer> flatMap = iteration.flatMap(new FlatMapFunction<Integer, Integer>() { @Override public void flatMap(Integer value, Collector<Integer> out) throws Exception { out.collect(value); } }).name("test_flatMap"); opMethod.invoke(flatMap, resource3); DataStream<Integer> increment = flatMap.filter(new FilterFunction<Integer>() { @Override public boolean filter(Integer value) throws Exception { return false; } }).name("test_filter"); opMethod.invoke(increment, resource4); DataStreamSink<Integer> sink = iteration.closeWith(increment).addSink(new SinkFunction<Integer>() { @Override public void invoke(Integer value) throws Exception { } }).disableChaining().name("test_sink"); sinkMethod.invoke(sink, resource5); JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(env.getStreamGraph()); for (JobVertex jobVertex : jobGraph.getVertices()) { if (jobVertex.getName().contains("test_source")) { assertTrue(jobVertex.getMinResources().equals(resource1)); } else if (jobVertex.getName().contains("Iteration_Source")) { assertTrue(jobVertex.getPreferredResources().equals(resource2)); } else if (jobVertex.getName().contains("test_flatMap")) { assertTrue(jobVertex.getMinResources().equals(resource3.merge(resource4))); } else if (jobVertex.getName().contains("Iteration_Tail")) { assertTrue(jobVertex.getPreferredResources().equals(ResourceSpec.DEFAULT)); } else if (jobVertex.getName().contains("test_sink")) { assertTrue(jobVertex.getMinResources().equals(resource5)); } } }