src/main/java/it/unibo/bd1819/utils/JobFactory.java
public static Job createDirectorsMovieJoin(final Configuration conf) throws Exception {
        FileSystem fs = FileSystem.get(conf);

        deleteOutputFolder(fs, outputPath);
        deleteOutputFolder(fs, basicprincipalsJoinPath);

        Job joinPrincipalBasicJob = Job.getInstance(conf, "Join between title.principals and title.basics");

        joinPrincipalBasicJob.setReducerClass(FindDirectorsMovieJoinReducer.class);

        joinPrincipalBasicJob.setJarByClass(Main.class);

        joinPrincipalBasicJob.setMapOutputKeyClass(Text.class);
        joinPrincipalBasicJob.setMapOutputValueClass(Text.class);

        joinPrincipalBasicJob.setOutputKeyClass(Text.class);
        joinPrincipalBasicJob.setOutputValueClass(Text.class);

        FileOutputFormat.setOutputPath(joinPrincipalBasicJob, basicprincipalsJoinPath);

        MultipleInputs.addInputPath(joinPrincipalBasicJob, titleBasicsPath,
                KeyValueTextInputFormat.class, FindMovieJoinMapper.class);

        MultipleInputs.addInputPath(joinPrincipalBasicJob,
                titlePrincipalsPath, KeyValueTextInputFormat.class, FindDirectorsJoinMapper.class);
        return joinPrincipalBasicJob;
    }
/**
     * Create a job to join the movies from title.basics and their directors from title.principals
     * @param conf the job configuration
     * @return a hadoop Job.
     * @throws Exception if something is wrong in the process
     */
Create a job to join the movies from title.basics and their directors from title.principals
@param conf the job configuration
@return a hadoop Job.
@throws Exception if something is wrong in the process
Create a job to join the movies from title . basics and their directors from title . principals @param conf the job configuration @return a hadoop Job . @throws Exception if something is wrong in the process
public static Job createDirectorsMovieJoin(final Configuration conf) throws Exception {
        FileSystem fs = FileSystem.get(conf);
        deleteOutputFolder(fs, outputPath);
        deleteOutputFolder(fs, basicprincipalsJoinPath);
        Job joinPrincipalBasicJob = Job.getInstance(conf, "Join between title.principals and title.basics");
        joinPrincipalBasicJob.setReducerClass(FindDirectorsMovieJoinReducer.class);
        joinPrincipalBasicJob.setJarByClass(Main.class);
        joinPrincipalBasicJob.setMapOutputKeyClass(Text.class);
        joinPrincipalBasicJob.setMapOutputValueClass(Text.class);
        joinPrincipalBasicJob.setOutputKeyClass(Text.class);
        joinPrincipalBasicJob.setOutputValueClass(Text.class);
        FileOutputFormat.setOutputPath(joinPrincipalBasicJob, basicprincipalsJoinPath);
        MultipleInputs.addInputPath(joinPrincipalBasicJob, titleBasicsPath,
                KeyValueTextInputFormat.class, FindMovieJoinMapper.class);
        MultipleInputs.addInputPath(joinPrincipalBasicJob,
                titlePrincipalsPath, KeyValueTextInputFormat.class, FindDirectorsJoinMapper.class);
        return joinPrincipalBasicJob;
    }
public static Job createDirectorsMovieJoin ( final Configuration conf ) throws Exception { FileSystem fs = FileSystem . get ( conf ) ; deleteOutputFolder ( fs , outputPath ) ; deleteOutputFolder ( fs , basicprincipalsJoinPath ) ; Job joinPrincipalBasicJob = Job . getInstance ( conf , "Join between title.principals and title.basics" ) ; joinPrincipalBasicJob . setReducerClass ( FindDirectorsMovieJoinReducer . class ) ; joinPrincipalBasicJob . setJarByClass ( Main . class ) ; joinPrincipalBasicJob . setMapOutputKeyClass ( Text . class ) ; joinPrincipalBasicJob . setMapOutputValueClass ( Text . class ) ; joinPrincipalBasicJob . setOutputKeyClass ( Text . class ) ; joinPrincipalBasicJob . setOutputValueClass ( Text . class ) ; FileOutputFormat . setOutputPath ( joinPrincipalBasicJob , basicprincipalsJoinPath ) ; MultipleInputs . addInputPath ( joinPrincipalBasicJob , titleBasicsPath , KeyValueTextInputFormat . class , FindMovieJoinMapper . class ) ; MultipleInputs . addInputPath ( joinPrincipalBasicJob , titlePrincipalsPath , KeyValueTextInputFormat . class , FindDirectorsJoinMapper . class ) ; return joinPrincipalBasicJob ; }
Create a job to join the movies from title.basics and their directors from title.principals
 @param conf the job configuration
 @return a hadoop Job.
Create a job to join the movies from title . basics and their directors from title . principals @param conf the job configuration @return a hadoop Job .