- mapreduce編程實(shí)例

　　前面的實(shí)例都是在數(shù)據(jù)上進(jìn)行一些簡(jiǎn)單的處理，為進(jìn)一步的操作打基礎(chǔ)?！皢伪黻P(guān)聯(lián)”這個(gè)實(shí)例要求從給出的數(shù)據(jù)中尋找所關(guān)心的數(shù)據(jù)，它是對(duì)原始數(shù)據(jù)所包含信息的挖掘。下面進(jìn)入這個(gè)實(shí)例。

　　4.1 實(shí)例描述

　　實(shí)例中給出child-parent（孩子——父母）表，要求輸出grandchild-grandparent（孫子——爺奶）表。

　　樣例輸入如下所示。

　　file：

　　child parent

　　Tom Lucy

　　Tom Jack

　　Jone Lucy

　　Jone Jack

　　Lucy Mary

　　Lucy Ben

　　Jack Alice

　　Jack Jesse

　　Terry Alice

　　Terry Jesse

　　Philip Terry

　　Philip Alma

　　Mark Terry

　　Mark Alma

　　家族樹(shù)狀關(guān)系譜：

　　樣例輸出如下所示。

　　file：

　　grandchild grandparent

　　Tom 　　Alice

　　Tom 　　Jesse

　　Jone 　　Alice

　　Jone 　　 Jesse

　　Tom 　　Mary

　　Tom 　　Ben

　　Jone 　　 Mary

　　Jone 　　 Ben

　　Philip 　　 Alice

　　Philip 　　Jesse

　　Mark 　　 Alice

　　Mark 　　 Jesse

　　4.2 設(shè)計(jì)思路

　　分析這個(gè)實(shí)例，顯然需要進(jìn)行單表連接，連接的是左表的parent列和右表的child列，且左表和右表是同一個(gè)表。

　　連接結(jié)果中除去連接的兩列就是所需要的結(jié)果——“grandchild--grandparent”表。要用MapReduce解決這個(gè)實(shí)例，首先應(yīng)該考慮如何實(shí)現(xiàn)表的自連接；其次就是連接列的設(shè)置；最后是結(jié)果的整理。

　　考慮到MapReduce的shuffle過(guò)程會(huì)將相同的key會(huì)連接在一起，所以可以將map結(jié)果的key設(shè)置成待連接的列，然后列中相同的值就自然會(huì)連接在一起了。再與最開(kāi)始的分析聯(lián)系起來(lái)：

　　要連接的是左表的parent列和右表的child列，且左表和右表是同一個(gè)表，所以在map階段將讀入數(shù)據(jù)分割成child和parent之后，會(huì)將parent設(shè)置成key，child設(shè)置成value進(jìn)行輸出，并作為左表；再將同一對(duì)child和parent中的child設(shè)置成key，parent設(shè)置成value進(jìn)行輸出，作為右表。為了區(qū)分輸出中的左右表，需要在輸出的value中再加上左右表的信息，比如在value的String最開(kāi)始處加上字符1表示左表，加上字符2表示右表。這樣在map的結(jié)果中就形成了左表和右表，然后在shuffle過(guò)程中完成連接。reduce接收到連接的結(jié)果，其中每個(gè)key的value-list就包含了“grandchild--grandparent”關(guān)系。取出每個(gè)key的value-list進(jìn)行解析，將左表中的child放入一個(gè)數(shù)組，右表中的parent放入一個(gè)數(shù)組，然后對(duì)兩個(gè)數(shù)組求笛卡爾積就是最后的結(jié)果了。

　　4.3 程序代碼

　　程序代碼如下所示。

　　package com.hebut.mr;

　　import java.io.IOException;

　　import java.util.*;

　　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 STjoin {

　　public static int time = 0;

　　/*

　　* map將輸出分割child和parent，然后正序輸出一次作為右表，

　　* 反序輸出一次作為左表，需要注意的是在輸出的value中必須

　　* 加上左右表的區(qū)別標(biāo)識(shí)。

　　*/

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

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

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

　　throws IOException， InterruptedException {

　　String childname = new String（）;// 孩子名稱

　　String parentname = new String（）;// 父母名稱

　　String relationtype = new String（）;// 左右表標(biāo)識(shí)

　　// 輸入的一行預(yù)處理文本

　　StringTokenizer itr=new StringTokenizer（value.toString（））;

　　String［］ values=new String［2］;

　　int i=0;

　　while（itr.hasMoreTokens（））{

　　values［i］=itr.nextToken（）;

　　i++;

　　}

　　if （values［0］.compareTo（“child”） ！= 0） {

　　childname = values［0］;

　　parentname = values［1］;

　　// 輸出左表

　　relationtype = “1”;

　　context.write（new Text（values［1］）， new Text（relationtype +

　　“+”+ childname + “+” + parentname））;

　　// 輸出右表

　　relationtype = “2”;

　　context.write（new Text（values［0］）， new Text（relationtype +

　　“+”+ childname + “+” + parentname））;

　　}

　　}

　　}

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

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

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

　　throws IOException， InterruptedException {

　　// 輸出表頭

　　if （0 == time） {

　　context.write（new Text（“grandchild”）， new Text（“grandparent”））;

　　time++;

　　}

　　int grandchildnum = 0;

　　String［］ grandchild = new String［10］;

　　int grandparentnum = 0;

　　String［］ grandparent = new String［10］;

　　Iterator ite = values.iterator（）;

　　while （ite.hasNext（）） {

　　String record = ite.next（）.toString（）;

　　int len = record.length（）;

　　int i = 2;

　　if （0 == len） {

　　continue;

　　}

　　// 取得左右表標(biāo)識(shí)

　　char relationtype = record.charAt（0）;

　　// 定義孩子和父母變量

　　String childname = new String（）;

　　String parentname = new String（）;

　　// 獲取value-list中value的child

　　while （record.charAt（i） ！= ‘+’） {

　　childname += record.charAt（i）;

　　i++;

　　}

　　i = i + 1;

　　// 獲取value-list中value的parent

　　while （i 《 len） {

　　parentname += record.charAt（i）;

　　i++;

　　}

　　// 左表，取出child放入grandchildren

　　if （‘1’ == relationtype） {

　　grandchild［grandchildnum］ = childname;

　　grandchildnum++;

　　}

　　// 右表，取出parent放入grandparent

　　if （‘2’ == relationtype） {

　　grandparent［grandparentnum］ = parentname;

　　grandparentnum++;

　　}

　　}

　　// grandchild和grandparent數(shù)組求笛卡爾兒積

　　if （0 ！= grandchildnum && 0 ！= grandparentnum） {

　　for （int m = 0; m 《 grandchildnum; m++） {

　　for （int n = 0; n 《 grandparentnum; n++） {

　　// 輸出結(jié)果

　　context.write（new Text（grandchild［m］）， newText（grandparent［n］））;

　　}

　　}

　　}

　　}

　　}

　　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［］ { “STjoin_in”， “STjoin_out” };

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

　　if （otherArgs.length ！= 2） {

　　System.err.println（“Usage： Single Table Join 《in》 《out》”）;

　　System.exit（2）;

　　}

　　Job job = new Job（conf， “Single Table Join”）;

　　job.setJarByClass（STjoin.class）;

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

　　job.setMapperClass（Map.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）;

　　}

　　}

　　4.4 代碼結(jié)果

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

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

　　?

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

　　從SecureCRT遠(yuǎn)處查看“Master.Hadoop”的也能證實(shí)我們上傳的文件，顯示其內(nèi)容如圖4.4-3所示：

　　?

　　2）運(yùn)行詳解

　?。?）Map處理：

　　（2）Shuffle處理

　　在shuffle過(guò)程中完成連接。

　　（3）Reduce處理

　　首先由語(yǔ)句“0 ！= grandchildnum && 0 ！= grandparentnum”得知，只要在“value-list”中沒(méi)有左表或者右表，則不會(huì)做處理，可以根據(jù)這條規(guī)則去除無(wú)效的shuffle連接。

　　然后根據(jù)下面語(yǔ)句進(jìn)一步對(duì)有效的shuffle連接做處理。

　　// 左表，取出child放入grandchildren

　　if （‘1’ == relationtype） {

　　grandchild［grandchildnum］ = childname;

　　grandchildnum++;

　　}

　　// 右表，取出parent放入grandparent

　　if （‘2’ == relationtype） {

　　grandparent［grandparentnum］ = parentname;

　　grandparentnum++;

　　}

　　針對(duì)一條數(shù)據(jù)進(jìn)行分析：

　　《Jack，1+Tom+Jack，

　　1+Jone+Jack，

　　2+Jack+Alice，

　　2+Jack+Jesse 》

　　分析結(jié)果：左表用“字符1”表示，右表用“字符2”表示，上面的《key，value-list》中的“key”表示左表與右表的連接鍵。而“value-list”表示以“key”連接的左表與右表的相關(guān)數(shù)據(jù)。

　　根據(jù)上面針對(duì)左表與右表不同的處理規(guī)則，取得兩個(gè)數(shù)組的數(shù)據(jù)。

　　然后根據(jù)下面語(yǔ)句進(jìn)行處理。

　　for （int m = 0; m 《 grandchildnum; m++） {

　　for （int n = 0; n 《 grandparentnum; n++） {

　　context.write（new Text（grandchild［m］）， new Text（grandparent［n］））;

　　}

　　}

　　處理結(jié)果如下面所示：

　　Tom Jesse

　　Tom Alice

　　Jone Jesse

　　Jone Alice

　　其他的有效shuffle連接處理都是如此。

　　3）查看運(yùn)行結(jié)果

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

　　?