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

        deleteOutputFolder(fs, directorActorsJoinPath);

        Job joinDirectorsActor = Job.getInstance(conf, "Join between Actors and Directors");

        joinDirectorsActor.setReducerClass(ActorDirectorJoinReducer.class);

        joinDirectorsActor.setJarByClass(Main.class);

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

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

        FileOutputFormat.setOutputPath(joinDirectorsActor, directorActorsJoinPath);

        MultipleInputs.addInputPath(joinDirectorsActor, aggregateDirectorPath,
                KeyValueTextInputFormat.class, FilteredDirectorMovieMapper.class);

        MultipleInputs.addInputPath(joinDirectorsActor, titlePrincipalsPath,
                KeyValueTextInputFormat.class, ActorJoinMapper.class);
        return joinDirectorsActor;
    }
/**
     * Join the director-movies table with the actors from title.principals
     * @param conf the job configuration.
     * @return an Hadoop job
     * @throws Exception if something goes wrong.
     */
Join the director-movies table with the actors from title.principals
@param conf the job configuration.
@return an Hadoop job
@throws Exception if something goes wrong.
Join the director - movies table with the actors from title . principals @param conf the job configuration . @return an Hadoop job @throws Exception if something goes wrong .
public static Job createDirectorsActorsJoin(final Configuration conf) throws Exception {
        FileSystem fs = FileSystem.get(conf);
        deleteOutputFolder(fs, aggregateDirectorPath);
        deleteOutputFolder(fs, directorActorsJoinPath);
        Job joinDirectorsActor = Job.getInstance(conf, "Join between Actors and Directors");
        joinDirectorsActor.setReducerClass(ActorDirectorJoinReducer.class);
        joinDirectorsActor.setJarByClass(Main.class);
        joinDirectorsActor.setMapOutputKeyClass(Text.class);
        joinDirectorsActor.setMapOutputValueClass(Text.class);
        joinDirectorsActor.setOutputKeyClass(Text.class);
        joinDirectorsActor.setOutputValueClass(Text.class);
        FileOutputFormat.setOutputPath(joinDirectorsActor, directorActorsJoinPath);
        MultipleInputs.addInputPath(joinDirectorsActor, aggregateDirectorPath,
                KeyValueTextInputFormat.class, FilteredDirectorMovieMapper.class);
        MultipleInputs.addInputPath(joinDirectorsActor, titlePrincipalsPath,
                KeyValueTextInputFormat.class, ActorJoinMapper.class);
        return joinDirectorsActor;
    }
public static Job createDirectorsActorsJoin ( final Configuration conf ) throws Exception { FileSystem fs = FileSystem . get ( conf ) ; deleteOutputFolder ( fs , aggregateDirectorPath ) ; deleteOutputFolder ( fs , directorActorsJoinPath ) ; Job joinDirectorsActor = Job . getInstance ( conf , "Join between Actors and Directors" ) ; joinDirectorsActor . setReducerClass ( ActorDirectorJoinReducer . class ) ; joinDirectorsActor . setJarByClass ( Main . class ) ; joinDirectorsActor . setMapOutputKeyClass ( Text . class ) ; joinDirectorsActor . setMapOutputValueClass ( Text . class ) ; joinDirectorsActor . setOutputKeyClass ( Text . class ) ; joinDirectorsActor . setOutputValueClass ( Text . class ) ; FileOutputFormat . setOutputPath ( joinDirectorsActor , directorActorsJoinPath ) ; MultipleInputs . addInputPath ( joinDirectorsActor , aggregateDirectorPath , KeyValueTextInputFormat . class , FilteredDirectorMovieMapper . class ) ; MultipleInputs . addInputPath ( joinDirectorsActor , titlePrincipalsPath , KeyValueTextInputFormat . class , ActorJoinMapper . class ) ; return joinDirectorsActor ; }
Join the director-movies table with the actors from title.principals
 @param conf the job configuration.
Join the director - movies table with the actors from title . principals @param conf the job configuration .