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

        FileSystem fs = FileSystem.get(conf);
        Path partitionPath = new Path(Paths.GENERIC_OUTPUT_PATH + "partition", "part.lst");
        deleteOutputFolder(fs, outputPath);
        deleteOutputFolder(fs, partitionPath);

        Job sortJob = Job.getInstance(conf, "Sort Job");

        sortJob.setJarByClass(ScalaMain.class);
        sortJob.setMapperClass(SortMapper.class);
        sortJob.setInputFormatClass(SequenceFileInputFormat.class);


        sortJob.setMapOutputKeyClass(IntWritable.class);
        sortJob.setMapOutputValueClass(Text.class);

        sortJob.setReducerClass(SortReducer.class);
        sortJob.setOutputKeyClass(Text.class);
        sortJob.setOutputValueClass(Text.class);
        FileInputFormat.addInputPath(sortJob, joinDirectorsNamePath);
        FileOutputFormat.setOutputPath(sortJob, outputPath);

        sortJob.setNumReduceTasks(3);
        TotalOrderPartitioner.setPartitionFile(sortJob.getConfiguration(), partitionPath);

        InputSampler.Sampler<IntWritable, Text> sampler = new InputSampler.RandomSampler<>(1, 1000);
        InputSampler.writePartitionFile(sortJob, sampler);

        sortJob.setPartitionerClass(TotalOrderPartitioner.class);

        return sortJob;
    }
/**
     * Sort globally the data emitted by the previous jobs.
     * @param conf the job configuration
     * @return a Hadoop job
     * @throws Exception if something goes wrong.
     */
Sort globally the data emitted by the previous jobs.
@param conf the job configuration
@return a Hadoop job
@throws Exception if something goes wrong.
Sort globally the data emitted by the previous jobs . @param conf the job configuration @return a Hadoop job @throws Exception if something goes wrong .
public static Job createSortJob(final Configuration conf) throws Exception {
        FileSystem fs = FileSystem.get(conf);
        Path partitionPath = new Path(Paths.GENERIC_OUTPUT_PATH + "partition", "part.lst");
        deleteOutputFolder(fs, outputPath);
        deleteOutputFolder(fs, partitionPath);
        Job sortJob = Job.getInstance(conf, "Sort Job");
        sortJob.setJarByClass(ScalaMain.class);
        sortJob.setMapperClass(SortMapper.class);
        sortJob.setInputFormatClass(SequenceFileInputFormat.class);
        sortJob.setMapOutputKeyClass(IntWritable.class);
        sortJob.setMapOutputValueClass(Text.class);
        sortJob.setReducerClass(SortReducer.class);
        sortJob.setOutputKeyClass(Text.class);
        sortJob.setOutputValueClass(Text.class);
        FileInputFormat.addInputPath(sortJob, joinDirectorsNamePath);
        FileOutputFormat.setOutputPath(sortJob, outputPath);
        sortJob.setNumReduceTasks(3);
        TotalOrderPartitioner.setPartitionFile(sortJob.getConfiguration(), partitionPath);
        InputSampler.Sampler<IntWritable, Text> sampler = new InputSampler.RandomSampler<>(1, 1000);
        InputSampler.writePartitionFile(sortJob, sampler);
        sortJob.setPartitionerClass(TotalOrderPartitioner.class);
        return sortJob;
    }
public static Job createSortJob ( final Configuration conf ) throws Exception { FileSystem fs = FileSystem . get ( conf ) ; Path partitionPath = new Path ( Paths . GENERIC_OUTPUT_PATH + "partition" , "part.lst" ) ; deleteOutputFolder ( fs , outputPath ) ; deleteOutputFolder ( fs , partitionPath ) ; Job sortJob = Job . getInstance ( conf , "Sort Job" ) ; sortJob . setJarByClass ( ScalaMain . class ) ; sortJob . setMapperClass ( SortMapper . class ) ; sortJob . setInputFormatClass ( SequenceFileInputFormat . class ) ; sortJob . setMapOutputKeyClass ( IntWritable . class ) ; sortJob . setMapOutputValueClass ( Text . class ) ; sortJob . setReducerClass ( SortReducer . class ) ; sortJob . setOutputKeyClass ( Text . class ) ; sortJob . setOutputValueClass ( Text . class ) ; FileInputFormat . addInputPath ( sortJob , joinDirectorsNamePath ) ; FileOutputFormat . setOutputPath ( sortJob , outputPath ) ; sortJob . setNumReduceTasks ( 3 ) ; TotalOrderPartitioner . setPartitionFile ( sortJob . getConfiguration ( ) , partitionPath ) ; InputSampler . Sampler < IntWritable , Text > sampler = new InputSampler . RandomSampler < > ( 1 , 1000 ) ; InputSampler . writePartitionFile ( sortJob , sampler ) ; sortJob . setPartitionerClass ( TotalOrderPartitioner . class ) ; return sortJob ; }
Sort globally the data emitted by the previous jobs.
Sort globally the data emitted by the previous jobs .