@Test public void testResourcesForDeltaIteration() 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(); Method opMethod = Operator.class.getDeclaredMethod("setResources", ResourceSpec.class); opMethod.setAccessible(true); Method deltaMethod = DeltaIteration.class.getDeclaredMethod("setResources", ResourceSpec.class); deltaMethod.setAccessible(true); Method sinkMethod = DataSink.class.getDeclaredMethod("setResources", ResourceSpec.class); sinkMethod.setAccessible(true); MapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> mapFunction = new MapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>>() { @Override public Tuple2<Long, Long> map(Tuple2<Long, Long> value) throws Exception { return value; } }; FilterFunction<Tuple2<Long, Long>> filterFunction = new FilterFunction<Tuple2<Long, Long>>() { @Override public boolean filter(Tuple2<Long, Long> value) throws Exception { return false; } }; ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); DataSet<Tuple2<Long, Long>> input = env.fromElements(new Tuple2<>(1L, 2L)); opMethod.invoke(input, resource1); DataSet<Tuple2<Long, Long>> map = input.map(mapFunction); opMethod.invoke(map, resource2); DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> iteration = map.iterateDelta(map, 100, 0). registerAggregator("test", new LongSumAggregator()); deltaMethod.invoke(iteration, resource3); DataSet<Tuple2<Long, Long>> delta = iteration.getWorkset().map(mapFunction); opMethod.invoke(delta, resource4); DataSet<Tuple2<Long, Long>> feedback = delta.filter(filterFunction); opMethod.invoke(feedback, resource5); DataSink<Tuple2<Long, Long>> sink = iteration.closeWith(delta, feedback). output(new DiscardingOutputFormat<Tuple2<Long, Long>>()); sinkMethod.invoke(sink, resource6); 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 sourceMapVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(0); JobVertex iterationHeadVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(1); JobVertex deltaVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(2); JobVertex iterationTailVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(3); JobVertex feedbackVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(4); JobVertex sinkVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(5); JobVertex iterationSyncVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(6); assertTrue(sourceMapVertex.getMinResources().equals(resource1.merge(resource2))); assertTrue(iterationHeadVertex.getPreferredResources().equals(resource3)); assertTrue(deltaVertex.getMinResources().equals(resource4)); assertTrue(iterationTailVertex.getPreferredResources().equals(ResourceSpec.DEFAULT)); assertTrue(feedbackVertex.getMinResources().equals(resource5)); assertTrue(sinkVertex.getPreferredResources().equals(resource6)); assertTrue(iterationSyncVertex.getMinResources().equals(resource3)); }
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; }