PageRank的MapReduce實現(xiàn)

1、12112( )( )( )( )( ) (1)(.) (1)( )( )( )( )nniiniPRtPRtPRtPRtPR AaaaaC tC tC tC t Reduce:-input: key: “index.html” value: “1.html”,”n.html”-output: key: ”index.html” value: “1.0 1.html,n.html”說明：Hadoop把Map函數(shù)輸出的key合并，相同key的value合并成一個集合作為reduce的value。輸出key網頁的PR值（初值為1.0） 對于MapReduce程序，Map函數(shù)的輸入為用戶指定的文件，

2、那么輸入的value值就是文件內容（可以配置成按行讀取或者把整個文件視作一個大字符串），key值是讀入文本的偏移量。程序員對輸入的key,value進行一系列操作，然后按(key,value)鍵值對的形式輸出。需要注意的是：需要注意的是：輸出的輸出的key,value是程序員自己定義的，可以和輸入的是程序員自己定義的，可以和輸入的key,value毫不相關。毫不相關。 系統(tǒng)獲取Map函數(shù)的輸出，把相同的key合并，再把key和value集合作為鍵值對作為reduce函數(shù)的輸入。程序員自定義reduce函數(shù)中的處理方法，輸出(key,value)鍵值對到磁盤文件。Reduce:-input: K

3、ey: “1.html” Value: ”index.html 0.5 23”,”2.html 2.4 2”, key: “2.html” value: “index.html 0.5 23”, “1.html 1.3 3”,-output: key:”1.html” value:“ index.html,2.html” 注意，這是注意，這是1.html的新的新PR值值Reduce:-input: key: word value: Reduce函數(shù)的操作： 對value中的網頁根據(jù)pageRank排序。-output: key: word value: 排序后的網頁 public static

4、 void main(String args) throws Exception Configuration conf = new Configuration(); String otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); if (otherArgs.length != 2) System.err.println(Usage: wordcount ); System.exit(2); Job job = new Job(conf, word count); job.setJarByClass(Word

5、Count.class); job.setMapperClass(TokenizerMapper.class); job.setCombinerClass(IntSumReducer.class); job.setReducerClass(IntSumReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(otherArgs0); FileOutputFormat.setOut

6、putPath(job, new Path(otherArgs1); System.exit(job.waitForCompletion(true) ? 0 : 1); 1 private final static IntWritable one = new IntWritable(1);2 private Text word = new Text(); 3 public void map(Object key, Text value, Context context) 4throws IOException, InterruptedException 5 StringTokenizer it

7、r = new StringTokenizer(value.toString();6 while (itr.hasMoreTokens() 7 word.set(itr.nextToken();8 context.write(word, one); 第1,2行，IntWritable和Text類我們可以看作就是一個int類型和一個string類型。由于hadoop使用了JAVA RPC機制來實現(xiàn)通訊，所以調用的基本類型需要用對象來傳遞。因此Hadoop就用IntWritable和Text類來對int和string進行封裝，對int和文本內容進行存儲、編碼、解碼，以此確保RPC的高效執(zhí)行。第3行

8、，Map函數(shù)的第一個參數(shù)，key是文本在文件中的偏移值，在這個程序中沒有使用。根據(jù)Main函數(shù)中的設置，第二個參數(shù)value是整個文件文本對象，因此第5行value.toString() 就把整個文本作為一個大字符串返回。該字符串返回到一個StringTokenizer的構造函數(shù)中。StringTokenizer是一個java自帶的對象，用于分裂字符串。例如，new StringTokenizer(str , “,”)就表示以逗號為分隔符，分裂字符串str。如果沒有第二個參數(shù)，就是默認以空格作為分隔符。所以，第五行，就是把整個文本以空格分開成多個字符串（每個字符串就是一個單詞了，因為單詞以空格

9、分開），保存到變量itr中。在接下來的循環(huán)語句中，就依次取出每個單詞，并設置為Text，然后用context.write()輸出。Map函數(shù)的第三個參數(shù)context是一個hadoop的類型，context.write()方法用于輸出一個鍵值對。context.write(word,one)就輸出了一個單詞和它的計數(shù)。每個單詞的計數(shù)當然為1，這個鍵值對輸出到系統(tǒng)后，系統(tǒng)根據(jù)key進行合并。例如，一個文本中含有3個單詞hello，那么map函數(shù)就是輸出3個鍵值對(“hello”,1)，系統(tǒng)根據(jù)key值進行合并，于是合并為 (“hello” , )，這個鍵值對就輸入給reduce函數(shù)。1 priv

10、ate IntWritable result = new IntWritable();2 public void reduce( Text key, Iterable values, Context context)3throws IOException, InterruptedException 4 int sum = 0;5 for (IntWritable val : values) 6 sum += val.get();7 8 result.set(sum);9 context.write(key, result); 看懂了Map函數(shù)，Reduce函數(shù)應該很容易懂了。Key就是每一個單詞，第57行的循環(huán)用于計算每一個key出現(xiàn)的次數(shù)。由于map函數(shù)中的輸出是context.write(word, one)，因此這里的每個val.get