@Test public void testCreateDag() throws Exception { List<JobExecutionPlan> jobExecutionPlans = new ArrayList<>(); for (JobTemplate jobTemplate: this.jobTemplates) { String jobSpecUri = Files.getNameWithoutExtension(new Path(jobTemplate.getUri()).getName()); jobExecutionPlans.add(new JobExecutionPlan(JobSpec.builder(jobSpecUri).withConfig(jobTemplate.getRawTemplateConfig()). withVersion("1").withTemplate(jobTemplate.getUri()).build(), specExecutor)); } Dag<JobExecutionPlan> dag = new JobExecutionPlanDagFactory().createDag(jobExecutionPlans); Assert.assertEquals(dag.getStartNodes().size(), 1); Assert.assertEquals(dag.getEndNodes().size(), 1); Assert.assertEquals(dag.getNodes().size(), 4); String startNodeName = new Path(dag.getStartNodes().get(0).getValue().getJobSpec().getUri()).getName(); Assert.assertEquals(startNodeName, "job1"); String templateUri = new Path(dag.getStartNodes().get(0).getValue().getJobSpec().getTemplateURI().get()).getName(); Assert.assertEquals(templateUri, "job1.job"); String endNodeName = new Path(dag.getEndNodes().get(0).getValue().getJobSpec().getUri()).getName(); Assert.assertEquals(endNodeName, "job4"); templateUri = new Path(dag.getEndNodes().get(0).getValue().getJobSpec().getTemplateURI().get()).getName(); Assert.assertEquals(templateUri, "job4.job"); Dag.DagNode<JobExecutionPlan> startNode = dag.getStartNodes().get(0); List<Dag.DagNode<JobExecutionPlan>> nextNodes = dag.getChildren(startNode); Set<String> nodeSet = new HashSet<>(); for (Dag.DagNode<JobExecutionPlan> node: nextNodes) { nodeSet.add(new Path(node.getValue().getJobSpec().getUri()).getName()); Dag.DagNode<JobExecutionPlan> nextNode = dag.getChildren(node).get(0); Assert.assertEquals(new Path(nextNode.getValue().getJobSpec().getUri()).getName(), "job4"); } Assert.assertTrue(nodeSet.contains("job2")); Assert.assertTrue(nodeSet.contains("job3")); }
public Dag<JobExecutionPlan> createDag(List<JobExecutionPlan> jobExecutionPlans) { Map<String, Dag.DagNode<JobExecutionPlan>> jobExecutionPlanMap = new HashMap<>(); List<Dag.DagNode<JobExecutionPlan>> dagNodeList = new ArrayList<>(); for (JobExecutionPlan jobExecutionPlan : jobExecutionPlans) { Dag.DagNode<JobExecutionPlan> dagNode = new Dag.DagNode<>(jobExecutionPlan); dagNodeList.add(dagNode); String jobName = getJobName(jobExecutionPlan); if (jobName != null) { jobExecutionPlanMap.put(jobName, dagNode); } } for (JobExecutionPlan jobExecutionPlan : jobExecutionPlans) { String jobName = getJobName(jobExecutionPlan); if (jobName == null) { continue; } Dag.DagNode<JobExecutionPlan> node = jobExecutionPlanMap.get(jobName); Collection<String> dependencies = getDependencies(jobExecutionPlan.getJobSpec().getConfig()); for (String dependency : dependencies) { Dag.DagNode<JobExecutionPlan> parentNode = jobExecutionPlanMap.get(dependency); node.addParentNode(parentNode); } } Dag<JobExecutionPlan> dag = new Dag<>(dagNodeList); return dag; }
JobExecutionPlanDagFactory { public Dag<JobExecutionPlan> createDag(List<JobExecutionPlan> jobExecutionPlans) { Map<String, Dag.DagNode<JobExecutionPlan>> jobExecutionPlanMap = new HashMap<>(); List<Dag.DagNode<JobExecutionPlan>> dagNodeList = new ArrayList<>(); for (JobExecutionPlan jobExecutionPlan : jobExecutionPlans) { Dag.DagNode<JobExecutionPlan> dagNode = new Dag.DagNode<>(jobExecutionPlan); dagNodeList.add(dagNode); String jobName = getJobName(jobExecutionPlan); if (jobName != null) { jobExecutionPlanMap.put(jobName, dagNode); } } for (JobExecutionPlan jobExecutionPlan : jobExecutionPlans) { String jobName = getJobName(jobExecutionPlan); if (jobName == null) { continue; } Dag.DagNode<JobExecutionPlan> node = jobExecutionPlanMap.get(jobName); Collection<String> dependencies = getDependencies(jobExecutionPlan.getJobSpec().getConfig()); for (String dependency : dependencies) { Dag.DagNode<JobExecutionPlan> parentNode = jobExecutionPlanMap.get(dependency); node.addParentNode(parentNode); } } Dag<JobExecutionPlan> dag = new Dag<>(dagNodeList); return dag; } }
JobExecutionPlanDagFactory { public Dag<JobExecutionPlan> createDag(List<JobExecutionPlan> jobExecutionPlans) { Map<String, Dag.DagNode<JobExecutionPlan>> jobExecutionPlanMap = new HashMap<>(); List<Dag.DagNode<JobExecutionPlan>> dagNodeList = new ArrayList<>(); for (JobExecutionPlan jobExecutionPlan : jobExecutionPlans) { Dag.DagNode<JobExecutionPlan> dagNode = new Dag.DagNode<>(jobExecutionPlan); dagNodeList.add(dagNode); String jobName = getJobName(jobExecutionPlan); if (jobName != null) { jobExecutionPlanMap.put(jobName, dagNode); } } for (JobExecutionPlan jobExecutionPlan : jobExecutionPlans) { String jobName = getJobName(jobExecutionPlan); if (jobName == null) { continue; } Dag.DagNode<JobExecutionPlan> node = jobExecutionPlanMap.get(jobName); Collection<String> dependencies = getDependencies(jobExecutionPlan.getJobSpec().getConfig()); for (String dependency : dependencies) { Dag.DagNode<JobExecutionPlan> parentNode = jobExecutionPlanMap.get(dependency); node.addParentNode(parentNode); } } Dag<JobExecutionPlan> dag = new Dag<>(dagNodeList); return dag; }  }
JobExecutionPlanDagFactory { public Dag<JobExecutionPlan> createDag(List<JobExecutionPlan> jobExecutionPlans) { Map<String, Dag.DagNode<JobExecutionPlan>> jobExecutionPlanMap = new HashMap<>(); List<Dag.DagNode<JobExecutionPlan>> dagNodeList = new ArrayList<>(); for (JobExecutionPlan jobExecutionPlan : jobExecutionPlans) { Dag.DagNode<JobExecutionPlan> dagNode = new Dag.DagNode<>(jobExecutionPlan); dagNodeList.add(dagNode); String jobName = getJobName(jobExecutionPlan); if (jobName != null) { jobExecutionPlanMap.put(jobName, dagNode); } } for (JobExecutionPlan jobExecutionPlan : jobExecutionPlans) { String jobName = getJobName(jobExecutionPlan); if (jobName == null) { continue; } Dag.DagNode<JobExecutionPlan> node = jobExecutionPlanMap.get(jobName); Collection<String> dependencies = getDependencies(jobExecutionPlan.getJobSpec().getConfig()); for (String dependency : dependencies) { Dag.DagNode<JobExecutionPlan> parentNode = jobExecutionPlanMap.get(dependency); node.addParentNode(parentNode); } } Dag<JobExecutionPlan> dag = new Dag<>(dagNodeList); return dag; }  Dag<JobExecutionPlan> createDag(List<JobExecutionPlan> jobExecutionPlans); }
JobExecutionPlanDagFactory { public Dag<JobExecutionPlan> createDag(List<JobExecutionPlan> jobExecutionPlans) { Map<String, Dag.DagNode<JobExecutionPlan>> jobExecutionPlanMap = new HashMap<>(); List<Dag.DagNode<JobExecutionPlan>> dagNodeList = new ArrayList<>(); for (JobExecutionPlan jobExecutionPlan : jobExecutionPlans) { Dag.DagNode<JobExecutionPlan> dagNode = new Dag.DagNode<>(jobExecutionPlan); dagNodeList.add(dagNode); String jobName = getJobName(jobExecutionPlan); if (jobName != null) { jobExecutionPlanMap.put(jobName, dagNode); } } for (JobExecutionPlan jobExecutionPlan : jobExecutionPlans) { String jobName = getJobName(jobExecutionPlan); if (jobName == null) { continue; } Dag.DagNode<JobExecutionPlan> node = jobExecutionPlanMap.get(jobName); Collection<String> dependencies = getDependencies(jobExecutionPlan.getJobSpec().getConfig()); for (String dependency : dependencies) { Dag.DagNode<JobExecutionPlan> parentNode = jobExecutionPlanMap.get(dependency); node.addParentNode(parentNode); } } Dag<JobExecutionPlan> dag = new Dag<>(dagNodeList); return dag; }  Dag<JobExecutionPlan> createDag(List<JobExecutionPlan> jobExecutionPlans);  }