pagerank算法及mapreduce中应用

2019-07-23

pagerank 算法

pagerank algorithm 基本讲解

pagerank 在大数据中的引用

pagerank 在mapreduce中的实现

pagerank 算法解析

概念

网页排名算法，最初由google的产品经理Page(至少是之一)发明。

每一个网页都有一个pageRank值，根据这个值进行排序。

实质：

模拟一个人的浏览过程，随机打开一个网页，在网页上花费几分钟，再从此网页上选择一个URL跳转到其它网页浏览，这样这个人无所事事的一直进行浏览网页，pagerank值来形容他呆在一个网页上的几率大小

基本模型

将网页看成一个节点，网页的链接看成一条边

用矩阵来描述整个网络的链接关系（边权值的集合）

pagerank的大小：

1、想象边的权值为投票的票数

2、边指向某个节点看为对某个节点进行投票

3、节点的得到票数之和，就是它的PageRank值

如图所示：

初始pagerank值：PR(A) = 0.1 PR(B) =0.1, PR(C) =0.1, PR(D)=0.1
投票数：A的三条边分别是1/3，B的两条边是1/2，C的一条边是1/1，D的两条边是1/2
所有节点都将自己的票数按照权值分配投给其它节点

矩阵：

M的含义：

i——行号

j——列号

M（i，j）代表：j节点投向i节点的票数

所以一次投票过程描述：

在这种情况下只要多投票几次，一定会收敛

终止条件的参考：

权值的差值的平均值在一定范围以内
得出的权值与上一次的权值没有变化
通过实现测试，观察最终结果应该迭代多少次停止

改进

节点间的构建的网络图可能不符合标准

存在：终止节点问题，陷阱问题

终止节点问题：

有些节点可能没有出边

相当于悠闲的浏览者在浏览完当前页面时候，找不到下一条链接。导致无法执行下一步。

*在数学模型中的体现 *

*最终迭代下去，会让最终的结果趋近零向量 *

陷阱问题

有节点只有跳向自己节点的链接，导致悠闲的浏览者不断的跳向自己，不公平的是自己的重要程度增大

在数学模型中的体现

行列式中一行只有一个元素存在值，其它都是零

最终迭代下去，最后的PR值，只有一个节点是有值的，其它节点的值都是零

解决方案

V‘ = aM V +（1- a）e

原理：

当这个浏览者陷入了麻烦，他不会傻到就在当前页面停留下来或者一直反复浏览自己当前的页面。或者理解为自己会找到方式将自己手中的票投给其它人，而不是留给自己

他会有一定几率跳向其它更目前页面完全没有关系的页面，或者理解为，随便找个人将自己手中的票投给他

（1 - a）——跳向其它不相关页面的概率

（e）—— 当前的PR向量

在mapreduce中的实现

map阶段

从输入端得到的关系，细分为片

计算出矩阵M 中，单个元素与PR向量中一个元素的乘积

按照K ——V 对的关系传递给reducer

以人物关系网的构建为例

输入数据格式：当前节点ID，节点当前的PR值，list（节点指向的ID：边的权值

分割为：

当前节点ID——PR值

节点指向的ID——PR值乘以边的权值

reduce阶段

mapreduce框架自动将ID值相等的分片整合到一起提交给reduce函数

reduce函数将它们相加得到这次迭代后该节点的PR值

输入格式：

节点的ID值，lists（PR值）

输出：

ID值，lists1+lists2+……..

样例函数实现

以大数据统计人物的重要程度为例

map函数

package cn.xkz.examples.PR;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.math.BigDecimal;
import java.text.DecimalFormat;

/**
 * pagerank 算法
 */
public class PRIterMapper extends Mapper<LongWritable, Text,Text,Text> {
    protected void map(LongWritable key,Text value,Context context) throws IOException,InterruptedException{
        String[] strFirst = value.toString().split("\t");
        String nameKey = strFirst[0];
        String[] strSecond = strFirst[1].split("#");
        Double num = Double.parseDouble(strSecond[0]);
        String strRelation = strSecond[1];
        String[] nameList = strSecond[1].split(";");

        for(String strTmp: nameList){
            String[] strSplit = strTmp.split(":");
            Double value_num = Double.valueOf(strSplit[1])*num;
            BigDecimal bigDecimal = BigDecimal.valueOf(value_num);
            DecimalFormat decimalFormat = new DecimalFormat("0.000000");
            context.write(new Text(strSplit[0]),new Text(decimalFormat.format(bigDecimal)));
        }

        //输出关系链
        strRelation = "#" + strRelation;
        context.write(new Text(nameKey),new Text(strRelation));
    }
}

reduce函数

package cn.xkz.examples.PR;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.math.BigDecimal;
import java.text.DecimalFormat;

/**
 * pagerank 算法对新KV统计
 */
public class PRIterReducer extends Reducer<Text,Text,Text,Text> {
    protected void reduce(Text key,Iterable<Text> value,Context context)throws IOException,InterruptedException {
        Double sum  = new Double(0);
        String list = new String();

        for(Text tmp: value){
            String strTmp = tmp.toString();
            if(strTmp.charAt(0) == '#' && strTmp.length() > 0){
                list  = new String(strTmp);
            }else if(strTmp.length() > 0){
                sum += Double.valueOf(strTmp);
            }
        }
        BigDecimal bigDecimal = BigDecimal.valueOf(sum);
        DecimalFormat decimalFormat = new DecimalFormat("0.000000");
        String strRelation = decimalFormat.format(bigDecimal) + list;
        context.write(key, new Text(strRelation));
    }
}

参考文章：

https://www.cnblogs.com/fengfenggirl/p/pagerank-introduction.html

https://juejin.im/post/5ac43ad1f265da238155d6fe