hadoop-mapreduce案例集

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

注：戴#號(hào)的是重復(fù)數(shù)據(jù)
file1數(shù)據(jù)：

2012-3-1 a
2012-3-2 b
2012-3-3 c #
2012-3-4 d #
2012-3-5 a #
2012-3-6 b
2012-3-7 c
2012-3-3 c #

file2數(shù)據(jù)：

2012-3-1 b
2012-3-2 a
2012-3-3 b
2012-3-4 d #
2012-3-5 a #
2012-3-6 c
2012-3-7 d
2012-3-3 c #

輸出結(jié)果：

2012-3-1 a
2012-3-1 b
2012-3-2 a
2012-3-2 b
2012-3-3 b
2012-3-3 c
2012-3-4 d
2012-3-5 a
2012-3-6 b
2012-3-6 c
2012-3-7 c
2012-3-7 d

設(shè)計(jì)思路：
數(shù)據(jù)去重的最終目標(biāo)是讓原始數(shù)據(jù)中出現(xiàn)次數(shù)超過一次的數(shù)據(jù)在輸出文件中只出現(xiàn)一次织盼。我們自然而然會(huì)想到將同一個(gè)數(shù)據(jù)的所有記錄都交給一臺(tái)reduce機(jī)器，無論這個(gè)數(shù)據(jù)出現(xiàn)多少次盗似，只要在最終結(jié)果中輸出一次就可以了哩陕。具體就是reduce的輸入應(yīng)該以數(shù)據(jù)作為key，而對(duì)value-list則沒有要求。當(dāng)reduce接收到一個(gè)<key萌踱，value-list>時(shí)就直接將key復(fù)制到輸出的key中葵礼，并將value設(shè)置成空值。
在MapReduce流程中并鸵，map的輸出<key鸳粉，value>經(jīng)過shuffle過程聚集成<key，value-list>后會(huì)交給reduce园担。所以從設(shè)計(jì)好的reduce輸入可以反推出map的輸出key應(yīng)為數(shù)據(jù)届谈，value任意。繼續(xù)反推弯汰，map輸出數(shù)據(jù)的key為數(shù)據(jù)艰山，而在這個(gè)實(shí)例中每個(gè)數(shù)據(jù)代表輸入文件中的一行內(nèi)容，所以map階段要完成的任務(wù)就是在采用Hadoop默認(rèn)的作業(yè)輸入方式之后咏闪，將value設(shè)置為key曙搬，并直接輸出（輸出中的value任意）。map中的結(jié)果經(jīng)過shuffle過程之后交給reduce鸽嫂。reduce階段不會(huì)管每個(gè)key有多少個(gè)value纵装，它直接將輸入的key復(fù)制為輸出的key，并輸出就可以了（輸出中的value被設(shè)置成空了）据某。

代碼：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
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;

public class DataNoHeavy {

    public static class DataMap extends Mapper<LongWritable, Text, Text, Text>{
        
        protected void map(LongWritable key, Text value, Context context) 
                throws java.io.IOException ,InterruptedException {
            Text line = value;
            context.write(line, new Text(""));
        };
    }
    
    public static class DataReduce extends Reducer<Text, Text, Text, Text>{
    
        protected void reduce(Text key, Iterable<Text> value, Context context) 
                throws java.io.IOException ,InterruptedException {
             context.write(key, new Text(""));
        };
    }
    
    public static void main(String[] args) {
        
            Configuration conf = new Configuration();
            //設(shè)置mapper的配置橡娄，既就是hadoop/conf/mapred-site.xml的配置信息
            conf.set("mapred.job.tracker", "hadoop01:9001");        
            try {
                //新建一個(gè)Job工作
                Job job = new Job(conf);
                //設(shè)置運(yùn)行類
                job.setJarByClass(JobRun.class);
                //設(shè)置要執(zhí)行的mapper類（自己書寫的）
                job.setMapperClass(DataMap.class);
                //設(shè)置要執(zhí)行的reduce類（自己書寫的）
                job.setReducerClass(DataReduce.class);
                //設(shè)置輸出key的類型
                job.setMapOutputKeyClass(Text.class);
                //設(shè)置輸出value的類型
                job.setMapOutputValueClass(Text.class);
                
                //設(shè)置ruduce任務(wù)的個(gè)數(shù)，默認(rèn)個(gè)數(shù)為一個(gè)（一般reduce的個(gè)數(shù)越多效率越高）
                job.setNumReduceTasks(1);
                
                //mapreduce 輸入數(shù)據(jù)的文件/目錄
                FileInputFormat.addInputPath(job, new Path("/usr/input/wc/Demo/dataNoHeavy"));
                //mapreduce 執(zhí)行后輸出的數(shù)據(jù)目錄
                FileOutputFormat.setOutputPath(job, new Path("/usr/output/Demo/dataNoHeavy"));
                //執(zhí)行完畢退出
                System.exit(job.waitForCompletion(true) ? 0 : 1);
                
            } catch (Exception e) {
                e.printStackTrace();
            }
    }
}

圖例：

數(shù)據(jù)去重.png

【2】數(shù)據(jù)排序
"數(shù)據(jù)排序"是許多實(shí)際任務(wù)執(zhí)行時(shí)要完成的第一項(xiàng)工作癣籽，比如學(xué)生成績(jī)?cè)u(píng)比挽唉、數(shù)據(jù)建立索引等。這個(gè)實(shí)例和數(shù)據(jù)去重類似筷狼，都是先對(duì)原始數(shù)據(jù)進(jìn)行初步處理瓶籽，為進(jìn)一步的數(shù)據(jù)操作打好基礎(chǔ)。下面進(jìn)入這個(gè)示例埂材。

數(shù)據(jù)文件：

file1:

file2：

file3:

26
54
6

輸出結(jié)果：

1    2
2    6
3    15
4    22
5    26
6    32
7    32
8    54
9    92
10    650
11    654
12    756
13    5956
14    65223

設(shè)計(jì)思路：
這個(gè)實(shí)例僅僅要求對(duì)輸入數(shù)據(jù)進(jìn)行排序塑顺，熟悉MapReduce過程的讀者會(huì)很快想到在MapReduce過程中就有排序，是否可以利用這個(gè)默認(rèn)的排序楞遏，而不需要自己再實(shí)現(xiàn)具體的排序呢茬暇？答案是肯定的首昔。
　　但是在使用之前首先需要了解它的默認(rèn)排序規(guī)則寡喝。它是按照key值進(jìn)行排序的，如果key為封裝int的IntWritable類型勒奇，那么MapReduce按照數(shù)字大小對(duì)key排序预鬓，如果key為封裝為String的Text類型，那么MapReduce按照字典順序對(duì)字符串排序。

了解了這個(gè)細(xì)節(jié)格二，我們就知道應(yīng)該使用封裝int的IntWritable型數(shù)據(jù)結(jié)構(gòu)了劈彪。也就是在map中將讀入的數(shù)據(jù)轉(zhuǎn)化成IntWritable型，然后作為key值輸出（value任意）顶猜。reduce拿到<key沧奴，value-list>之后，將輸入的key作為value輸出长窄，并根據(jù)value-list中元素的個(gè)數(shù)決定輸出的次數(shù)滔吠。輸出的key（即代碼中的linenum）是一個(gè)全局變量，它統(tǒng)計(jì)當(dāng)前key的位次挠日。需要注意的是這個(gè)程序中沒有配置Combiner疮绷，也就是在MapReduce過程中不使用Combiner。這主要是因?yàn)槭褂胢ap和reduce就已經(jīng)能夠完成任務(wù)了嚣潜。

代碼：


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
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;

public class DataSort {
    
    public static class DataSortMap extends Mapper<LongWritable, Text, IntWritable, IntWritable> {
        protected void map(LongWritable key, Text value, Context context) 
                throws java.io.IOException ,InterruptedException {
            String line = value.toString();
            context.write(new IntWritable(Integer.parseInt(line)), new IntWritable(1));
        };
    }
    
    public static class DataSortReduce extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> {
        private static IntWritable linenum = new IntWritable(1);
        protected void reduce(IntWritable key, Iterable<IntWritable> value, Context context) 
                throws java.io.IOException ,InterruptedException {
            for(IntWritable val:value){
                context.write(linenum, key);
                linenum = new IntWritable(linenum.get()+1);
            }
        };
    }

    public static void main(String[] args) {
        Configuration conf = new Configuration();
        //設(shè)置mapper的配置冬骚，既就是hadoop/conf/mapred-site.xml的配置信息
        conf.set("mapred.job.tracker", "hadoop01:9001");        
        try {
            //新建一個(gè)Job工作
            Job job = new Job(conf);
            //設(shè)置運(yùn)行類
            job.setJarByClass(JobRun.class);
            //設(shè)置要執(zhí)行的mapper類（自己書寫的）
            job.setMapperClass(DataSortMap.class);
            //設(shè)置要執(zhí)行的reduce類（自己書寫的）
            job.setReducerClass(DataSortReduce.class);
            //設(shè)置輸出key的類型
            job.setOutputKeyClass(IntWritable.class);
            //設(shè)置輸出value的類型
            job.setOutputValueClass(IntWritable.class);
            
            //設(shè)置ruduce任務(wù)的個(gè)數(shù)，默認(rèn)個(gè)數(shù)為一個(gè)（一般reduce的個(gè)數(shù)越多效率越高）
            //job.setNumReduceTasks(1);
            
            //mapreduce 輸入數(shù)據(jù)的文件/目錄
            FileInputFormat.addInputPath(job, new Path("/usr/input/wc/Demo/dataSort"));
            //mapreduce 執(zhí)行后輸出的數(shù)據(jù)目錄
            FileOutputFormat.setOutputPath(job, new Path("/usr/output/Demo/dataSort"));
            //執(zhí)行完畢退出
            System.exit(job.waitForCompletion(true) ? 0 : 1 );          
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

數(shù)據(jù)排序.png

【3】平均值
"平均成績(jī)"主要目的還是在重溫經(jīng)典"WordCount"例子懂算，可以說是在基礎(chǔ)上的微變化版只冻，該實(shí)例主要就是實(shí)現(xiàn)一個(gè)計(jì)算學(xué)生平均成績(jī)的例子。
　對(duì)輸入文件中數(shù)據(jù)進(jìn)行就算學(xué)生平均成績(jī)犯犁。輸入文件中的每行內(nèi)容均為一個(gè)學(xué)生的姓名和他相應(yīng)的成績(jī)属愤，如果有多門學(xué)科，則每門學(xué)科為一個(gè)文件酸役。要求在輸出中每行有兩個(gè)間隔的數(shù)據(jù)住诸，其中，第一個(gè)代表學(xué)生的姓名涣澡，第二個(gè)代表其平均成績(jī)贱呐。

數(shù)據(jù)：

math：

張三    88
李四    99
王五    66
趙六    77

china:

張三    78
李四    89
王五    96
趙六    67

english：

張三    80
李四    82
王五    84
趙六    86

輸出結(jié)果：

張三    82
李四    90
王五    82
趙六    76

設(shè)計(jì)思路：
計(jì)算學(xué)生平均成績(jī)是一個(gè)仿"WordCount"例子，用來重溫一下開發(fā)MapReduce程序的流程入桂。程序包括兩部分的內(nèi)容：Map部分和Reduce部分奄薇，分別實(shí)現(xiàn)了map和reduce的功能。
Map處理的是一個(gè)純文本文件抗愁，文件中存放的數(shù)據(jù)時(shí)每一行表示一個(gè)學(xué)生的姓名和他相應(yīng)一科成績(jī)馁蒂。Mapper處理的數(shù)據(jù)是由InputFormat分解過的數(shù)據(jù)集，其中InputFormat的作用是將數(shù)據(jù)集切割成小數(shù)據(jù)集InputSplit蜘腌，每一個(gè)InputSlit將由一個(gè)Mapper負(fù)責(zé)處理沫屡。此外，InputFormat中還提供了一個(gè)RecordReader的實(shí)現(xiàn)撮珠，并將一個(gè)InputSplit解析成<key,value>對(duì)提供給了map函數(shù)沮脖。InputFormat的默認(rèn)值是TextInputFormat，它針對(duì)文本文件，按行將文本切割成InputSlit勺届，并用LineRecordReader將InputSplit解析成<key,value>對(duì)驶俊，key是行在文本中的位置，value是文件中的一行免姿。
Map的結(jié)果會(huì)通過partion分發(fā)到Reducer饼酿，Reducer做完Reduce操作后，將通過以格式OutputFormat輸出胚膊。
Mapper最終處理的結(jié)果對(duì)<key,value>嗜湃，會(huì)送到Reducer中進(jìn)行合并，合并的時(shí)候澜掩，有相同key的鍵/值對(duì)則送到同一個(gè)Reducer上购披。Reducer是所有用戶定制Reducer類地基礎(chǔ)，它的輸入是key和這個(gè)key對(duì)應(yīng)的所有value的一個(gè)迭代器肩榕，同時(shí)還有Reducer的上下文刚陡。Reduce的結(jié)果由Reducer.Context的write方法輸出到文件中。

代碼：


import java.io.IOException;
import java.util.Iterator;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
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;

public class AvgSource {

    public static class AvgSourceMap extends Mapper<LongWritable, Text, Text, IntWritable> {
        
        protected void map(LongWritable key, Text value, Context context) 
                throws java.io.IOException ,InterruptedException {
            String line = value.toString();
            StringTokenizer tok = new StringTokenizer(line, "\n");
            while(tok.hasMoreElements()){
                String name = tok.nextToken();
                String source = tok.nextToken();
                context.write(new Text(name), new IntWritable(Integer.parseInt(source)));
            }
        };
    }
    
    public static class AvgSourceReduce extends Reducer<Text, IntWritable, Text, IntWritable> {
        
        protected void reduce(Text key, Iterable<IntWritable> value, Context context) 
                throws java.io.IOException ,InterruptedException {
            int sum = 0 ;
            int count = 0;
            Iterator<IntWritable> iterator = value.iterator();
            while(iterator.hasNext()){
                sum+= iterator.next().get();
                count++;
            }
            
            int avg = (int)sum/count;
            context.write(key, new IntWritable(avg));
            
        };
    }
    
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        //設(shè)置mapper的配置株汉，既就是hadoop/conf/mapred-site.xml的配置信息
        conf.set("mapred.job.tracker", "hadoop01:9001");
        
        try {
            //新建一個(gè)Job工作
            Job job = new Job(conf);
            //設(shè)置運(yùn)行類
            job.setJarByClass(JobRun.class);
            //設(shè)置要執(zhí)行的mapper類（自己書寫的）
            job.setMapperClass(AvgSourceMapper.class);
            //設(shè)置要執(zhí)行的reduce類（自己書寫的）
            job.setReducerClass(AvgSourceReduce.class);
            //設(shè)置輸出key的類型
            job.setMapOutputKeyClass(Text.class);
            //設(shè)置輸出value的類型
            job.setMapOutputValueClass(IntWritable.class);
            
            //設(shè)置ruduce任務(wù)的個(gè)數(shù)筐乳，默認(rèn)個(gè)數(shù)為一個(gè)（一般reduce的個(gè)數(shù)越多效率越高）
            job.setNumReduceTasks(1);
            
            //mapreduce 輸入數(shù)據(jù)的文件/目錄
            FileInputFormat.addInputPath(job, new Path("/usr/input/wc/Demo/avgSource"));
            //mapreduce 執(zhí)行后輸出的數(shù)據(jù)目錄
            FileOutputFormat.setOutputPath(job, new Path("/usr/output/Demo/avgSource"));
            //執(zhí)行完畢退出
            System.exit(job.waitForCompletion(true) ? 0 : 1);
                
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

平均數(shù).png

【4】單表鏈接
前面的實(shí)例都是在數(shù)據(jù)上進(jìn)行一些簡(jiǎn)單的處理，為進(jìn)一步的操作打基礎(chǔ)乔妈。"單表關(guān)聯(lián)"這個(gè)實(shí)例要求從給出的數(shù)據(jù)中尋找所關(guān)心的數(shù)據(jù)蝙云，它是對(duì)原始數(shù)據(jù)所包含信息的挖掘。下面進(jìn)入這個(gè)實(shí)例路召。

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

數(shù)據(jù)：
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

家譜關(guān)系：

家譜關(guān)系.png

輸出結(jié)果：

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

代碼：


import java.io.IOException;
import java.util.Iterator;
import java.util.StringTokenizer;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
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;

public class SingleRelation {
    
    public static int time = 0;
    /*
     * map將輸出分割child和parent股淡，然后正序輸出一次作為右表身隐，
     * 反序輸出一次作為左表，需要注意的是在輸出的value中必須
     * 加上左右表的區(qū)別標(biāo)識(shí)唯灵。
     */
    public static class SingleRelationMap extends Mapper<LongWritable, Text, Text, Text> {
        protected void map(LongWritable key, Text value, Context context) 
                throws IOException ,InterruptedException {
            String child = new String();//孩子名字
            String parent = new String();//父母名字
            String relation = new String();//左右表標(biāo)示
            
            // 輸入的一行預(yù)處理文本
            StringTokenizer st = new StringTokenizer(value.toString());
            String[] values = new String[2];
            int i = 0;
            while(st.hasMoreTokens()){
                values[i] = st.nextToken();
                i++;
            }
            
            if(values[0].compareTo("child") != 0){
                child = values[0];
                parent = values[1];
                
                //輸出左表
                relation = "1";
                context.write(new Text(parent), new Text(relation+"+"+child+"+"+parent));
                //輸出右表
                relation = "2";
                context.write(new Text(child), new Text(relation+"+"+child+"+"+parent));
            }
        };
    }
    
    public static class SingleRelationReduce extends Reducer<Text, Text, Text, Text> {
        protected void reduce(Text key, Iterable<Text> value, 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 = value.iterator();
            while (ite.hasNext()) {
                String record = ite.next().toString();
                int len = record.length();
                int i = 2;
                if(0 == len){
                    continue;
                }
                
                // 取得左右表標(biāo)識(shí)
                char relation = record.charAt(0);
                // 定義孩子和父母變量
                String child = new String();
                String parent = new String();
                // 獲取value-list中value的child
                while(record.charAt(i) != '+'){
                    child += record.charAt(i);
                    i++;
                }
                
                i = i + 1;
                
                while(i < len){
                    parent += record.charAt(i);
                    i++;
                }
                
                // 左表贾铝，取出child放入grandchildren
                 if ('1' == relation) {
                     grandchild[grandchildnum] = child;
                     grandchildnum++;
                    }
                // 右表，取出parent放入grandparent
                 if ('2' == relation) {
                     grandparent[grandparentnum] = parent;
                     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]), new Text(grandparent[n]));
                    }
                }
            }
            
        };
    }

    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.set("mapred.job.tracker", "hadoop01:9001");
        
        try {
            Job job = new Job(conf);
            job.setJarByClass(JobRun.class);
            // 設(shè)置Map和Reduce處理類
            job.setMapperClass(SingleRelationMap.class);
            job.setReducerClass(SingleRelationReduce.class);
            
            // 設(shè)置輸出類型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(Text.class);
            
         // 設(shè)置輸入和輸出目錄

            FileInputFormat.addInputPath(job, new Path("/usr/input/wc/Demo/single"));

            FileOutputFormat.setOutputPath(job, new Path("/usr/output/Demo/single"));

            System.exit(job.waitForCompletion(true) ? 0 : 1);
            
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

運(yùn)行過程解析

map處理

child        parent                àà                    忽略此行
Tom        Lucy                   àà                <Lucy埠帕，1+Tom+Lucy>
                                            　　　　　　　　<Tom垢揩，2+Tom+Lucy >
Tom        Jack                    àà                <Jack，1+Tom+Jack>
                                            　　　　　　　　<Tom敛瓷，2+Tom+Jack>
Jone        Lucy                　àà                <Lucy叁巨，1+Jone+Lucy>
                                            　　　　　　　　<Jone，2+Jone+Lucy>
Jone        Jack                    àà                <Jack琐驴，1+Jone+Jack>
                                            　　　　　　　　<Jone俘种，2+Jone+Jack>
Lucy        Mary                   àà                <Mary，1+Lucy+Mary>
                                            　　　　　　　　<Lucy绝淡，2+Lucy+Mary>
Lucy        Ben                    àà                <Ben宙刘，1+Lucy+Ben>
                                            　　　　　　　　　<Lucy，2+Lucy+Ben>
Jack        Alice                    àà                <Alice牢酵，1+Jack+Alice>
                                            　　　　　　　　　　<Jack悬包，2+Jack+Alice>
Jack        Jesse                   àà                <Jesse，1+Jack+Jesse>
                                            　　　　　　　　　　<Jack馍乙，2+Jack+Jesse>
Terry        Alice                   àà                <Alice布近，1+Terry+Alice>
                                            　　　　　　　　　　<Terry，2+Terry+Alice>
Terry        Jesse                  àà                <Jesse丝格，1+Terry+Jesse>
                                            　　　　　　　　　　<Terry撑瞧，2+Terry+Jesse>
Philip        Terry                  àà                <Terry，1+Philip+Terry>
                                            　　　　　　　　　　<Philip显蝌，2+Philip+Terry>
Philip        Alma                   àà                <Alma预伺，1+Philip+Alma>
                                            　　　　　　　　　　<Philip，2+Philip+Alma>
Mark        Terry                   àà                <Terry曼尊，1+Mark+Terry>
                                            　　　　　　　　　　<Mark酬诀，2+Mark+Terry>
Mark        Alma                　àà                <Alma，1+Mark+Alma>
                                            　　　　　　　　　　<Mark骆撇，2+Mark+Alma>

Shuffle處理

shuffle.png

reduce處理
首先由語句"0 != grandchildnum && 0 != grandparentnum"得知瞒御，只要在"value-list"中沒有左表或者右表，則不會(huì)做處理神郊，可以根據(jù)這條規(guī)則去除無效的shuffle連接*肴裙。

2016-11-20_213049.png

然后根據(jù)下面語句進(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ù)如下所示：

2016-11-20_213624.png

然后根據(jù)下面語句進(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]));
    }
}

2016-11-20_214216.png

最后編輯于：2017.12.04 14:22:34

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者

人面猴
序言：七十年代末谊惭，一起剝皮案震驚了整個(gè)濱河市，隨后出現(xiàn)的幾起案子驶忌，更是在濱河造成了極大的恐慌擂仍，老刑警劉巖囤屹，帶你破解...
沈念sama閱讀 216,496評(píng)論 6贊 501
死咒
序言：濱河連續(xù)發(fā)生了三起死亡事件，死亡現(xiàn)場(chǎng)離奇詭異逢渔，居然都是意外死亡肋坚，警方通過查閱死者的電腦和手機(jī)，發(fā)現(xiàn)死者居然都...
沈念sama閱讀 92,407評(píng)論 3贊 392
救了他兩次的神仙讓他今天三更去死
文/潘曉璐我一進(jìn)店門肃廓，熙熙樓的掌柜王于貴愁眉苦臉地迎上來智厌，“玉大人，你說我怎么就攤上這事盲赊∠撑簦” “怎么了？”我有些...
開封第一講書人閱讀 162,632評(píng)論 0贊 353
道士緝兇錄：失蹤的賣姜人
文/不壞的土叔我叫張陵哀蘑，是天一觀的道長(zhǎng)吝沫。經(jīng)常有香客問我，道長(zhǎng)递礼，這世上最難降的妖魔是什么惨险？我笑而不...
開封第一講書人閱讀 58,180評(píng)論 1贊 292
?港島之戀（遺憾婚禮）
正文為了忘掉前任，我火速辦了婚禮脊髓，結(jié)果婚禮上辫愉，老公的妹妹穿的比我還像新娘。我一直安慰自己将硝，他們只是感情好恭朗，可當(dāng)我...
茶點(diǎn)故事閱讀 67,198評(píng)論 6贊 388
惡毒庶女頂嫁案：這布局不是一般人想出來的
文/花漫我一把揭開白布。她就那樣靜靜地躺著依疼，像睡著了一般痰腮。火紅的嫁衣襯著肌膚如雪。梳的紋絲不亂的頭發(fā)上律罢，一...
開封第一講書人閱讀 51,165評(píng)論 1贊 299
城市分裂傳說
那天膀值，我揣著相機(jī)與錄音，去河邊找鬼误辑。笑死沧踏，一個(gè)胖子當(dāng)著我的面吹牛，可吹牛的內(nèi)容都是我干的巾钉。我是一名探鬼主播翘狱，決...
沈念sama閱讀 40,052評(píng)論 3贊 418
雙鴛鴦連環(huán)套：你想象不到人心有多黑
文/蒼蘭香墨我猛地睜開眼，長(zhǎng)吁一口氣：“原來是場(chǎng)噩夢(mèng)啊……” “哼砰苍！你這毒婦竟也來了潦匈？” 一聲冷哼從身側(cè)響起阱高，我...
開封第一講書人閱讀 38,910評(píng)論 0贊 274
萬榮殺人案實(shí)錄
序言：老撾萬榮一對(duì)情侶失蹤，失蹤者是張志新（化名）和其女友劉穎茬缩，沒想到半個(gè)月后赤惊，有當(dāng)?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體，經(jīng)...
沈念sama閱讀 45,324評(píng)論 1贊 310
?護(hù)林員之死
正文獨(dú)居荒郊野嶺守林人離奇死亡寒屯，尸身上長(zhǎng)有42處帶血的膿包…… 初始之章·張勛以下內(nèi)容為張勛視角年9月15日...
茶點(diǎn)故事閱讀 37,542評(píng)論 2贊 332
?白月光啟示錄
正文我和宋清朗相戀三年，在試婚紗的時(shí)候發(fā)現(xiàn)自己被綠了黍少。大學(xué)時(shí)的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片寡夹。...
茶點(diǎn)故事閱讀 39,711評(píng)論 1贊 348
活死人
序言：一個(gè)原本活蹦亂跳的男人離奇死亡，死狀恐怖厂置，靈堂內(nèi)的尸體忽然破棺而出菩掏，到底是詐尸還是另有隱情，我是刑警寧澤昵济，帶...
沈念sama閱讀 35,424評(píng)論 5贊 343
?日本核電站爆炸內(nèi)幕
正文年R本政府宣布智绸，位于F島的核電站，受9級(jí)特大地震影響访忿，放射性物質(zhì)發(fā)生泄漏瞧栗。R本人自食惡果不足惜，卻給世界環(huán)境...
茶點(diǎn)故事閱讀 41,017評(píng)論 3贊 326
男人毒藥：我在死后第九天來索命
文/蒙蒙一海铆、第九天我趴在偏房一處隱蔽的房頂上張望迹恐。院中可真熱鬧，春花似錦卧斟、人聲如沸殴边。這莊子的主人今日做“春日...
開封第一講書人閱讀 31,668評(píng)論 0贊 22
一樁弒父案珍语，背后竟有這般陰謀
文/蒼蘭香墨我抬頭看了看天上的太陽锤岸。三九已至，卻和暖如春板乙，著一層夾襖步出監(jiān)牢的瞬間是偷，已是汗流浹背。一陣腳步聲響...
開封第一講書人閱讀 32,823評(píng)論 1贊 269
情欲美人皮
我被黑心中介騙來泰國(guó)打工募逞，沒想到剛下飛機(jī)就差點(diǎn)兒被人妖公主榨干…… 1. 我叫王不留晓猛，地道東北人。一個(gè)月前我還...
沈念sama閱讀 47,722評(píng)論 2贊 368
代替公主和親
正文我出身青樓凡辱，卻偏偏與公主長(zhǎng)得像戒职，于是被迫代替她去往敵國(guó)和親。傳聞我的和親對(duì)象是個(gè)殘疾皇子透乾，可洞房花燭夜當(dāng)晚...
茶點(diǎn)故事閱讀 44,611評(píng)論 2贊 353

hadoop-mapreduce案例集

推薦閱讀更多精彩內(nèi)容