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); @Override boolean preVisit(PlanNode node); @Override void postVisit(PlanNode node); static final String MERGE_ITERATION_AUX_TASKS_KEY; }
@Test public void testResourcesForChainedOperators() 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); ResourceSpec resource6 = new ResourceSpec(0.6, 600); ResourceSpec resource7 = new ResourceSpec(0.7, 700); 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)); }
@Test public void testResourcesForDeltaIteration() 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); ResourceSpec resource6 = new ResourceSpec(0.6, 600); 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)); }