數據挖掘之權重計算（PageRank）

劉  勇  Email:[e-mail]

簡介

　　鑒於在Web抓取服務和文本挖掘之句子向量中對權重值的計算需要，本文基於MapReduce計算模型實現了PageRank算法。為驗證本文算法的有效性，本文采用177萬余條源URL到目標URL鏈接的數據集，並迭代101次來展開測試，測試結果表明：對上述數據集進行測試，總計耗時40.29分鍾。因此，在權重評定的算法設計與實現中引入該思想，具有較好的現實意義。

引言

　　在Web抓取服務中，由於采用多個定向爬蟲對網頁進行抓取，因此其面臨2個重要問題，1）爬蟲的調度問題，不同的爬蟲的抓取頻率決定了獲取該站點的信息數量；2）爬蟲的深度問題，在某個站點內抓取越深，其獲取的信息越陳舊，而爬蟲設計者的本意是及時回頭，即到達一定深度后，返回站點首頁或者回退至上一步。因此，基於上述現實問題，本文作者擬將Google的PageRank算法的排名思想引入至該應用中，通過其PageRank值來確定各站點的抓取頻率，但是采用這種設計的結果，則是設計一個全網爬蟲。

　　在文本挖掘研究中，對句子權重的研究中，擬引入PageRank思想來計算句子向量的權重中，主要是基於句子相互間存在語義關聯。

　　鑒於上述原因，結合Web數據規模日益擴展的需要，采用MapReduce計算模型實現PageRank算法。

 PageRank設計

　　1） 有向圖

　　互聯網中的網頁可以視為有向圖，每一個網頁可以視為一個節點，若網頁A中有一個鏈接到B，則存在一條有向邊AàB。如圖1所示，為一個簡單的網絡鏈接有向圖。

 

圖1 網絡鏈接有向圖

　　根據圖1對PageRank排名思想解釋如下：對於A節點，其存在3條鏈出（A-->B，A-->C，A-->D），因此B、C、D分別獲得A節點PageRank（PR）值的1/3；但是，節點A又存在2條鏈入（B--> A，C-->A），即節點B的PR的1/2和節點C的PR對A有貢獻。其它節點，與之類同，在此不做贅述。

　　根據上述分析，很明顯PageRank算法需要做多次迭代，以期使各節點的PR值趨於穩定（最終達到收斂）。本文作者設計時，擬采用100為最大迭代次數。

　　2） 終止與陷阱

　　若將Web抓取服務（網絡爬蟲）視為馬爾科夫鏈過程，則上述收斂問題，必須滿足有向圖是強關聯的。但是現實網絡世界里，強關聯有向圖不太現實，即有些站點（網址）中的信息中根本沒有鏈接或者鏈接已失效，即面臨節點終止問題，如圖2所示。

 

圖2網絡鏈接有向圖

　　在圖2中，節點C沒有鏈出，即Web抓取服務在C節點面臨終止，其造成PR值在多次迭代之后收斂於0。按照網絡爬蟲設計的本意，當面臨節點終止時，應該跳轉至其它的節點，最常見的做法為跳轉至站點首頁或者返回上一級。

　　此外，有些站點（網址）為了方便向用戶提供服務，會在網頁醒目位置多次嵌入本網頁的鏈接，如門戶網站。但是，若某一個站點（網址）只嵌入本網頁的鏈接，則必然造成網絡爬蟲掉入陷阱中，進入死循環，即面臨節點陷阱問題，如圖3所示。

 

圖3 網絡鏈接有向圖

　　在圖3中，節點C只存在自身的鏈出，即Web抓取服務在C節點陷入死循環，其造成PR值在多次迭代之后，C節點的PR值為1，其它節點的PR值均為0。按照網絡爬蟲設計的本意，當面節點陷阱時，應該跳轉至其它的節點，最常見的做法為跳轉至站點首頁或者返回上一級。

       因此，本文作者綜合上述2種特殊情形，引入以下策略，每個節點，都有權力選擇進入下一個節點，或者回退至上一個節點。具體數學表達式如公式1所示：

          （公式1）

       如公式1所示，α為影響因子， 和分別為當前迭代中某節點的PR值和上一次迭代中某節點的PR。鑒於每個節點均有權利回退至上一個節點，由於表征上一次迭代中所有鏈入之和，故此引入該參數來表征回退至上一次的可能性。

　　3） 算法設計

　　本文在設計中，采用兩級MapReduce計算模型來實現，第一級MapReduce生成網絡鏈接的有向圖；第二級MapReduce用於迭代PR，以確定PR是否收斂。需要指出，本文在設計過程中，Reducer均設定為5個。

　　針對第一級MapReduce，對其Mapper和Reducer簡要描述如下：

　　Mapper:完成源URL和目標URL的標識，如A-->B；

　　Reducer: 根據源URL（有向圖），實現某一節點至其所有鏈出的標識，如A-->B-->C-->D，需要指出，上述標識表示，A-->B，A-->C，A-->D。采用該設計主要是為了節省節點存儲空間。

　　針對第二級MapReduce，對其Mapper和Reducer簡要描述如下：

　　Mapper: 完成鏈出節點及其PR獲取，如B節點存在B-->A，B-->D，則A和D均分B的PR值；

　　Reducer:針對Mapper中每個節點及其PR，對該節點PR值求和，並采用公式1進行量化。

　　其中第二級MapReduce采用多次迭代，若迭代過程中，節點的PR已收斂，則退出迭代。

　　4）  測試結果

　　本文測試環境描述如下，采用10台物理機組成Hadoop集群，CPU：Intel(R) Core(TM) i5-4440 CPU @ 3.10GHz，內存：4G，Hadoop：2.7.1，以上描述為集群的大概配置，其中某個節點的配置可能不一致，本文作者也並未對每個節點進行詳細確認。本文測試數據集采用177萬余條鏈接，迭代101次（迭代次數當時控制失誤，本意為100次，作者比較懶就沒有重新再次迭代了），總計耗時為40.29分鍾。從整體而言，其處理速率在1小時內，還是能夠接受的。

總結

　　本文對基於MapReduce的PageRank算法進行研究與實現，經過實際數據集進行測試，測試結果表明，該測試結果處理速率還是能夠接受的。但是，本文作者的意圖並不是為了實現該算法，而是將該算法的設計思想引入后續Web抓取服務的優化與改進之中，以及后續文本挖掘中對權重值計算的需要之中。

　　程序源代碼：

import java.io.IOException;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

public class GraphSet {
    
    public static class GraphMapper extends Mapper<Object, Text, Text, Text> {
        public static Logger logger = LogManager.getLogger(GraphMapper.class);
        
        public void map(Object key, Text value, Context context) 
        {
            String[] link = value.toString().split("\t");
            try {
                context.write(new Text(link[0]), new Text(link[1]));
            } catch (IOException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    
    
    public static class GraphReducer extends Reducer<Text, Text, NullWritable, Text> {
        public static Logger logger = LogManager.getLogger(GraphReducer.class);
        public void reduce(Text key, Iterable<Text> values, Context context) 
        {
            StringBuilder sb = new StringBuilder();
            sb.append(key.toString());
            for (Text e : values) {
                sb.append("\t");
                sb.append(e.toString());
            }
            
            try {
                context.write(NullWritable.get(), new Text(sb.toString()));
            } catch (IOException e1) {
                e1.printStackTrace();
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }
        }
    }
    
}

import java.io.IOException;
import java.util.Map;
import java.util.HashMap;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

public class PageRank {
    public static double alpha = 0.8;
    
    public static class PageRankMapper extends Mapper<Object, Text, Text, DoubleWritable> {
        public static Map<String, Double> mapPR = new HashMap<String, Double>();
        public static Logger logger = LogManager.getLogger(PageRankMapper.class);
                
        public void map(Object key, Text value, Context context) {
            String[] str = value.toString().split("\t");                 
            int size = str.length;
            double linkOut = mapPR.get(str[0]);
            try {
                for (int i = 0; i < size-1; i++) {
                    DoubleWritable dw = new DoubleWritable();
                    dw.set(linkOut / (size-1)); // count the output of links
                    context.write(new Text(str[i+1]), dw);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    
    
    public static class PageRankReducer extends Reducer<Text, DoubleWritable, NullWritable, Text> {
        public static Logger logger = LogManager.getLogger(PageRankReducer.class);
        private int numberOfUrl = 0;
        
        public void setup(Context context) {
            numberOfUrl = context.getConfiguration().getInt("numberOfUrl", Integer.MAX_VALUE);
        }
        
        
        public void reduce(Text key, Iterable<DoubleWritable> values, Context context) {
            double factor = 0;
            double sum = 0;
            for (DoubleWritable d : values)
                sum += d.get();

            if (PageRankMapper.mapPR.containsKey(key.toString()))
                factor = PageRankMapper.mapPR.get((key.toString()));
            else
                factor = (double)1/(2*numberOfUrl);
            
            sum = alpha*sum + (1-alpha)*factor;  
            String ret = key.toString() + "\t" + String.valueOf(sum);
            try {
                context.write(NullWritable.get(), new Text(ret));
            } catch (IOException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.Map;
import java.util.HashMap;
import java.util.Set;
import java.util.TreeSet;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.gta.graph.GraphSet;
import com.gta.graph.GraphSet.GraphMapper;
import com.gta.graph.GraphSet.GraphReducer;
import com.gta.pagerank.PageRank.PageRankMapper;
import com.gta.pagerank.PageRank.PageRankReducer;

public class PR {
    public static final int MAX = 100;
    public static final int TASK = 5;
    public static final double THRESHOLD = 1e-10; 
    public static final String INPUT_PATH = "hdfs://10.1.130.10:9000/user/hadoop/pagerank/input/";
    public static final String OUTPUT_PATH = "hdfs://10.1.130.10:9000/user/hadoop/pagerank/output/";
    public static final String TMP_PATH = "hdfs://10.1.130.10:9000/user/hadoop/pagerank/tmp/";
    public static Logger logger = LogManager.getLogger(PR.class);
    private Configuration conf = null;
    private int numberOfUrl = 0;
    
    public PR() 
    {
        conf = new Configuration();
    }
    
    
    public void initGraph() throws IOException, InterruptedException, ClassNotFoundException
    {
        Job job = Job.getInstance(conf, "Init Graph");
        job.setJarByClass(GraphSet.class);
        job.setMapperClass(GraphMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setNumReduceTasks(TASK);
        job.setReducerClass(GraphReducer.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);
        FileInputFormat.addInputPath(job, new Path(INPUT_PATH));
        FileOutputFormat.setOutputPath(job, new Path(TMP_PATH));
        job.waitForCompletion(true);
        initPRMap(TMP_PATH);
        conf.setInt("numberOfUrl", numberOfUrl);
    }
    
    
    public void pageRank() throws IOException, InterruptedException, ClassNotFoundException
    {
        int iter = 0;
        while (iter <= MAX) {
            Job job = Job.getInstance(conf, "PageRank");
            job.setJarByClass(PageRank.class);
            job.setMapperClass(PageRankMapper.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(DoubleWritable.class);
            job.setNumReduceTasks(TASK);
            job.setReducerClass(PageRankReducer.class);
            job.setOutputKeyClass(NullWritable.class);
            job.setOutputValueClass(Text.class);
            FileInputFormat.addInputPath(job, new Path(TMP_PATH));
            FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH + iter));
            job.waitForCompletion(true);
            Map<String, Double> newPR = getNewPageRank(OUTPUT_PATH + iter);
            if (compare(PageRankMapper.mapPR, newPR))
                break;
            else {
                for (String key : newPR.keySet()) 
                    PageRankMapper.mapPR.put(key, newPR.get(key));
            }
            
            iter++;
        }
    }
    
    
    public void initPRMap(String filePath) 
    {
        String fileName = filePath + "/part-r-0000";
        Set<String> set = new TreeSet<String>();
        try {
            for (int i = 0; i < TASK; i++) {
                FileSystem fs = FileSystem.get(URI.create((fileName+i)), conf);
                FSDataInputStream is = fs.open(new Path((fileName+i).toString()));
                BufferedReader br = new BufferedReader(new InputStreamReader(is, "UTF-8"));
                String s = null;
                while ((s = br.readLine()) != null) {
                    String[] str = s.split("\t");
                    if (str.length == 0) 
                        break;
                    set.add(str[0]);
                }
                br.close();
            }
            
            numberOfUrl = set.size();
            for (String s : set)
                PageRankMapper.mapPR.put(s, (double)1/numberOfUrl);
            
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    
    
    public Map<String, Double> getNewPageRank(String filePath)
    {
        Map<String, Double> newPR = new HashMap<String, Double>();
        String fileName = filePath + "/part-r-0000";
        try {
            for (int i = 0; i < TASK; i++) {
                FileSystem fs = FileSystem.get(URI.create((fileName+i)), conf);
                FSDataInputStream is = fs.open(new Path((fileName+i).toString()));
                BufferedReader br = new BufferedReader(new InputStreamReader(is, "UTF-8"));
                String s = null;
                while ((s = br.readLine()) != null) {
                    String[] elements = s.split("\t");
                    newPR.put(elements[0], Double.parseDouble(elements[1]));
                }
                br.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return newPR;
    }
    
    
    public boolean compare(Map<String, Double> oldPR, Map<String, Double> newPR) {
        boolean ret = false;
        int newPRSize = newPR.size();
        int oldPRSize = oldPR.size();
        if (oldPRSize == newPRSize) {
            int count = 0;
            for (String key : oldPR.keySet()) {
                if (newPR.containsKey(key)) {
                    if (Math.abs(newPR.get(key) - oldPR.get(key)) <= THRESHOLD)
                        count++;
                }
            }

            if (count == newPRSize)
                ret = true;
        }
        return ret;
    }


    public static void main(String[] args) {
        PR pr = new PR();
        try {
            long start = System.currentTimeMillis();
            pr.initGraph();
            pr.pageRank();
            long end = System.currentTimeMillis();
            PR.logger.info("共耗時： " + (end-start));
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    
}

 　　本文插圖參考：http://www.cnblogs.com/fengfenggirl/p/pagerank-introduction.html

 

 

---

　　作者：志青雲集
　　出處：http://www.cnblogs.com/lyssym
　　如果，您認為閱讀這篇博客讓您有些收獲，不妨點擊一下右下角的【推薦】。
　　如果，您希望更容易地發現我的新博客，不妨點擊一下左下角的【關注我】。
　　如果，您對我的博客所講述的內容有興趣，請繼續關注我的后續博客，我是【志青雲集】。
　　本文版權歸作者和博客園共有，歡迎轉載，但未經作者同意必須保留此段聲明，且在文章頁面明顯位置給出原文連接。

---