//////////////////////////////////////////////////////////////////
// pass all the validation, start to build the dependencies.
//////////////////////////////////////////////////////////////////
Job newJob = new ConfigurableJob (new JobConf (aNewJobConf, this.getClass ()));
newJob.setJobName (aNewJobConf.get ("mapred.job.name", aNewJobConf.get("mapreduce.job.name", "Mobius Job")));
for ( String anInputOfNewJob : inputFolders.split (",") )
{
// Added to track inputs for local PC sampling
inputPaths.add(anInputOfNewJob);
Job dependsOn = jobTopology.get (this.getFS().makeQualified(new Path(anInputOfNewJob)).toUri());
if ( dependsOn != null )
{
List<Job> dependingJobs = newJob.getDependingJobs ();
boolean alreadyInDependency = dependingJobs != null && dependingJobs.contains (dependsOn);
if ( alreadyInDependency )
{
// already added, do nothing.
}
else
{
LOGGER.info (newJob.getJobName () + " depends on " + dependsOn.getJobName ());
newJob.addDependingJob (dependsOn);
}
}
}
// put the output of this <code>newJob</code> into job topology
// so that later if a job read this <code>newJob</code>'s output
// as its input, then the system can detect the dependency.
URI outputPathURI = this.getFS().makeQualified(new Path(outputPath)).toUri();
LOGGER.info("Adding Job:"+newJob.getJobName()+"\tOutput:["+outputPath.toString()+"]");
jobTopology.put (outputPathURI, newJob);
}