Mapreduce初析

　　Mapreduce是一個計算框架，既然是做計算的框架，那么表現形式就是有個輸入（input），mapreduce操作這個輸入（input），通過本身定義好的計算模型，得到一個輸出（output），這個輸出就是我們所需要的結果。

　　我們要學習的就是這個計算模型的運行規則。在運行一個mapreduce計算任務時候，任務過程被分為兩個階段：map階段和reduce階段，每個階段都是用鍵值對（key/value）作為輸入（input）和輸出（output）。而程序員要做的就是定義好這兩個階段的函數：map函數和reduce函數。

　　mapreduce編程實例

　　1、數據去重

　　“數據去重”主要是為了掌握和利用并行化思想來對數據進行有意義的篩選。統計大數據集上的數據種類個數、從網站日志中計算訪問地等這些看似龐雜的任務都會涉及數據去重。下面就進入這個實例的MapReduce程序設計。

　　1.1 實例描述

　　對數據文件中的數據進行去重。數據文件中的每行都是一個數據。

　　樣例輸入如下所示：

　　1）file1：

　　?

　　2）file2：

　?

　　樣例輸出如下所示：

?

　　1.2 設計思路

　　數據去重的最終目標是讓原始數據中出現次數超過一次的數據在輸出文件中只出現一次。我們自然而然會想到將同一個數據的所有記錄都交給一臺reduce機器，無論這個數據出現多少次，只要在最終結果中輸出一次就可以了。具體就是reduce的輸入應該以數據作為key，而對value-list則沒有要求。當reduce接收到一個《key，value-list》時就直接將key復制到輸出的key中，并將value設置成空值。

　　在MapReduce流程中，map的輸出《key，value》經過shuffle過程聚集成《key，value- list》后會交給reduce。所以從設計好的reduce輸入可以反推出map的輸出key應為數據，value任意。繼續反推，map輸出數 據的key為數據，而在這個實例中每個數據代表輸入文件中的一行內容，所以map階段要完成的任務就是在采用Hadoop默認的作業輸入方式之后，將 value設置為key，并直接輸出（輸出中的value任意）。map中的結果經過shuffle過程之后交給reduce。reduce階段不會管每 個key有多少個value，它直接將輸入的key復制為輸出的key，并輸出就可以了（輸出中的value被設置成空了）。

　　1.3 程序代碼

　　程序代碼如下所示：

　　package com.hebut.mr;

　　import java.io.IOException;

　　import org.apache.hadoop.conf.Configuration;

　　import org.apache.hadoop.fs.Path;

　　import org.apache.hadoop.io.IntWritable;

　　import org.apache.hadoop.io.Text;

　　import org.apache.hadoop.mapreduce.Job;

　　import org.apache.hadoop.mapreduce.Mapper;

　　import org.apache.hadoop.mapreduce.Reducer;

　　import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

　　import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

　　import org.apache.hadoop.util.GenericOptionsParser;

　　public class Dedup {

　　//map將輸入中的value復制到輸出數據的key上，并直接輸出

　　public static class Map extends Mapper《Object，Text，Text，Text》{

　　private static Text line=new Text（）;//每行數據

　　//實現map函數

　　public void map（Object key，Text value，Context context）

　　throws IOException，InterruptedException{

　　line=value;

　　context.write（line， new Text（“”））;

　　}

　　}

　　//reduce將輸入中的key復制到輸出數據的key上，并直接輸出

　　public static class Reduce extends Reducer《Text，Text，Text，Text》{

　　//實現reduce函數

　　public void reduce（Text key，Iterable《Text》 values，Context context）

　　throws IOException，InterruptedException{

　　context.write（key， new Text（“”））;

　　}

　　}

　　public static void main（String［］ args） throws Exception{

　　Configuration conf = new Configuration（）;

　　//這句話很關鍵

　　conf.set（“mapred.job.tracker”， “192.168.1.2:9001”）;

　　String［］ ioArgs=new String［］{“dedup_in”，“dedup_out”};

　　String［］ otherArgs = new GenericOptionsParser（conf， ioArgs）.getRemainingArgs（）;

　　if （otherArgs.length ！= 2） {

　　System.err.println（“Usage： Data Deduplication 《in》 《out》”）;

　　System.exit（2）;

　　}

　　Job job = new Job（conf， “Data Deduplication”）;

　　job.setJarByClass（Dedup.class）;

　　//設置Map、Combine和Reduce處理類

　　job.setMapperClass（Map.class）;

　　job.setCombinerClass（Reduce.class）;

　　job.setReducerClass（Reduce.class）;

　　//設置輸出類型

　　job.setOutputKeyClass（Text.class）;

　　job.setOutputValueClass（Text.class）;

　　//設置輸入和輸出目錄

　　FileInputFormat.addInputPath（job， new Path（otherArgs［0］））;

　　FileOutputFormat.setOutputPath（job， new Path（otherArgs［1］））;

　　System.exit（job.waitForCompletion（true） ？ 0 ： 1）;

　　}　}

　　1.4 代碼結果

　　1）準備測試數據

　　通過Eclipse下面的“DFS Locations”在“/user/hadoop”目錄下創建輸入文件“dedup_in”文件夾（備注：“dedup_out”不需要創建。）如圖1.4-1所示，已經成功創建。

?

　　然后在本地建立兩個txt文件，通過Eclipse上傳到“/user/hadoop/dedup_in”文件夾中，兩個txt文件的內容如“實例描述”那兩個文件一樣。如圖1.4-2所示，成功上傳之后。

　　從SecureCRT遠處查看“Master.Hadoop”的也能證實我們上傳的兩個文件。

　　?

　　查看兩個文件的內容如圖1.4-3所示：

　　?

　　2）查看運行結果

　　這時我們右擊Eclipse 的“DFS Locations”中“/user/hadoop”文件夾進行刷新，這時會發現多出一個“dedup_out”文件夾，且里面有3個文件，然后打開雙 其“part-r-00000”文件，會在Eclipse中間把內容顯示出來。如圖1.4-4所示。

　?

　　此時，你可以對比一下和我們之前預期的結果是否一致。