test map reduce

gistfile1.java

String desc = String.format( "Iteration %s of %s", step, iterations ); 

merge( new Job().setDelegate( IterJob.Map.class )
                .setInput( graph_by_source ,
                           "/pr/out/rank_vector" ,
                           "/pr/out/dangling" ,
                           "/pr/out/nonlinked" ,
                           "broadcast:/pr/out/nr_nodes" )
                .setOutput( "shuffle:default",
                            "broadcast:dangling_rank_sum" )
                .setDescription( desc ) );

reduce( new Job().setDelegate( IterJob.Reduce.class )
                 .setCombiner( IterJob.Combine.class )
                 .setInput( "shuffle:default",
                            "broadcast:/pr/out/nr_nodes",
                            "broadcast:/pr/out/nr_dangling",
                            "broadcast:/pr/out/teleportation_grant" )
                 .setOutput( "/pr/out/rank_vector",
                             "broadcast:rank_sum" )
                 .setDescription( desc ) );

// now reduce the broadcast rank sum to an individual file for analysis
// and reading
reduce( new Job().setDelegate( GlobalRankSumJob.Reduce.class )
                 .setInput( "shuffle:rank_sum" )
                 .setOutput( "/pr/out/rank_sum" ) );

// ***
// 
// write out the new ranking vector
if ( step < iterations - 1 ) {

    // now compute the dangling rank sum for the next iteration

    reduce( TeleportationGrantJob.Reduce.class, 
            new Input( "shuffle:dangling_rank_sum",
                       "broadcast:/pr/out/nr_nodes" ),
            new Output( "/pr/out/teleportation_grant" ) );
    
}

++step;