Mapreduce -- PageRank_ZenDei技術網路在線

PageRank 簡單理解為網頁排名，但是網頁是根據什麼排名的，接下來就簡單介紹一下。舉例：假設網頁 A 的內容中有網頁 B，C 和 D 的鏈接，並且 A 的 PageRank的值為0.25。那接下里我們就可以計算在網頁 A 中的其他網頁的PageRank的值了。我們拿網頁 B 來進行說明， ...

PageRank 簡單理解為網頁排名，但是網頁是根據什麼排名的，接下來就簡單介紹一下。

舉例：

假設網頁 A 的內容中有網頁 B，C 和 D 的鏈接，並且 A 的 PageRank的值為0.25。

那接下里我們就可以計算在網頁 A 中的其他網頁的PageRank的值了。我們拿網頁 B 來進行說明，

在網頁 A 中的網頁 B 的 PageRank 為 0.25 * (1/n) 其中n為網頁 A 中網頁鏈接數，結果則為 0.25*(1/3)。

可以簡單理解為A的PageRank被B，C 和 D 平分了，B分到了0.25的三分之一。

然後將所有網頁中的關於網頁B的pagerank值求出來，就是網頁B真實的pagerank了。

但是上面的例子沒有考慮到如下的特殊情況：

1 網頁A中只有指向自己的網頁鏈接。

2 網頁A中沒有任何鏈接。

如果出現以上情況，會導致pagerank結果不准確。

所以出現了下麵的公式：

result = sum * n + (1-n)/N

sum 為上面計算出來的，如網頁B在所有網頁中的pagerank值的總和。

n 可以理解為停留在當前網頁繼續進行網頁跳轉瀏覽的概率

1-n 可以理解為不訪問當前網頁的任何鏈接，從瀏覽器的地址欄輸入，轉去其他網頁的概率。

N 為網頁的數量

下麵介紹通過MapReduce實現PageRank

簡單的流程分析：

Map

取一行數據進行說明

A    B    C    D

網頁A中有網頁B,C,D的鏈接;

剛開始給網頁A一個預設的pagerank值，然後根據這個值計算其他網頁鏈接在網頁A中的PageRank。處理後的數據如下：

A    0.25    B    C    D
B    0.25*(1/3)
C    0.25*(1/3)
D    0.25*(1/3)

Reduce

然後通過 Reduce 計算出各個網頁在其他網頁中的 PageRank 的總和 sum，然後代入公式計算實際的PageRank，並更新 A 0.25 B C D 中的數據，這裡將0.25更新為計算出來真實的 PageRank。

重覆計算各個網頁的 PageRank 的值，直到 PageRank 的結果收斂，值趨於穩定。

A    0.25    B    C    D  =>A    ?    B    C    D

測試數據：

A    B       C       D
B    A       D
C    C
D    B       C

測試代碼：

RunJob.java

  1 import org.apache.hadoop.conf.Configuration;
  2 import org.apache.hadoop.fs.FileSystem;
  3 import org.apache.hadoop.fs.Path;
  4 import org.apache.hadoop.io.Text;
  5 import org.apache.hadoop.mapreduce.Job;
  6 import org.apache.hadoop.mapreduce.Mapper;
  7 import org.apache.hadoop.mapreduce.Reducer;
  8 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  9 import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
 10 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
 11 
 12 import java.io.IOException;
 13 
 14 /**
 15  * Created by Edward on 2016/7/13.
 16  */
 17 public class RunJob {
 18 
 19     public static enum ValueEnum{
 20         CLOSURE_VALUE;
 21     }
 22 
 23     public static void main(String[] args)
 24     {
 25 
 26         //access hdfs's user
 27         System.setProperty("HADOOP_USER_NAME","root");
 28 
 29         Configuration conf = new Configuration();
 30         conf.set("fs.defaultFS", "hdfs://node1:8020");
 31 
 32         try {
 33             int i = 0;
 34             while(true) {
 35                 i++;
 36                 conf.setInt("count", i);
 37 
 38                 FileSystem fs = FileSystem.get(conf);
 39                 Job job = Job.getInstance(conf);
 40                 job.setJarByClass(RunJob.class);
 41                 job.setMapperClass(MyMapper.class);
 42                 job.setReducerClass(MyReducer.class);
 43 
 44                 //需要指定 map out 的 key 和 value
 45                 job.setOutputKeyClass(Text.class);
 46                 job.setOutputValueClass(Text.class);
 47 
 48                 //設置輸入類的
 49                 job.setInputFormatClass(KeyValueTextInputFormat.class);
 50                 if(i==1)
 51                     FileInputFormat.addInputPath(job, new Path("/test/pagerank/input"));
 52                 else
 53                     FileInputFormat.addInputPath(job, new Path("/test/pagerank/output/pr"+(i-1)));
 54 
 55                 Path path = new Path("/test/pagerank/output/pr"+i);
 56                 if (fs.exists(path))//如果目錄存在，則刪除目錄
 57                 {
 58                     fs.delete(path, true);
 59                 }
 60                 FileOutputFormat.setOutputPath(job, path);
 61 
 62                 boolean b = job.waitForCompletion(true);
 63                 if (b) {
 64                     long closure =job.getCounters().findCounter(ValueEnum.CLOSURE_VALUE).getValue();
 65                     double avg= closure/4000.0;//計算收斂的平均值，浮動小於0.001則認為收斂
 66                     System.out.println("執行第"+i+"次, closure="+closure+",avg="+avg);
 67                     if(avg < 0.001){
 68                         System.out.println("總共執行了"+i+"次，之後收斂");
 69                         break;
 70                     }
 71                 }
 72             }
 73 
 74         } catch (Exception e) {
 75             e.printStackTrace();
 76         }
 77     }
 78 
 79 
 80     public static class MyMapper extends Mapper<Text, Text, Text, Text> {
 81         @Override
 82         protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
 83             AdjacentNodes adjacentNodes  = new AdjacentNodes();
 84             int count = context.getConfiguration().getInt("count", 1);
 85 
 86             if(count == 1)
 87             {
 88                 AdjacentNodes firstAdj = new AdjacentNodes();
 89                 firstAdj.setValue(1.0);
 90                 firstAdj.setNodes(value.toString().split("\t"));
 91                 adjacentNodes = firstAdj;
 92             }
 93             else
 94             {
 95                 //格式化 value: 1.0 B C D
 96                 adjacentNodes.formatInfo(value.toString());
 97             }
 98             //A 1.0 B C D
 99             context.write(key, new Text(adjacentNodes.toString()));
100 
101             double pagerank = adjacentNodes.getValue()/adjacentNodes.getNum();
102             for(int i=0; i<adjacentNodes.getNum(); i++)
103             {
104                 String node = adjacentNodes.getNodes()[i];
105                 //B 0.333
106                 context.write(new Text(node), new Text(pagerank+""));
107             }
108         }
109     }
110 
111     public static class MyReducer extends Reducer<Text, Text, Text, Text> {
112         @Override
113         protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
114 
115             AdjacentNodes adjacentNodes = new AdjacentNodes();
116 
117             Double sum = 0.0;
118             for(Text adj : values)
119             {
120                 String str = adj.toString();
121                 if(str.split("\t").length>1)
122                 {
123                     adjacentNodes.formatInfo(str);
124                 }
125                 else{
126                     sum += Double.parseDouble(str);   //對節點的 pagerank 求和，
127                 }
128             }
129 
130             //計算pagerank
131             double n = 0.80;
132             double pagerank = sum * n + (1-n)/4.0;// 計算pagerank 公式 sum * n + (1-n)/N
133 
134             //計算收斂的差值
135             int closure =(int)(Math.abs(pagerank - adjacentNodes.getValue()) * 1000);
136             //通過context.getCounter(ENUM)方法，每次執行reduce將closure的值進行累加，結果傳遞給主函數，
137 
138             context.getCounter(ValueEnum.CLOSURE_VALUE).increment(closure);
139 
140             adjacentNodes.setValue(pagerank);
141             context.write(key, new Text(adjacentNodes.toString()));
142         }
143     }
144 }

AdjacentNodes.java

/**
 * Created by Edward on 2016/7/14.
 */
public class AdjacentNodes {

    double value = 0.0;
    int num = 0;
    String[] nodes = null;

    public void formatInfo(String str)
    {
        String[] val = str.split("\t");
        this.setValue(Double.parseDouble(val[0]));
        this.setNum(val.length-1);

        if(this.num != 0)
            this.nodes = new String[this.num];

        for(int i=1; i<val.length; i++)
        {
            this.nodes[i-1] = val[i];
        }
    }

    public boolean isEmpty()
    {
        if(this.num == 0)
            return true;
        else
            return false;
    }

    public void setNodes(String[] nodes) {
        this.nodes = nodes;
        this.num = nodes.length;
    }

    public void setNum(int num) {
        this.num = num;
    }

    public void setValue(double value) {
        this.value = value;
    }

    public double getValue() {
        return value;
    }

    public int getNum() {
        return num;
    }

    public String[] getNodes() {
        return nodes;
    }


    @Override
    public String toString() {

        StringBuffer stringBuffer = new StringBuffer(this.value+"");

        for(int i=0; i<this.num; i++)
        {
            stringBuffer.append("\t"+this.nodes[i]);
        }

        return stringBuffer.toString();
    }
}

結果數據：

A       0.10135294176208584     B       C       D
B       0.12838069628609527     A       D
C       0.6560527651143326      C
D       0.12838069628609527     B       C

總共執行了24次，之後收斂;

PageRank值越高，越值得推薦。