[关闭]
@Arslan6and6 2016-08-29T02:12:24.000000Z 字数 13676 阅读 618

【作业十】殷杰

第五章、高级Hadoop2.x

---MapReduce高级应用练习 【二次排序案例】和【MapReduce Join案例】

作业描述：

依据上课讲解【二次排序案例】和【MapReduce Join案例】进行编码实现和要点分析，以下几点必须在作业中体现：

1） 理解【二次排序】功能，使用自己理解的方式表达（包括自定义数据类型、分区、分组、排序）。

2） 编码实现二次排序功能，提供源代码文件

3） 理解 MapReduce Join 的几种方式，编码实现 Reduce Join，提供源代码，说出思路。

1） 理解【二次排序】功能，使用自己理解的方式表达（包括自定义数据类型、分区、分组、排序）。

1.二次排序概念：
首先按照第一字段排序，然后再对第一字段相同的行按照第二字段排序，注意不能破坏第一次排序的结果 。
例如：将下列内容的文件用二次排序程序执行
01.jpg-5.6kB
要得到下列结果
02.jpg-5.9kB

2、思路
将键设为综合原始键和值的复合键。
键comparator应该以复合键排序，即原本的键和值组合。
partitioner和复合键comparator的分组只考虑本来键的分区和分组。

3、实现分析

(1)自定义复合键key,由第一个字段即原本的key和第二个字段value组成，数据类型设定为PairWritable，即第一个字段Text和第二个字段IntWritable数据类型组合。并设定一个PairWritable继承WritableComparable接口，以实现序列化和反序列化。

(2)为程序调优设置map端shuffle分区。并创建FirstPartitioner分区类。
05.jpg-15.1kB
该过程以第一个字段key进行分区，以key为依据自然就是比较key。为整个MR过程中第一次比较。本例中，设置分区类后a,1 a,100 a,3被分到一个reducetast,以此类推。如不设置分区类则按行分区。
04.jpg-53.1kB

(3)为程序调优设置reduce端shuffle分组。并创建FirstPartitioner分组类。如不设置分组则按默认复合键个数分组。
06.jpg-12kB
只要第一个字段key相同，而不是复合健相同。就把对于的value放进同一个组。
07.jpg-68.8kB
03.jpg-68kB

(4)在PairWritable类中以第一个字段为key为value排序，而不是以复合键。
08.jpg-32.3kB

2） 编码实现二次排序功能，提供源代码文件

SecondarySortMapReduce.java

  1. package org.apache.hadoop.sort;
  3. import java.io.IOException;
  5. import org.apache.hadoop.conf.Configuration;
  6. import org.apache.hadoop.conf.Configured;
  7. import org.apache.hadoop.fs.Path;
  8. import org.apache.hadoop.io.IntWritable;
  9. import org.apache.hadoop.io.LongWritable;
  10. import org.apache.hadoop.io.Text;
  11. import org.apache.hadoop.mapreduce.Job;
  12. import org.apache.hadoop.mapreduce.Mapper;
  13. import org.apache.hadoop.mapreduce.Mapper.Context;
  14. import org.apache.hadoop.mapreduce.Reducer;
  15. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  16. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  17. import org.apache.hadoop.util.Tool;
  18. import org.apache.hadoop.util.ToolRunner;
  20. public class SecondarySortMapReduce extends Configured implements Tool {
  21.     // step 1 : mapper
  22.     /**
  23.      * public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
  24.      */
  25.     public static class WordCountMapper extends //
  26.             Mapper<LongWritable, Text, PairWritable, IntWritable> {
  28.         private PairWritable mapOutputKey = new PairWritable();
  29.         private IntWritable mapOutputValue = new IntWritable();
  31.         @Override
  32.         public void map(LongWritable key, Text value, Context context)
  33.                 throws IOException, InterruptedException {
  35.             // line -- split
  36.             String[] strs = value.toString().split(",");
  38.             // map output key
  39.             mapOutputKey.set(strs[0], Integer.valueOf(strs[1]));
  41.             // map output value
  42.             mapOutputValue.set(Integer.valueOf(strs[1]));
  44.             context.write(mapOutputKey, mapOutputValue);
  45.         }
  47.     }
  49.     // step 2 : reducer
  50.     /**
  51.      * 
  52.      * a#1 1 a#3 3 a#100 100
  53.      * 
  54.      * @author beifeng
  55.      * 
  56.      */
  57.     public static class WordCountReducer extends //
  58.             Reducer<PairWritable, IntWritable, Text, IntWritable> {
  60.         private Text outputKey = new Text();
  62.         @Override
  63.         public void reduce(PairWritable key, Iterable<IntWritable> values,
  64.                 Context context) throws IOException, InterruptedException {
  66.             // set output key
  67.             outputKey.set(key.getFirst());
  69.             for (IntWritable value : values) {
  70.                 context.write(outputKey, value);
  71.             }
  72.         }
  73.     }
  75.     // step 3 : job
  77.     public int run(String[] args) throws Exception {
  79.         // 1 : get configuration
  80.         Configuration configuration = super.getConf();
  82.         // 2 : create job
  83.         Job job = Job.getInstance(//
  84.                 configuration,//
  85.                 this.getClass().getSimpleName());
  86.         job.setJarByClass(SecondarySortMapReduce.class);
  88.         // job.setNumReduceTasks(tasks);
  90.         // 3 : set job
  91.         // input --> map --> reduce --> output
  92.         // 3.1 : input
  93.         Path inPath = new Path(args[0]);
  94.         FileInputFormat.addInputPath(job, inPath);
  96.         // 3.2 : mapper
  97.         job.setMapperClass(WordCountMapper.class);
  98.         // TODO
  99.         job.setMapOutputKeyClass(PairWritable.class);
  100.         job.setMapOutputValueClass(IntWritable.class);
  102.         // ====================shuffle==========================
  103.         // 1: partition
  104.         job.setPartitionerClass(FirstPartitioner.class);
  105.         // 2: sort
  106.         // job.setSortComparatorClass(cls);
  107.         // 3: combine
  108.         // job.setCombinerClass(cls);
  109.         // 4: compress
  110.         // set by configuration
  111.         // 5 : group
  112.         job.setGroupingComparatorClass(FirstGroupingComparator.class);
  114.         // ====================shuffle==========================
  116.         // 3.3 : reducer
  117.         job.setReducerClass(WordCountReducer.class);
  118.         // TODO
  119.         job.setOutputKeyClass(Text.class);
  120.         job.setOutputValueClass(IntWritable.class);
  122.         // 3.4 : output
  123.         Path outPath = new Path(args[1]);
  124.         FileOutputFormat.setOutputPath(job, outPath);
  126.         // 4 : submit job
  127.         boolean isSuccess = job.waitForCompletion(true);
  128.         return isSuccess ? 0 : 1;
  130.     }
  132.     public static void main(String[] args) throws Exception {
  134.         //args = new String[] {
  135.         //      "hdfs://hadoop-senior.ibeifeng.com:8020/input/sort.txt",
  136.         //      "hdfs://hadoop-senior.ibeifeng.com:8020/output2/" 
  137.         //      };
  138.         // get configuration
  139.         Configuration configuration = new Configuration();
  141.         // configuration.set(name, value);
  143.         // run job
  144.         int status = ToolRunner.run(//
  145.                 configuration,//
  146.                 new SecondarySortMapReduce(),//
  147.                 args);
  149.         // exit program
  150.         System.exit(status);
  151.     }
  152. }

PairWritable.java

  1. package org.apache.hadoop.sort;
  3. import java.io.DataInput;
  4. import java.io.DataOutput;
  5. import java.io.IOException;
  7. import org.apache.hadoop.io.WritableComparable;
  9. public class PairWritable implements WritableComparable<PairWritable> {
  11.     private String first ;
  12.     private int second ;
  16.     public PairWritable() {
  17.     }
  19.     public PairWritable(String first, int second) {
  20.         this.set(first, second); 
  21.     }
  22.     public void set(String first, int second) {
  23.         this.setFirst(first);
  24.         this.setSecond(second);
  25.     }
  26.     public String getFirst() {
  27.         return first;
  28.     }
  30.     public void setFirst(String first) {
  31.         this.first = first;
  32.     }
  34.     public int getSecond() {
  35.         return second;
  36.     }
  38.     public void setSecond(int second) {
  39.         this.second = second;
  40.     }
  42.     @Override
  43.     public int hashCode() {
  44.         final int prime = 31;
  45.         int result = 1;
  46.         result = prime * result + ((first == null) ? 0 : first.hashCode());
  47.         result = prime * result + second;
  48.         return result;
  49.     }
  51.     @Override
  52.     public boolean equals(Object obj) {
  53.         if (this == obj)
  54.             return true;
  55.         if (obj == null)
  56.             return false;
  57.         if (getClass() != obj.getClass())
  58.             return false;
  59.         PairWritable other = (PairWritable) obj;
  60.         if (first == null) {
  61.             if (other.first != null)
  62.                 return false;
  63.         } else if (!first.equals(other.first))
  64.             return false;
  65.         if (second != other.second)
  66.             return false;
  67.         return true;
  68.     }
  70.     public void write(DataOutput out) throws IOException {
  71.         out.writeUTF(this.getFirst());
  72.         out.writeInt(this.getSecond()); 
  73.     }
  75.     public void readFields(DataInput in) throws IOException {
  76.         this.setFirst(in.readUTF());
  77.         this.setSecond(in.readInt());
  78.     } 
  80.     public int compareTo(PairWritable o) {
  82.         // first
  83.         int comp = this.getFirst().compareTo(o.getFirst()) ;
  84.         if( 0 != comp ){
  85.             return comp;
  86.         }
  88.         // second 
  89.         return Integer.valueOf(this.getSecond()).compareTo(Integer.valueOf(o.getSecond())); 
  90.     }
  92.     @Override
  93.     public String toString() {
  94.         return first + "," + second ;
  95.     }
  96. }

FirstPartitioner.java

  1. package org.apache.hadoop.sort;
  3. import org.apache.hadoop.io.IntWritable;
  4. import org.apache.hadoop.mapreduce.Partitioner;
  6. public class FirstPartitioner extends Partitioner<PairWritable, IntWritable> {
  8.     @Override
  9.     public int getPartition(PairWritable key, IntWritable value,
  10.             int numPartitions) {
  11.         // TODO Auto-generated method stub
  12.         return (key.getFirst().hashCode() & Integer.MAX_VALUE) % numPartitions;
  13.     }
  14. }

FirstGroupingComparator.java

  1. package org.apache.hadoop.sort;
  3. import org.apache.hadoop.io.RawComparator;
  4. import org.apache.hadoop.io.WritableComparator;
  6. public class FirstGroupingComparator implements RawComparator<PairWritable> {
  8.     public int compare(PairWritable o1, PairWritable o2) {
  9.         return o1.getFirst().compareTo(o2.getFirst()); 
  11.     }
  13.     public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
  14.         return WritableComparator.compareBytes(b1, 0, l1-4, b2, 0, l2-4);
  15.     }
  16. }

3） 理解 MapReduce Join 的几种方式，编码实现 Reduce Join，提供源代码，说出思路。

a)不同方式

map join :两个待连接的表，其中一个非常大，另一个非常小，可将小表直接放于内存中，DistributeCache实现。
reduce join:两个待连接的大表
semi join：map 端Join和reduce 端Join结合。

b)：案例及思路

表文件信息：
customers.txt 按字段顺序排列 顾客编号 顾客姓名 顾客电话
以下为文件内容
1,Stephanie Leung,555-555-5555
2,Edward Kim,123-456-7890
3,Jose Madriz,281-330-8004
4,David Stork,408-555-0000
orders.txt 按字段顺序排列 顾客编号 商品名称 交易价格 交易时间
以下为文件内容
3,A,12.95,02-Jun-2008
1,B,88.25,20-May-2008
2,C,32.00,30-Nov-2007
3,D,25.02,22-Jan-2009

需求：把两个文件的内容按关联信息列示在一个虚拟表文件中。如 
1,Stephanie Leung,555-555-5555,B,88.25,20-May-2008
2,Edward Kim,123-456-7890,C,32.00,30-Nov-2007
...

分析思路：

map端：用关联字段作为key输出，本例中关联字段均为第一列顾客编号，用flag标签标记文件内容来源。
reduce端： 区分标签文件内容，按key关联2文件内容，输出到1个表文件。同一顾客编号多条订单信息按该顾客订单数量分行列示。

代码：
DistributedCacheWCMapReduce.java

  1. package com.ibeifeng.hadoop.mapreduce.cache;
  3. import java.io.BufferedReader;
  4. import java.io.IOException;
  5. import java.io.InputStream;
  6. import java.io.InputStreamReader;
  7. import java.net.URI;
  8. import java.util.ArrayList;
  9. import java.util.List;
  10. import java.util.StringTokenizer;
  12. import org.apache.hadoop.conf.Configuration;
  13. import org.apache.hadoop.conf.Configured;
  14. import org.apache.hadoop.filecache.DistributedCache;
  15. import org.apache.hadoop.fs.FileSystem;
  16. import org.apache.hadoop.fs.Path;
  17. import org.apache.hadoop.io.IntWritable;
  18. import org.apache.hadoop.io.LongWritable;
  19. import org.apache.hadoop.io.Text;
  20. import org.apache.hadoop.mapreduce.Job;
  21. import org.apache.hadoop.mapreduce.Mapper;
  22. import org.apache.hadoop.mapreduce.Reducer;
  23. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  24. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  25. import org.apache.hadoop.util.Tool;
  26. import org.apache.hadoop.util.ToolRunner;
  28. public class DistributedCacheWCMapReduce extends Configured implements Tool{
  30.     /**
  31.      * Mapper 
  32.      */
  33.     public static class DistributedCacheWCMapper extends
  34.             Mapper<LongWritable, Text, Text, IntWritable> {
  35.         // cache
  36.         List<String> list = new ArrayList<String>();
  38.         private final static IntWritable mapOutputValue = new IntWritable(1);
  39.         private Text mapOutputKey = new Text();
  41.         @SuppressWarnings("deprecation")
  42.         @Override
  43.         public void setup(Context context) throws IOException,
  44.                 InterruptedException {
  45.             // step 1: get configuraiton
  46.             Configuration conf = context.getConfiguration();
  47.             // step 2: get cache uri
  48.             URI[] uris = DistributedCache.getCacheFiles(conf);
  49.             // step 3: path
  50.             Path paht = new Path(uris[0]);
  51.             // step 4: file system
  52.             FileSystem fs = FileSystem.get(conf);
  53.             // step 5: in stream
  54.             InputStream inStream = fs.open(paht);
  55.             // step 6: read data
  56.             InputStreamReader isr = new InputStreamReader(inStream);
  57.             BufferedReader bf = new BufferedReader(isr);
  58.             String line ;
  59.             while ((line = bf.readLine()) != null){
  60.                 if(line.trim().length() > 0){
  61.             //      add element
  62.                     list.add(line);
  63.                 }
  64.             }
  65.             bf.close();
  66.             isr.close();
  67.             inStream.close();
  68.     //      fs.close();
  69.         }
  71.         @Override
  72.         public void map(LongWritable key, Text value, Context context)
  73.                 throws IOException, InterruptedException {
  74.             String lineValue = value.toString();
  75.             StringTokenizer st = new StringTokenizer(lineValue);
  76.             while(st.hasMoreTokens()){
  77.                 String wordValue = st.nextToken();
  79.                 if(list.contains(wordValue)){
  80.                     continue;
  81.                 }
  83.                 // set map output key
  84.                 mapOutputKey.set(wordValue);
  85.                 // output
  86.                 context.write(mapOutputKey, mapOutputValue);
  87.             }
  88.         }
  90.         @Override
  91.         public void cleanup(Context context) throws IOException,
  92.                 InterruptedException {
  93.             super.cleanup(context);
  94.         }
  95.     }
  96.     /**
  97.      * Reducer 
  98.      */
  99.     public static class DistributedCacheWCReducer extends
  100.             Reducer<Text, IntWritable, Text, IntWritable> {
  101.         public IntWritable outputValue = new IntWritable();
  103.         @Override
  104.         public void setup(Context context) throws IOException,
  105.                 InterruptedException {
  106.             super.setup(context);
  107.         }
  109.         @Override
  110.         public void reduce(Text key, Iterable<IntWritable> values,
  111.                 Context context) throws IOException, InterruptedException {
  112.             int sum = 0 ;
  113.             // iterator
  114.             for(IntWritable value : values){
  115.                 sum += value.get() ;
  116.             }
  117.             // set 
  118.             outputValue.set(sum);
  119.             // job output 
  120.             context.write(key, outputValue);
  121.         }
  123.         @Override
  124.         public void cleanup(Context context) throws IOException,
  125.                 InterruptedException {
  126.             super.cleanup(context);
  127.         }
  129.     }
  132.     /**
  133.      * Driver :Job create,set,submit,run,monitor
  134.      */
  135.     public int run(String[] args) throws Exception {
  136.         // step 1:get configuration
  137.         Configuration conf = this.getConf();
  139.         // step 2:create job
  140.         Job job = this.parseInputAndOutput(this,conf,args);
  142.         // step 4:set job 
  144.         // 2:mapper class
  145.         job.setMapperClass(DistributedCacheWCMapper.class);
  146.         job.setMapOutputKeyClass(Text.class);
  147.         job.setMapOutputValueClass(IntWritable.class);
  149.         // 4:reducer class
  150.         job.setReducerClass(DistributedCacheWCReducer.class);
  151.         job.setOutputKeyClass(Text.class);
  152.         job.setOutputValueClass(IntWritable.class);
  154.         // step 5:submit job
  155.         boolean isSuccess = job.waitForCompletion(true);
  157.         return isSuccess ? 0 : 1;
  158.     }
  160.     public Job parseInputAndOutput(Tool tool,Configuration conf,String[] args) throws Exception {
  162.         // validate args
  163.         if(args.length != 2){
  164.             System.err.println("Usage : " + tool.getClass().getSimpleName() + " [generic options] <input> <output>");
  165.             ToolRunner.printGenericCommandUsage(System.err);
  166.             return null;
  167.         }
  169.         // step 2:create job
  170.         Job job = Job.getInstance(conf, tool.getClass().getSimpleName());
  172. //      job.addCacheFile(uri);
  174.         // step 3:set job run class
  175.         job.setJarByClass(tool.getClass());
  177.         // 1:input format
  178.         Path inputPath = new Path(args[0]);
  179.         FileInputFormat.addInputPath(job, inputPath);
  181.         // 5:output format
  182.         Path outputPath = new Path(args[1]);
  183.         FileOutputFormat.setOutputPath(job, outputPath);
  185.         return job;
  186.     }
  188.     @SuppressWarnings("deprecation")
  189.     public static void main(String[] args) throws Exception {
  191.         //args = new String[]{
  192.         //      "hdfs://hadoop-yarn.beifeng.com:8020/user/beifeng/mr/distributedcache/input",//
  193.         //      "hdfs://hadoop-yarn.beifeng.com:8020/user/beifeng/mr/distributedcache/ouput"
  194.         //};
  196.         // mapreduce-default.xml,mapreduce-site.xml
  197.         Configuration conf = new Configuration();
  199.         // set distributed cache
  200.         // ===============================================================
  201.         URI uri = new URI("/user/beifeng/cachefile/cache.txt");
  202.         DistributedCache.addCacheFile(uri, conf);
  203.         // ===============================================================
  205.         // run mapreduce
  206.         int status = ToolRunner.run(conf, new DistributedCacheWCMapReduce(), args);
  208.         // exit program
  209.         System.exit(status);
  210.     }
  211. }

map输出value数据类型为自定义数据类型，命名为DataJoinWritable。 故创建DataJoinWritable类。

  1. package org.apache.hadoop.join;
  3. import java.io.DataInput;
  4. import java.io.DataOutput;
  5. import java.io.IOException;
  7. import org.apache.hadoop.io.Writable;
  9. public class DataJoinWritable implements Writable {
  11.     private String tag ;
  12.     private String data ;
  14.     public DataJoinWritable() {
  16.     }
  18.     public DataJoinWritable(String tag, String data) {
  19.         this.set(tag, data); 
  20.     }
  21.     public void set(String tag, String data) {
  22.         this.setTag(tag);
  23.         this.setData(data);
  24.     }
  25.     public String getTag() {
  26.         return tag;
  27.     }
  29.     public void setTag(String tag) {
  30.         this.tag = tag;
  31.     }
  33.     public String getData() {
  34.         return data;
  35.     }
  37.     public void setData(String data) {
  38.         this.data = data;
  39.     }
  41.     @Override
  42.     public int hashCode() {
  43.         final int prime = 31;
  44.         int result = 1;
  45.         result = prime * result + ((data == null) ? 0 : data.hashCode());
  46.         result = prime * result + ((tag == null) ? 0 : tag.hashCode());
  47.         return result;
  48.     }
  50.     @Override
  51.     public boolean equals(Object obj) {
  52.         if (this == obj)
  53.             return true;
  54.         if (obj == null)
  55.             return false;
  56.         if (getClass() != obj.getClass())
  57.             return false;
  58.         DataJoinWritable other = (DataJoinWritable) obj;
  59.         if (data == null) {
  60.             if (other.data != null)
  61.                 return false;
  62.         } else if (!data.equals(other.data))
  63.             return false;
  64.         if (tag == null) {
  65.             if (other.tag != null)
  66.                 return false;
  67.         } else if (!tag.equals(other.tag))
  68.             return false;
  69.         return true;
  70.     }
  72.     public void write(DataOutput out) throws IOException {
  73.         out.writeUTF(this.getTag());
  74.         out.writeUTF(this.getData());
  75.     }
  77.     public void readFields(DataInput in) throws IOException {
  78.         this.setTag(in.readUTF());
  79.         this.setData(in.readUTF());
  80.     }
  82.     @Override
  83.     public String toString() {
  84.         return tag + "," + data ;
  85.     }
  86. }
添加新批注
在作者公开此批注前，只有你和作者可见。
回复批注