@@ -0,0 +1,1880 @@
+import org.apache.hadoop.io.Text;%0Aimport org.apache.hadoop.mapreduce.Reducer;%0A%0Aimport java.io.IOException;%0A%0Apublic class Reduce extends Reducer%3CText, Text, Text, Text%3E %7B%0A%0A%09public static final double CONVERGENCE_SCALING_FACTOR = 1000.0;%0A%09public static final double DAMPING_FACTOR = 0.85;%0A%09public static String CONF_NUM_NODES_GRAPH = %22pagerank.numnodes%22;%0A%09private int numberOfNodesInGraph;%0A%09public static final boolean debug = false;%0A%0A%09public static enum Counter %7B%0A%09%09CONV_DELTAS%0A%09%7D%0A%0A%09@Override%0A%09protected void setup(Context context) throws IOException,%0A%09%09%09InterruptedException %7B%0A%09%09numberOfNodesInGraph = context.getConfiguration().getInt(%0A%09%09%09%09CONF_NUM_NODES_GRAPH, 0);%0A%09%7D%0A%0A%09private Text outValue = new Text();%0A%0A%09public void reduce(Text key, Iterable%3CText%3E values, Context context)%0A%09%09%09throws IOException, InterruptedException %7B%0A%0A%09%09if (debug)%0A%09%09%09System.out.println(%22input -%3E K%5B%22 + key + %22%5D%22);%0A%0A%09%09double summedPageRanks = 0;%0A%09%09Node originalNode = new Node();%0A%0A%09%09for (Text textValue : values) %7B%0A%09%09%09if (debug)%0A%09%09%09%09System.out.println(%22  input -%3E V%5B%22 + textValue + %22%5D%22);%0A%09%09%09Node node = Node.fromMR(textValue.toString());%0A%0A%09%09%09if (node.containsAdjacentNodes()) %7B%0A%09%09%09%09originalNode = node;%0A%09%09%09%7D else %7B%0A%09%09%09%09summedPageRanks += node.getPageRank();%0A%09%09%09%7D%0A%09%09%7D%0A%0A%09%09double dampingFactor = ((1.0 - DAMPING_FACTOR) / (double) numberOfNodesInGraph);%0A%0A%09%09double newPageRank = dampingFactor + (DAMPING_FACTOR * summedPageRanks);%0A%0A%09%09double delta = originalNode.getPageRank() - newPageRank;%0A%0A%09%09originalNode.setPageRank(newPageRank);%0A%0A%09%09outValue.set(originalNode.toString());%0A%0A%09%09if (debug)%0A%09%09%09System.out%0A%09%09%09%09%09.println(%22  output -%3E K%5B%22 + key + %22%5D,V%5B%22 + outValue + %22%5D%22);%0A%09%09context.write(key, outValue);%0A%0A%09%09int scaledDelta = Math.abs((int) (delta * CONVERGENCE_SCALING_FACTOR));%0A%0A%09%09if (debug)%0A%09%09%09System.out.println(%22Delta = %22 + scaledDelta);%0A%0A%09%09context.getCounter(Counter.CONV_DELTAS).increment(scaledDelta);%0A%09%7D%0A%7D%0A

Create Reduce.java