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

        List<String> principalsDatasetRecords = new ArrayList<>();
        int moviesDirected = 0;

        for(Text val : values) {
            if (val.toString().contains(FindDirectorsJoinMapper.PRINCIPALS_JOIN_PREFIX)){
                principalsDatasetRecords.add(val.toString()
                        .replace(FindDirectorsJoinMapper.PRINCIPALS_JOIN_PREFIX, EMPTY_STRING));
            }

            if(val.toString().contains(FindMovieJoinMapper.MOVIES_JOIN_PREFIX)) {
                moviesDirected++;
            }

        }

        for(String directorID : principalsDatasetRecords) {
            if(moviesDirected > 0) {
                context.write(key, new Text(directorID + CUSTOM_VALUE_SEPARATOR + moviesDirected));
            }
        }

    }
/**
     * Collect all the values corresponding to a specific key and for each director
     * print the number of movies he directed.
     * @throws IOException
     * @throws InterruptedException
     */
Collect all the values corresponding to a specific key and for each director
print the number of movies he directed.
Collect all the values corresponding to a specific key and for each director print the number of movies he directed .
public void reduce(Text key, Iterable<Text> values,
                       Context context
    ) throws IOException, InterruptedException {
        List<String> principalsDatasetRecords = new ArrayList<>();
        int moviesDirected = 0;
        for(Text val : values) {
            if (val.toString().contains(FindDirectorsJoinMapper.PRINCIPALS_JOIN_PREFIX)){
                principalsDatasetRecords.add(val.toString()
                        .replace(FindDirectorsJoinMapper.PRINCIPALS_JOIN_PREFIX, EMPTY_STRING));
            }
            if(val.toString().contains(FindMovieJoinMapper.MOVIES_JOIN_PREFIX)) {
                moviesDirected++;
            }
        }
        for(String directorID : principalsDatasetRecords) {
            if(moviesDirected > 0) {
                context.write(key, new Text(directorID + CUSTOM_VALUE_SEPARATOR + moviesDirected));
            }
        }
    }
public void reduce ( Text key , Iterable < Text > values , Context context ) throws IOException , InterruptedException { List < String > principalsDatasetRecords = new ArrayList < > ( ) ; int moviesDirected = 0 ; for ( Text val : values ) { if ( val . toString ( ) . contains ( FindDirectorsJoinMapper . PRINCIPALS_JOIN_PREFIX ) ) { principalsDatasetRecords . add ( val . toString ( ) . replace ( FindDirectorsJoinMapper . PRINCIPALS_JOIN_PREFIX , EMPTY_STRING ) ) ; } if ( val . toString ( ) . contains ( FindMovieJoinMapper . MOVIES_JOIN_PREFIX ) ) { moviesDirected ++ ; } } for ( String directorID : principalsDatasetRecords ) { if ( moviesDirected > 0 ) { context . write ( key , new Text ( directorID + CUSTOM_VALUE_SEPARATOR + moviesDirected ) ) ; } } }
Collect all the values corresponding to a specific key and for each director
 print the number of movies he directed.
Collect all the values corresponding to a specific key and for each director print the number of movies he directed .