@Test public void testResourcesForChainedSourceSink() throws Exception { ResourceSpec resource1 = ResourceSpec.newBuilder().setCpuCores(0.1).setHeapMemoryInMB(100).build(); ResourceSpec resource2 = ResourceSpec.newBuilder().setCpuCores(0.2).setHeapMemoryInMB(200).build(); ResourceSpec resource3 = ResourceSpec.newBuilder().setCpuCores(0.3).setHeapMemoryInMB(300).build(); ResourceSpec resource4 = ResourceSpec.newBuilder().setCpuCores(0.4).setHeapMemoryInMB(400).build(); ResourceSpec resource5 = ResourceSpec.newBuilder().setCpuCores(0.5).setHeapMemoryInMB(500).build(); 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))); }
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);  }