src/main/java/it/unibo/bd1819/utils/JobFactory.java
public static Job createAggregatorJob(final Configuration conf) throws Exception {

        FileSystem fs = FileSystem.get(conf);
        deleteOutputFolder(fs, aggregateDirectorPath);

        Job aggregationJob = Job.getInstance(conf, "Aggregation job for the directors");

        aggregationJob.setJarByClass(Main.class);
        aggregationJob.setMapperClass(AggregateDirectorsMapper.class);
        aggregationJob.setInputFormatClass(KeyValueTextInputFormat.class);

        aggregationJob.setReducerClass(AggregateDirectorsReducer.class);
        aggregationJob.setOutputKeyClass(Text.class);
        aggregationJob.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(aggregationJob, basicprincipalsJoinPath);
        FileOutputFormat.setOutputPath(aggregationJob, aggregateDirectorPath);

        return aggregationJob;
    }
/**
     * Create a job for counting how many films has a director made
     * @param conf the job configuration.
     * @return an hadoop Job.
     * @throws Exception if something is wrong in the process.
     */
Create a job for counting how many films has a director made
@param conf the job configuration.
@return an hadoop Job.
@throws Exception if something is wrong in the process.
Create a job for counting how many films has a director made @param conf the job configuration . @return an hadoop Job . @throws Exception if something is wrong in the process .
public static Job createAggregatorJob(final Configuration conf) throws Exception {
        FileSystem fs = FileSystem.get(conf);
        deleteOutputFolder(fs, aggregateDirectorPath);
        Job aggregationJob = Job.getInstance(conf, "Aggregation job for the directors");
        aggregationJob.setJarByClass(Main.class);
        aggregationJob.setMapperClass(AggregateDirectorsMapper.class);
        aggregationJob.setInputFormatClass(KeyValueTextInputFormat.class);
        aggregationJob.setReducerClass(AggregateDirectorsReducer.class);
        aggregationJob.setOutputKeyClass(Text.class);
        aggregationJob.setOutputValueClass(Text.class);
        FileInputFormat.addInputPath(aggregationJob, basicprincipalsJoinPath);
        FileOutputFormat.setOutputPath(aggregationJob, aggregateDirectorPath);
        return aggregationJob;
    }
public static Job createAggregatorJob ( final Configuration conf ) throws Exception { FileSystem fs = FileSystem . get ( conf ) ; deleteOutputFolder ( fs , aggregateDirectorPath ) ; Job aggregationJob = Job . getInstance ( conf , "Aggregation job for the directors" ) ; aggregationJob . setJarByClass ( Main . class ) ; aggregationJob . setMapperClass ( AggregateDirectorsMapper . class ) ; aggregationJob . setInputFormatClass ( KeyValueTextInputFormat . class ) ; aggregationJob . setReducerClass ( AggregateDirectorsReducer . class ) ; aggregationJob . setOutputKeyClass ( Text . class ) ; aggregationJob . setOutputValueClass ( Text . class ) ; FileInputFormat . addInputPath ( aggregationJob , basicprincipalsJoinPath ) ; FileOutputFormat . setOutputPath ( aggregationJob , aggregateDirectorPath ) ; return aggregationJob ; }
Create a job for counting how many films has a director made
 @param conf the job configuration.
Create a job for counting how many films has a director made @param conf the job configuration .