ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 大数据 > Hadoop > 一个MapReduce 程序示例 细节决定成败(七) ：自定义Key 及RecordReader

一个MapReduce 程序示例 细节决定成败(七) ：自定义Key 及RecordReader

原创 Hadoop 作者：self_control 时间：2016-05-30 14:51:31 0 删除 编辑
上一篇中，演示了如何使用CombineFileInputFormat 来优化当有多个输入小文件时，减少起动的map task个数。
在自定义的MyCombineFileInputFormat中的MyRecordReader是简单代理了LineRecordReader。
其它我也还可以在这个地方做更多的东西。
本次实验是使用自定义的RecordReader从split中自定义 key value。

自定义MyKey
自定义的key 需要实现WritableComparable  接口。

点击(此处)折叠或打开

  1. package wordcount;

  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 MyKey implements WritableComparable<MyKey> {

  11.         private char c;
  12.         @Override
  13.         public void write(DataOutput out) throws IOException {
  14.                 out.writeChar(c);
  15.         }

  17.         @Override
  18.         public void readFields(DataInput in) throws IOException {
  19.                 c= in.readChar();
  20.         }

  22.         @Override
  23.         public int compareTo(MyKey key) {
  24.                 if(c==key.c)
  25.                         return 0;
  26.                 else if(c> key.c)
  27.                         return 1;
  28.                 else
  29.                         return -1;
  30.         }

  32.         public char getC() {
  33.                 return c;
  34.         }

  36.         public void setC(char c) {
  37.                 this.c = c;
  38.         }


  41. }

自定义CombinedFilesInputFormat 自定义RecordReader

点击(此处)折叠或打开

  1. package wordcount;

  3. import java.io.IOException;

  5. import org.apache.commons.lang.StringUtils;
  6. import org.apache.hadoop.io.IntWritable;
  7. import org.apache.hadoop.io.LongWritable;
  8. import org.apache.hadoop.io.Text;
  9. import org.apache.hadoop.mapreduce.InputSplit;
  10. import org.apache.hadoop.mapreduce.RecordReader;
  11. import org.apache.hadoop.mapreduce.TaskAttemptContext;
  12. import org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat;
  13. import org.apache.hadoop.mapreduce.lib.input.CombineFileRecordReader;
  14. import org.apache.hadoop.mapreduce.lib.input.CombineFileSplit;
  15. import org.apache.hadoop.mapreduce.lib.input.FileSplit;
  16. import org.apache.hadoop.mapreduce.lib.input.LineRecordReader;
  17. import org.apache.log4j.Logger;


  20. public class MyCombinedFilesInputFormat extends CombineFileInputFormat<LongWritable, Text> {

  22.         @SuppressWarnings({ "unchecked", "rawtypes" })
  23.         @Override
  24.         public RecordReader<LongWritable, Text> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException {
  25.                 return new CombineFileRecordReader((CombineFileSplit) split,context,MyCombinedFilesRecordReader.class);
  26.         }

  28.         public static class MyCombinedFilesRecordReader extends RecordReader<MyKey, IntWritable> {
  29.                 private int index;
  30.                 private LineRecordReader reader;

  32.                 private String tValue;
  33.                 private int pos=0;

  35.                 private MyKey key = new MyKey();

  37.                 Logger log = Logger.getLogger(MyCombinedFilesRecordReader.class);
  38.                 public MyCombinedFilesRecordReader(CombineFileSplit split, TaskAttemptContext context, Integer index) {
  39.                         this.index = index;
  40.                         reader = new LineRecordReader();
  41.                 }

  43.                 @Override
  44.                 public void initialize(InputSplit split, TaskAttemptContext context)
  45.                                 throws IOException, InterruptedException {
  46.                         CombineFileSplit cfsplit = (CombineFileSplit) split;
  47.                         FileSplit fileSplit = new FileSplit(cfsplit.getPath(index),
  48.                                                                                                 cfsplit.getOffset(index),
  49.                                                                                                 cfsplit.getLength(index),
  50.                                                                                                 cfsplit.getLocations()
  51.                                         );
  52.                         reader.initialize(fileSplit, context);
  53.                 }

  55.                 @Override
  56.                 public boolean nextKeyValue() throws IOException, InterruptedException {
  57.                         if(StringUtils.isEmpty(tValue)||pos>=tValue.length()-1){
  58.                                 if(reader.nextKeyValue()){
  59.                                         pos = 0;
  60.                                         this.tValue = reader.getCurrentValue().toString();
  61.                                         return true;
  62.                                 }
  63.                                 else{
  64.                                         return false;
  65.                                 }
  66.                         }
  67.                         else{
  68.                                 pos ++;
  69.                                 if(tValue.charAt(pos)<='z' && tValue.charAt(pos)>='A'){
  70.                                         return true;
  71.                                 }
  72.                                 else{
  73.                                         return nextKeyValue();
  74.                                 }
  75.                         }
  76.                 }

  78.                 @Override
  79.                 public MyKey getCurrentKey() throws IOException,
  80.                                 InterruptedException {
  81.                         key.setC(tValue.charAt(pos));
  82.                         return key;
  83.                 }

  85.                 @Override
  86.                 public IntWritable getCurrentValue() throws IOException, InterruptedException {
  87.                         return new IntWritable(1);
  88.                 }

  90.                 @Override
  91.                 public float getProgress() throws IOException, InterruptedException {
  92.                         return reader.getProgress();
  93.                 }

  95.                 @Override
  96.                 public void close() throws IOException {
  97.                         reader.close();
  98.                 }

  100.         }
  101. }
Mapper程序代码
可以看到使用MyRecordReader返回自定义Key后，Map 函数得到了很大的简化。

点击(此处)折叠或打开

  1. public static class MyWordCountMapper extends
  2.                         Mapper<MyKey, NullWritable, Text, IntWritable> {
  3.                 Text mKey = new Text();
  4.                 IntWritable mValue = new IntWritable(1);
  5.                 @Override
  6.                 protected void map(MyKey key, NullWritable value, Context context)
  7.                                 throws IOException, InterruptedException {
  8.                         mKey.set(String.valueOf(key.getC()));
  9.                         context.write(mKey, mValue);
  10.                 }
  11.         }
为了方便以后查看，把主程序代码也贴上来。

点击(此处)折叠或打开

  1. package wordcount;

  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.NullWritable;
  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.TextInputFormat;
  15. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  16. import org.apache.hadoop.util.Tool;
  17. import org.apache.hadoop.util.ToolRunner;
  18. import org.apache.log4j.Logger;

  20. public class MyWordCountJob extends Configured implements Tool {
  21.         Logger log = Logger.getLogger(MyWordCountJob.class);

  23.         public static class MyWordCountMapper extends
  24.                         Mapper<MyKey, NullWritable, Text, IntWritable> {
  25.                 Text mKey = new Text();
  26.                 IntWritable mValue = new IntWritable(1);
  27.                 @Override
  28.                 protected void map(MyKey key, NullWritable value, Context context)
  29.                                 throws IOException, InterruptedException {
  30.                         mKey.set(String.valueOf(key.getC()));
  31.                         context.write(mKey, mValue);
  32.                 }
  33.         }

  35.         public static class MyWordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
  36.                 Text rkey = new Text();
  37.                 IntWritable rvalue = new IntWritable(1);
  38.                 @Override
  39.                 protected void reduce(Text key, Iterable<IntWritable> values,Context context)
  40.                                 throws IOException, InterruptedException {

  42.                         int n=0;
  43.                         for(IntWritable value :values){
  44.                                 n+= value.get();
  45.                         }
  46.                         rvalue.set(n);
  47.                         context.write(key, rvalue);
  48.                 }
  49.         }

  51.         @Override
  52.         public int run(String[] args) throws Exception {
  53.                 //valid the parameters
  54.                 if(args.length !=2){
  55.                         return -1;
  56.                 }

  58.                 Job job = Job.getInstance(getConf(), "MyWordCountJob");
  59.                 job.setJarByClass(MyWordCountJob.class);

  61.                 Path inPath = new Path(args[0]);
  62.                 Path outPath = new Path(args[1]);

  64.                 outPath.getFileSystem(getConf()).delete(outPath,true);
  65.                 TextInputFormat.setInputPaths(job, inPath);
  66.                 TextOutputFormat.setOutputPath(job, outPath);


  69.                 job.setMapperClass(MyWordCountJob.MyWordCountMapper.class);
  70.                 job.setReducerClass(MyWordCountJob.MyWordCountReducer.class);

  72.                 job.setInputFormatClass(MyCombinedFilesInputFormat.class);
  73.                 MyCombinedFilesInputFormat.setMaxInputSplitSize(job, 1024*1024*64);
  74.                 job.setOutputFormatClass(TextOutputFormat.class);
  75.                 job.setMapOutputKeyClass(Text.class);
  76.                 job.setMapOutputValueClass(IntWritable.class);
  77.                 job.setOutputKeyClass(Text.class);
  78.                 job.setOutputValueClass(IntWritable.class);


  81.                 return job.waitForCompletion(true)?0:1;
  82.         }
  83.         public static void main(String [] args){
  84.                 int result = 0;
  85.                 try {
  86.                         result = ToolRunner.run(new Configuration(), new MyWordCountJob(), args);
  87.                 } catch (Exception e) {
  88.                         e.printStackTrace();
  89.                 }
  90.                 System.exit(result);
  91.         }

  93. }

而且使用自定义的MyRecordReader的好处不限于此，稍后一篇讨论使用TotalOrderPartitioner时会发现，对于词频统计，使用TotalOrderPartitioner 自定义RecordReader是必要的。

来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/30066956/viewspace-2109264/ ，如需转载，请注明出处，否则将追究法律责任。

0
0
上一篇： 一个MapReduce 程序示例 细节决定成败(六) ：CombineFileInputFormat
下一篇： 一个MapReduce 程序示例 细节决定成败(八)：TotalOrderPartitioner
请登录后发表评论 登录
全部评论
self_control

注册时间：2014-12-16

  • 博文量
    49
  • 访问量
    406892

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593