src/main/java/it/unibo/bd1819/reducers/ActorDirectorJoinReducer.java
public void reduce(Text key, Iterable<Text> values,
                       Context context
    ) throws IOException, InterruptedException {

        List<String> directorRecords = new ArrayList<>();
        List<String> actorRecord = new ArrayList<>();

        for(Text record: values) {
            if(record.toString().contains(FilteredDirectorMovieMapper.FILTERED_DIRECTORS_JOIN_PREFIX)) {
                directorRecords.add(record.toString()
                        .replace(FilteredDirectorMovieMapper.FILTERED_DIRECTORS_JOIN_PREFIX, ""));
            }
            if(record.toString().contains(ActorJoinMapper.ACTOR_JOIN_PREFIX)) {
                actorRecord.add(record.toString().replace(ActorJoinMapper.ACTOR_JOIN_PREFIX, ""));
            }
        }

        for(String directorData : directorRecords) {
            for(String actorData: actorRecord) {
                context.write(buildDirectorActorKey(directorData, actorData),
                        buildDirectorActorValue(directorData));
            }
        }
    }
/**
     * Reduce the join between directors and actors.
     * @throws IOException
     * @throws InterruptedException
     */
Reduce the join between directors and actors.
Reduce the join between directors and actors .
public void reduce(Text key, Iterable<Text> values,
                       Context context
    ) throws IOException, InterruptedException {
        List<String> directorRecords = new ArrayList<>();
        List<String> actorRecord = new ArrayList<>();
        for(Text record: values) {
            if(record.toString().contains(FilteredDirectorMovieMapper.FILTERED_DIRECTORS_JOIN_PREFIX)) {
                directorRecords.add(record.toString()
                        .replace(FilteredDirectorMovieMapper.FILTERED_DIRECTORS_JOIN_PREFIX, ""));
            }
            if(record.toString().contains(ActorJoinMapper.ACTOR_JOIN_PREFIX)) {
                actorRecord.add(record.toString().replace(ActorJoinMapper.ACTOR_JOIN_PREFIX, ""));
            }
        }
        for(String directorData : directorRecords) {
            for(String actorData: actorRecord) {
                context.write(buildDirectorActorKey(directorData, actorData),
                        buildDirectorActorValue(directorData));
            }
        }
    }
public void reduce ( Text key , Iterable < Text > values , Context context ) throws IOException , InterruptedException { List < String > directorRecords = new ArrayList < > ( ) ; List < String > actorRecord = new ArrayList < > ( ) ; for ( Text record : values ) { if ( record . toString ( ) . contains ( FilteredDirectorMovieMapper . FILTERED_DIRECTORS_JOIN_PREFIX ) ) { directorRecords . add ( record . toString ( ) . replace ( FilteredDirectorMovieMapper . FILTERED_DIRECTORS_JOIN_PREFIX , "" ) ) ; } if ( record . toString ( ) . contains ( ActorJoinMapper . ACTOR_JOIN_PREFIX ) ) { actorRecord . add ( record . toString ( ) . replace ( ActorJoinMapper . ACTOR_JOIN_PREFIX , "" ) ) ; } } for ( String directorData : directorRecords ) { for ( String actorData : actorRecord ) { context . write ( buildDirectorActorKey ( directorData , actorData ) , buildDirectorActorValue ( directorData ) ) ; } } }
Reduce the join between directors and actors.
Reduce the join between directors and actors .