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

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

        Job joinDirectorsName = Job.getInstance(conf, "Join between Names and Directors");

        joinDirectorsName.setReducerClass(DirectorsNameReducer.class);

        joinDirectorsName.setJarByClass(ScalaMain.class);

        joinDirectorsName.setMapOutputKeyClass(Text.class);
        joinDirectorsName.setMapOutputValueClass(Text.class);
        joinDirectorsName.setOutputKeyClass(IntWritable.class);
        joinDirectorsName.setOutputValueClass(Text.class);

        FileOutputFormat.setOutputPath(joinDirectorsName, joinDirectorsNamePath);

        MultipleInputs.addInputPath(joinDirectorsName, joinActorsNamePath,
                KeyValueTextInputFormat.class, DirectorNameJoinMapper.class);

        MultipleInputs.addInputPath(joinDirectorsName, nameBasicsPath,
                KeyValueTextInputFormat.class, NameJoinerMapper.class);
        joinDirectorsName.setOutputFormatClass(SequenceFileOutputFormat.class);
        return joinDirectorsName;
    }
/**
     * Join director ids with their names
     * @param conf the job configuration
     * @return an Hadoop job
     * @throws Exception if something goes wronh.
     */
Join director ids with their names
@param conf the job configuration
@return an Hadoop job
@throws Exception if something goes wronh.
Join director ids with their names @param conf the job configuration @return an Hadoop job @throws Exception if something goes wronh .
public static Job createDirectorsNameJoin(final Configuration conf) throws Exception {
        FileSystem fs = FileSystem.get(conf);
        deleteOutputFolder(fs, joinDirectorsNamePath);
        Job joinDirectorsName = Job.getInstance(conf, "Join between Names and Directors");
        joinDirectorsName.setReducerClass(DirectorsNameReducer.class);
        joinDirectorsName.setJarByClass(ScalaMain.class);
        joinDirectorsName.setMapOutputKeyClass(Text.class);
        joinDirectorsName.setMapOutputValueClass(Text.class);
        joinDirectorsName.setOutputKeyClass(IntWritable.class);
        joinDirectorsName.setOutputValueClass(Text.class);
        FileOutputFormat.setOutputPath(joinDirectorsName, joinDirectorsNamePath);
        MultipleInputs.addInputPath(joinDirectorsName, joinActorsNamePath,
                KeyValueTextInputFormat.class, DirectorNameJoinMapper.class);
        MultipleInputs.addInputPath(joinDirectorsName, nameBasicsPath,
                KeyValueTextInputFormat.class, NameJoinerMapper.class);
        joinDirectorsName.setOutputFormatClass(SequenceFileOutputFormat.class);
        return joinDirectorsName;
    }
public static Job createDirectorsNameJoin ( final Configuration conf ) throws Exception { FileSystem fs = FileSystem . get ( conf ) ; deleteOutputFolder ( fs , joinDirectorsNamePath ) ; Job joinDirectorsName = Job . getInstance ( conf , "Join between Names and Directors" ) ; joinDirectorsName . setReducerClass ( DirectorsNameReducer . class ) ; joinDirectorsName . setJarByClass ( ScalaMain . class ) ; joinDirectorsName . setMapOutputKeyClass ( Text . class ) ; joinDirectorsName . setMapOutputValueClass ( Text . class ) ; joinDirectorsName . setOutputKeyClass ( IntWritable . class ) ; joinDirectorsName . setOutputValueClass ( Text . class ) ; FileOutputFormat . setOutputPath ( joinDirectorsName , joinDirectorsNamePath ) ; MultipleInputs . addInputPath ( joinDirectorsName , joinActorsNamePath , KeyValueTextInputFormat . class , DirectorNameJoinMapper . class ) ; MultipleInputs . addInputPath ( joinDirectorsName , nameBasicsPath , KeyValueTextInputFormat . class , NameJoinerMapper . class ) ; joinDirectorsName . setOutputFormatClass ( SequenceFileOutputFormat . class ) ; return joinDirectorsName ; }
Join director ids with their names
 @param conf the job configuration
 @return an Hadoop job
 @throws Exception if something goes wronh.
Join director ids with their names @param conf the job configuration @return an Hadoop job @throws Exception if something goes wronh .