MapReduce Framework to Implement PageRank Algorithm

Pagerank is the famous algorithm for early Google search engine. The best way to grasp the algorithm is getting one's hands dirty and implement the algorithm.

Simple PageRank Model

The simple network doesn't include any dead ends. Our goal is to get the page ranks for each node in the graph. There are two main assumptions of PageRank:

1. The surfer would randomly jump from one node to another node.

2. The page rank of a node is the sum of ranks of all its source nodes.

The second process involves repeating until the ranks converge. We can choose 3 types of stopping criterion.
1. Iterate mapper and reducer for N times. 
2. Repeat until the ranking of top-N nodes doesn't change.
3. Stop the iterative process when the page rank scores for all of the nodes converge (convergence tolerance=0.005)

In this implementation, I choose the 1st criterion as demo.

mapper.py


#!/usr/bin/env python

import sys

if __name__=="__main__":

    for line in sys.stdin:
        node, neighbour, rank = line.split()
        neighbour_list = neighbour.split(",")
        count = len(neighbour_list)
        print '%s\t[%s]' % (node, neighbour)
        for item in neighbour_list:
            print '%s\t%f' % (item,float(rank)/count)


reducer.py

#!/usr/bin/env python

import sys

if __name__=="__main__":
    alpha = 0.85
    cur_key = None
    cur_value = 0
    #count = 575712
    #count = 2
    neighbour = None
    for line in sys.stdin:
        node, value = line.split()
        if node == cur_key:
            if value.find('[')==-1:
                cur_value += float(value)
            else:
                neighbour = value.lstrip('[').rstrip(']')
        else:
            if cur_key:
                print '%s\t%s\t%s' % (cur_key,neighbour,(1-alpha)*cur_value+alpha)
                neighbour = None

            cur_key = node
            if value.find('[')==-1:
                cur_value = float(value)
            else:
                neighbour = value.lstrip('[').rstrip(']')
    print '%s\t%s\t%s' % (cur_key,neighbour,(1-alpha)*cur_value+alpha)

iterate.sh

#!/bin/bash
echo "the 0th job"
hadoop jar $HADOOP_PREFIX/contrib/streaming/hadoop-streaming-1.2.1.jar -D mapred.reduce.tasks=2 -mapper mapper.py -reducer reducer.py -file mapper.py -file reducer.py -input input/processed.graph5 -output output/output0

for i in $(seq 1 1 100)
do
        echo "the $i th job"
        hadoop jar $HADOOP_PREFIX/contrib/streaming/hadoop-streaming-1.2.1.jar -D mapred.reduce.tasks=2 -mapper mapper.py -reducer reducer.py -file mapper.py -file reducer.py -input output/output$(($i-1)) -output output/output$i
        if [ $i -gt 2 ]
        then
                hadoop dfs -rmr output/output$(($i-2))
                echo 'delete unused file to save disk'
        else
                echo 'do not delete'
        fi
done