@@ -0,0 +1,1192 @@
+import org.apache.hadoop.io.Text;%0Aimport org.apache.hadoop.mapreduce.Mapper;%0A%0Aimport java.io.IOException;%0A%0Apublic class Map extends Mapper%3CText, Text, Text, Text%3E %7B%0A%0A%09private Text outKey = new Text();%0A%09private Text outValue = new Text();%0A%09public static boolean debug = false;%0A%0A%09@Override%0A%09protected void map(Text key, Text value, Context context)%0A%09%09%09throws IOException, InterruptedException %7B%0A%0A%09%09context.write(key, value);%0A%0A%09%09Node node = Node.fromMR(value.toString());%0A%0A%09%09if (node.getAdjacentNodeNames() != null%0A%09%09%09%09&& node.getAdjacentNodeNames().length %3E 0) %7B%0A%09%09%09double outboundPageRank = node.getPageRank()%0A%09%09%09%09%09/ (double) node.getAdjacentNodeNames().length;%0A%0A%09%09%09// go through all the nodes and propagate PageRank to them%0A%09%09%09//%0A%09%09%09for (int i = 0; i %3C node.getAdjacentNodeNames().length; i++) %7B%0A%0A%09%09%09%09String neighbor = node.getAdjacentNodeNames()%5Bi%5D;%0A%09%09%09%09if (!neighbor.equals(%22%22)) %7B%0A%09%09%09%09%09outKey.set(neighbor);%0A%0A%09%09%09%09%09Node adjacentNode = new Node()%0A%09%09%09%09%09%09%09.setPageRank(outboundPageRank);%0A%0A%09%09%09%09%09outValue.set(adjacentNode.toString());%0A%09%09%09%09%09if (debug)%0A%09%09%09%09%09%09System.out.println(%22  output -%3E K%5B%22 + outKey + %22%5D,V%5B%22%0A%09%09%09%09%09%09%09%09+ outValue + %22%5D%22);%0A%09%09%09%09%09context.write(outKey, outValue);%0A%09%09%09%09%7D%0A%09%09%09%7D%0A%09%09%7D%0A%09%7D%0A%7D%0A

Create Map.java