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

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

        Job joinActorsName = Job.getInstance(conf, "Join between Names and Actors");

        joinActorsName.setReducerClass(ActorsNameJoinReducer.class);

        joinActorsName.setJarByClass(Main.class);


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

        FileOutputFormat.setOutputPath(joinActorsName, joinActorsNamePath);

        MultipleInputs.addInputPath(joinActorsName, threeActorsDirectorPath,
                KeyValueTextInputFormat.class, ActorNameJoinMapper.class);

        MultipleInputs.addInputPath(joinActorsName, nameBasicsPath,
                KeyValueTextInputFormat.class, NameJoinerMapper.class);
        return joinActorsName;
    }
/**
     * Join actors id with their names.
     * @param conf the job configuration
     * @return a Hadoop job
     * @throws Exception if something goes wrong.
     */
Join actors id with their names.
@param conf the job configuration
@return a Hadoop job
@throws Exception if something goes wrong.
Join actors id with their names . @param conf the job configuration @return a Hadoop job @throws Exception if something goes wrong .
public static Job createActorsNameJoin(final Configuration conf) throws Exception {
        FileSystem fs = FileSystem.get(conf);
        deleteOutputFolder(fs, joinActorsNamePath);
        Job joinActorsName = Job.getInstance(conf, "Join between Names and Actors");
        joinActorsName.setReducerClass(ActorsNameJoinReducer.class);
        joinActorsName.setJarByClass(Main.class);
        joinActorsName.setOutputKeyClass(Text.class);
        joinActorsName.setOutputValueClass(Text.class);
        FileOutputFormat.setOutputPath(joinActorsName, joinActorsNamePath);
        MultipleInputs.addInputPath(joinActorsName, threeActorsDirectorPath,
                KeyValueTextInputFormat.class, ActorNameJoinMapper.class);
        MultipleInputs.addInputPath(joinActorsName, nameBasicsPath,
                KeyValueTextInputFormat.class, NameJoinerMapper.class);
        return joinActorsName;
    }
public static Job createActorsNameJoin ( final Configuration conf ) throws Exception { FileSystem fs = FileSystem . get ( conf ) ; deleteOutputFolder ( fs , joinActorsNamePath ) ; Job joinActorsName = Job . getInstance ( conf , "Join between Names and Actors" ) ; joinActorsName . setReducerClass ( ActorsNameJoinReducer . class ) ; joinActorsName . setJarByClass ( Main . class ) ; joinActorsName . setOutputKeyClass ( Text . class ) ; joinActorsName . setOutputValueClass ( Text . class ) ; FileOutputFormat . setOutputPath ( joinActorsName , joinActorsNamePath ) ; MultipleInputs . addInputPath ( joinActorsName , threeActorsDirectorPath , KeyValueTextInputFormat . class , ActorNameJoinMapper . class ) ; MultipleInputs . addInputPath ( joinActorsName , nameBasicsPath , KeyValueTextInputFormat . class , NameJoinerMapper . class ) ; return joinActorsName ; }
Join actors id with their names.
Join actors id with their names .