mapreduce編程實例

　　Mapreduce初析

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

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

　　mapreduce編程實例

　　1、數(shù)據(jù)去重

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

　　1.1 實例描述

　　對數(shù)據(jù)文件中的數(shù)據(jù)進(jìn)行去重。數(shù)據(jù)文件中的每行都是一個數(shù)據(jù)。

　　樣例輸入如下所示：

　　1）file1：

　　?

　　2）file2：

　?

　　樣例輸出如下所示：

?

　　1.2 設(shè)計思路

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

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

　　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復(fù)制到輸出數(shù)據(jù)的key上，并直接輸出

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

　　private static Text line=new Text（）;//每行數(shù)據(jù)

　　//實現(xiàn)map函數(shù)

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

　　throws IOException，InterruptedException{

　　line=value;

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

　　}

　　}

　　//reduce將輸入中的key復(fù)制到輸出數(shù)據(jù)的key上，并直接輸出

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

　　//實現(xiàn)reduce函數(shù)

　　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（）;

　　//這句話很關(guān)鍵

　　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）;

　　//設(shè)置Map、Combine和Reduce處理類

　　job.setMapperClass（Map.class）;

　　job.setCombinerClass（Reduce.class）;

　　job.setReducerClass（Reduce.class）;

　　//設(shè)置輸出類型

　　job.setOutputKeyClass（Text.class）;

　　job.setOutputValueClass（Text.class）;

　　//設(shè)置輸入和輸出目錄

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

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

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

　　}　}

　　1.4 代碼結(jié)果

　　1）準(zhǔn)備測試數(shù)據(jù)

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

?

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

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

　　?

　　查看兩個文件的內(nèi)容如圖1.4-3所示：

　　?

　　2）查看運行結(jié)果

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

　?

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