[关闭]
@zhangyy 2017-08-28T13:05:20.000000Z 字数 12692 阅读 160

mapreduce 的二次排序

hadoop的部分

  • 一： 理解二次排序的功能， 使用自己理解的方式表达（包括自定义数据类型，分区，分组，排序）
  • 二： 编写实现二次排序功能， 提供源码文件。
  • 三：理解mapreduce join 的几种 方式，编码实现reduce join，提供源代码，说出思路。

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

  1.       1) partitioner
  2.                 job.setPartitionerClass(FirstPartitioner.class);
  3.        2) sort
  4.                 job.setSortComparatorClass(cls);
  6.      3) combine
  7.             job.setCombinerClass(cls);
  8.      4) compress
  9.          set by configuration
  10.      5) group
  11.     job.setGroupingComparatorClass(FirstGroupingComparator.class);

二： 编写实现二次排序功能， 提供源码文件。

  1. package org.apache.hadoop.studyhadoop.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.Reducer;
  14. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  15. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  16. import org.apache.hadoop.util.Tool;
  17. import org.apache.hadoop.util.ToolRunner;
  19. /**
  20.  * 
  21.  * @author zhangyy
  22.  *
  23.  */
  24. public class SecondarySortMapReduce extends Configured implements Tool{
  26.     // step 1: mapper class
  27.     /**
  28.      * public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
  29.      */
  30.     public static class SecondarySortMapper extends //
  31.         Mapper<LongWritable,Text,PairWritable,IntWritable>{
  33.         private PairWritable mapOutputKey = new PairWritable() ;
  34.         private IntWritable mapOutputValue = new IntWritable() ;
  36.         @Override
  37.         public void map(LongWritable key, Text value, Context context)
  38.                 throws IOException, InterruptedException {
  39.             // line value
  40.             String lineValue = value.toString();
  41.             // split
  42.             String[] strs = lineValue.split(",") ;
  43.             // invalidate
  44.             if(2 != strs.length){
  45.                 return ;
  46.             }
  48.             // set map  output key and value
  49.             mapOutputKey.set(strs[0], Integer.valueOf(strs[1]));
  50.             mapOutputValue.set(Integer.valueOf(strs[1]));
  52.             // output
  53.             context.write(mapOutputKey, mapOutputValue);
  54.         }
  55.     }
  57.     // step 2: reducer class
  58.     /**
  59.      * public class Reducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT>
  60.      */
  61.     public static class SecondarySortReducer extends //
  62.         Reducer<PairWritable,IntWritable,Text,IntWritable>{
  63.         private Text outputKey = new Text() ;
  65.         @Override
  66.         public void reduce(PairWritable key, Iterable<IntWritable> values,
  67.                 Context context)
  68.                 throws IOException, InterruptedException {
  69.             // set output key
  70.             outputKey.set(key.getFirst());
  72.             // iterator
  73.             for(IntWritable value : values){
  74.                 // output
  75.                 context.write(outputKey, value);
  76.             }
  77.         }
  78.     }
  81.     // step 3: driver
  82.     public int run(String[] args) throws Exception {
  83.         // 1: get configuration
  84.         Configuration configuration = super.getConf() ;
  86.         // 2: create job
  87.         Job job = Job.getInstance(//
  88.             configuration, //
  89.             this.getClass().getSimpleName()//
  90.         );
  91.         job.setJarByClass(this.getClass());
  93.         // 3: set job
  94.         // input  -> map  -> reduce -> output
  95.         // 3.1: input
  96.         Path inPath = new Path(args[0]) ;
  97.         FileInputFormat.addInputPath(job, inPath);
  99.         // 3.2: mapper
  100.         job.setMapperClass(SecondarySortMapper.class);
  101.         job.setMapOutputKeyClass(PairWritable.class);
  102.         job.setMapOutputValueClass(IntWritable.class);
  104. // ===========================Shuffle======================================     
  105.         // 1) partitioner
  106.                 job.setPartitionerClass(FirstPartitioner.class);
  107.         // 2) sort
  108. //              job.setSortComparatorClass(cls);
  109.         // 3) combine
  110. //              job.setCombinerClass(cls);
  111.         // 4) compress
  112.             // set by configuration
  113.         // 5) group
  114.         job.setGroupingComparatorClass(FirstGroupingComparator.class);
  115. // ===========================Shuffle======================================     
  118.         // 3.3: reducer
  119.         job.setReducerClass(SecondarySortReducer.class);
  120.         job.setOutputKeyClass(IntWritable.class);
  121.         job.setOutputValueClass(IntWritable.class);
  122.         // set reducer number
  123.         job.setNumReduceTasks(2);
  125.         // 3.4: output
  126.         Path outPath = new Path(args[1]);
  127.         FileOutputFormat.setOutputPath(job, outPath);
  129.         // 4: submit job 
  130.         boolean isSuccess = job.waitForCompletion(true);
  132.         return isSuccess ? 0 : 1 ;
  133.     }
  135.     public static void main(String[] args) throws Exception {
  136.         args = new String[]{
  137.                 "hdfs://namenode01.hadoop.com:8020/input/sort" ,//
  138.                 "hdfs://namenode01.hadoop.com:8020/output"
  139.             };      
  141.         // create configuration
  142.         Configuration configuration = new Configuration();
  144.         // run job
  145.         int status = ToolRunner.run(//
  146.             configuration, //
  147.             new SecondarySortMapReduce(), //
  148.             args
  149.         ) ;
  151.         // exit program
  152.         System.exit(status);
  153.     }
  154. }

PairWritable.java

  1. package org.apache.hadoop.studyhadoop.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;
  14.     public PairWritable() {
  15.     }
  17.     public PairWritable(String first, int second) {
  18.         this.set(first, second);
  19.     }
  21.     public void set(String first, int second) {
  22.         this.first = first;
  23.         this.setSecond(second);
  24.     }
  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 - Integer.MAX_VALUE;
  36.     }
  38.     public void setSecond(int second) {
  39.         this.second = second + Integer.MAX_VALUE;
  40.     }
  42.     public void write(DataOutput out) throws IOException {
  43.         out.writeUTF(first);
  44.         out.writeInt(second);
  45.     }
  47.     public void readFields(DataInput in) throws IOException {
  48.         this.first = in.readUTF();
  49.         this.second = in.readInt();
  50.     }
  52.     public int compareTo(PairWritable o) {
  53.         // compare first
  54.         int comp =this.first.compareTo(o.getFirst()) ;
  56.         // eqauls
  57.         if(0 != comp){
  58.             return comp ;
  59.         }
  61.         // compare
  62.         return Integer.valueOf(this.getSecond()).compareTo(Integer.valueOf(o.getSecond())) ;
  63.     }
  65. }

FirstPartitioner.java

  1. package org.apache.hadoop.studyhadoop.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.         return (key.getFirst().hashCode() & Integer.MAX_VALUE) % numPartitions;
  12.     }
  14. }

FirstGroupingComparator.java

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

上传数据处理：
hdfs dfs -put sort /input

运行输出：
12.png-116.8kB
2.png-41.5kB

3.png-10.7kB

三：理解mapreduce join 的几种 方式，编码实现reduce join，提供源代码，说出思路。

mapreduce join 有三种:

  1.    3.1 map 的端的join 
  2.    map阶段不能获取所有需要的join字段，即：同一个key对应的字段可能位于不同map中。Reduce side join是非常低效的，因为shuffle阶段要进行大量的数据传输。
  3. Map side join是针对以下场景进行的优化：两个待连接表中，有一个表非常大，而另一个表非常小，以至于小表可以直接存放到内存中。这样，我们可以将小表复制多份，让每个map task内存中存在一份（比如存放到hash table中），然后只扫描大表：对于大表中的每一条记录key/value，在hash table中查找是否有相同的key的记录，如果有，则连接后输出即可。
  4. 为了支持文件的复制，Hadoop提供了一个类DistributedCache 去实现。
  5.    3.2 reduce 的端的join 
  6.    在map阶段，map函数同时读取两个文件File1和File2，为了区分两种来源的key/value数据对，对每条数据打一个标签（tag）,比如：tag=0表示来自文件File1，tag=2表示来自文件File2。即：map阶段的主要任务是对不同文件中的数据打标签。
  7. 在reduce阶段，reduce函数获取key相同的来自File1和File2文件的value list， 然后对于同一个key，对File1和File2中的数据进行join（笛卡尔乘积）。即：reduce阶段进行实际的连接操作
  8.    3.3 SemiJoin
  9.    SemiJoin，也叫半连接，是从分布式数据库中借鉴过来的方法。它的产生动机是：对于reduce side join，跨机器的数据传输量非常大，这成了join操作的一个瓶颈，如果能够在map端过滤掉不会参加join操作的数据，则可以大大节省网络IO。
  10. 实现方法很简单：选取一个小表，假设是File1，将其参与join的key抽取出来，保存到文件File3中，File3文件一般很小，可以放到内存中。在map阶段，使用DistributedCache将File3复制到各个TaskTracker上，然后将File2中不在File3中的key对应的记录过滤掉，剩下的reduce阶段的工作与reduce side join相同

3.4 编程代码：
DataJoinMapReduce.java

  1. package org.apache.hadoop.studyhadoop.join;
  3. import java.io.IOException;
  4. import java.util.ArrayList;
  5. import java.util.List;
  7. import org.apache.hadoop.conf.Configuration;
  8. import org.apache.hadoop.conf.Configured;
  9. import org.apache.hadoop.fs.Path;
  10. import org.apache.hadoop.io.IntWritable;
  11. import org.apache.hadoop.io.LongWritable;
  12. import org.apache.hadoop.io.NullWritable;
  13. import org.apache.hadoop.io.Text;
  14. import org.apache.hadoop.mapreduce.Job;
  15. import org.apache.hadoop.mapreduce.Mapper;
  16. import org.apache.hadoop.mapreduce.Mapper.Context;
  17. import org.apache.hadoop.mapreduce.Reducer;
  18. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  19. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  20. import org.apache.hadoop.util.Tool;
  21. import org.apache.hadoop.util.ToolRunner;
  23. /**
  24.  * 
  25.  * @author zhangyy
  26.  * 
  27.  */
  28. public class DataJoinMapReduce extends Configured implements Tool {
  29.     // step 1 : mapper
  30.     /**
  31.      * public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
  32.      */
  33.     public static class WordCountMapper extends //
  34.             Mapper<LongWritable, Text, LongWritable, DataJoinWritable> {
  36.         private LongWritable mapOutputKey = new LongWritable();
  37.         private DataJoinWritable mapOutputValue = new DataJoinWritable();
  39.         @Override
  40.         public void map(LongWritable key, Text value, Context context)
  41.                 throws IOException, InterruptedException {
  43.             // split
  44.             String[] strs = value.toString().split(",");
  46.             // invalidate
  47.             if ((3 != strs.length) && (4 != strs.length)) {
  48.                 return;
  49.             }
  51.             // set mapoutput key
  52.             Long cid = Long.valueOf(strs[0]);
  53.             mapOutputKey.set(cid);
  55.             // set name
  56.             String name = strs[1];
  58.             // customer
  59.             if (3 == strs.length) {
  60.                 String phone = strs[2];
  61.                 mapOutputValue.set("customer", name + "," + phone);
  62.             }
  63.             // order
  64.             if (4 == strs.length) {
  65.                 String price = strs[2];
  66.                 String date = strs[3];
  67.                 mapOutputValue.set("order", name + "," + price + "," + date);
  68.             }
  70.             context.write(mapOutputKey, mapOutputValue);
  71.         }
  73.     }
  75.     // step 2 : reducer
  76.     public static class WordCountReducer extends //
  77.             Reducer<LongWritable, DataJoinWritable, NullWritable, Text> {
  79.         private Text outputValue = new Text();
  81.         @Override
  82.         public void reduce(LongWritable key, Iterable<DataJoinWritable> values,
  83.                 Context context) throws IOException, InterruptedException {
  85.             String customerInfo = new String();
  86.             List<String> orderList = new ArrayList<String>();
  88.             for (DataJoinWritable value : values) {
  89.                 if ("customer".equals(value.getTag())) {
  90.                     customerInfo = value.getData();
  91.                 } else if ("order".equals(value.getTag())) {
  92.                     orderList.add(value.getData());
  93.                 }
  94.             }
  96.             for (String order : orderList) {
  97.                 outputValue.set(key.toString() + "," + customerInfo + ","
  98.                         + order);
  99.                 context.write(NullWritable.get(), outputValue);
  100.             }
  102.         }
  103.     }
  105.     // step 3 : job
  107.     public int run(String[] args) throws Exception {
  109.         // 1 : get configuration
  110.         Configuration configuration = super.getConf();
  112.         // 2 : create job
  113.         Job job = Job.getInstance(//
  114.                 configuration,//
  115.                 this.getClass().getSimpleName());
  116.         job.setJarByClass(DataJoinMapReduce.class);
  118.         // job.setNumReduceTasks(tasks);
  120.         // 3 : set job
  121.         // input --> map --> reduce --> output
  122.         // 3.1 : input
  123.         Path inPath = new Path(args[0]);
  124.         FileInputFormat.addInputPath(job, inPath);
  126.         // 3.2 : mapper
  127.         job.setMapperClass(WordCountMapper.class);
  128.         // TODO
  129.         job.setMapOutputKeyClass(LongWritable.class);
  130.         job.setMapOutputValueClass(DataJoinWritable.class);
  132.         // ====================shuffle==========================
  133.         // 1: partition
  134.         // job.setPartitionerClass(cls);
  135.         // 2: sort
  136.         // job.setSortComparatorClass(cls);
  137.         // 3: combine
  138.         // job.setCombinerClass(cls);
  139.         // 4: compress
  140.         // set by configuration
  141.         // 5 : group
  142.         // job.setGroupingComparatorClass(cls);
  144.         // ====================shuffle==========================
  146.         // 3.3 : reducer
  147.         job.setReducerClass(WordCountReducer.class);
  148.         // TODO
  149.         job.setOutputKeyClass(NullWritable.class);
  150.         job.setOutputValueClass(Text.class);
  152.         // 3.4 : output
  153.         Path outPath = new Path(args[1]);
  154.         FileOutputFormat.setOutputPath(job, outPath);
  156.         // 4 : submit job
  157.         boolean isSuccess = job.waitForCompletion(true);
  158.         return isSuccess ? 0 : 1;
  160.     }
  162.     public static void main(String[] args) throws Exception {
  164.         args = new String[] {
  165.                 "hdfs://namenode01.hadoop.com:8020/join",
  166.                 "hdfs://namenode01.hadoop.com:8020/output3/" 
  167.                 };
  169.         // get configuration
  170.         Configuration configuration = new Configuration();
  172.         // configuration.set(name, value);
  174.         // run job
  175.         int status = ToolRunner.run(//
  176.                 configuration,//
  177.                 new DataJoinMapReduce(),//
  178.                 args);
  180.         // exit program
  181.         System.exit(status);
  182.     }
  184. }

DataJoinWritable.java

  1. package org.apache.hadoop.studyhadoop.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.     }
  87. }

上传文件：
hdfs dfs -put customers.txt /join
hdfs dfs -put orders.txt /join

运行结果：
1.png-142.4kB
2.png-49.7kB
3.png-28.3kB

添加新批注
在作者公开此批注前，只有你和作者可见。
回复批注