@Test public void testResourcesForChainedOperators() 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(); ResourceSpec resource6 = ResourceSpec.newBuilder().setCpuCores(0.6).setHeapMemoryInMB(600).build(); ResourceSpec resource7 = ResourceSpec.newBuilder().setCpuCores(0.7).setHeapMemoryInMB(700).build(); Method opMethod = Operator.class.getDeclaredMethod("setResources", ResourceSpec.class); opMethod.setAccessible(true); Method sinkMethod = DataSink.class.getDeclaredMethod("setResources", ResourceSpec.class); sinkMethod.setAccessible(true); MapFunction<Long, Long> mapFunction = new MapFunction<Long, Long>() { @Override public Long map(Long value) throws Exception { return value; } }; FilterFunction<Long> filterFunction = new FilterFunction<Long>() { @Override public boolean filter(Long value) throws Exception { return false; } }; ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); DataSet<Long> input = env.fromElements(1L, 2L, 3L); opMethod.invoke(input, resource1); DataSet<Long> map1 = input.map(mapFunction); opMethod.invoke(map1, resource2); DataSet<Long> filter1 = map1.filter(filterFunction); opMethod.invoke(filter1, resource3); IterativeDataSet<Long> startOfIteration = filter1.iterate(10); opMethod.invoke(startOfIteration, resource4); DataSet<Long> map2 = startOfIteration.map(mapFunction); opMethod.invoke(map2, resource5); DataSet<Long> feedback = map2.filter(filterFunction); opMethod.invoke(feedback, resource6); DataSink<Long> sink = startOfIteration.closeWith(feedback).output(new DiscardingOutputFormat<Long>()); sinkMethod.invoke(sink, resource7); Plan plan = env.createProgramPlan(); Optimizer pc = new Optimizer(new Configuration()); OptimizedPlan op = pc.compile(plan); JobGraphGenerator jgg = new JobGraphGenerator(); JobGraph jobGraph = jgg.compileJobGraph(op); JobVertex sourceMapFilterVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(0); JobVertex iterationHeadVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(1); JobVertex feedbackVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(2); JobVertex sinkVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(3); JobVertex iterationSyncVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(4); assertTrue(sourceMapFilterVertex.getMinResources().equals(resource1.merge(resource2).merge(resource3))); assertTrue(iterationHeadVertex.getPreferredResources().equals(resource4)); assertTrue(feedbackVertex.getMinResources().equals(resource5.merge(resource6))); assertTrue(sinkVertex.getPreferredResources().equals(resource7)); assertTrue(iterationSyncVertex.getMinResources().equals(resource4)); }
public JobGraph compileJobGraph(OptimizedPlan program) { return compileJobGraph(program, null); }
JobGraphGenerator implements Visitor<PlanNode> { public JobGraph compileJobGraph(OptimizedPlan program) { return compileJobGraph(program, null); } }
JobGraphGenerator implements Visitor<PlanNode> { public JobGraph compileJobGraph(OptimizedPlan program) { return compileJobGraph(program, null); } JobGraphGenerator();  JobGraphGenerator(Configuration config); }
JobGraphGenerator implements Visitor<PlanNode> { public JobGraph compileJobGraph(OptimizedPlan program) { return compileJobGraph(program, null); } JobGraphGenerator();  JobGraphGenerator(Configuration config); JobGraph compileJobGraph(OptimizedPlan program); JobGraph compileJobGraph(OptimizedPlan program, JobID jobId); static void addUserArtifactEntries(Collection<Tuple2<String, DistributedCache.DistributedCacheEntry>> userArtifacts, JobGraph jobGraph); @Override boolean preVisit(PlanNode node); @Override void postVisit(PlanNode node); }
JobGraphGenerator implements Visitor<PlanNode> { public JobGraph compileJobGraph(OptimizedPlan program) { return compileJobGraph(program, null); } JobGraphGenerator();  JobGraphGenerator(Configuration config); JobGraph compileJobGraph(OptimizedPlan program); JobGraph compileJobGraph(OptimizedPlan program, JobID jobId); static void addUserArtifactEntries(Collection<Tuple2<String, DistributedCache.DistributedCacheEntry>> userArtifacts, JobGraph jobGraph); @Override boolean preVisit(PlanNode node); @Override void postVisit(PlanNode node); static final String MERGE_ITERATION_AUX_TASKS_KEY; }