码迷，mamicode.com
首页 > 其他好文 > 详细

Hadoop中的GenericWritable

时间：2016-01-31 21:22:10      阅读：375      评论：0      收藏：0      [点我收藏+]

标签：

一：背景

某些业务的数据来源可能不同，且数据源中的分割方式也不同，导致在MapReduce编程时使用的格式化类会不同，为了包装不同的Map输出，Hadoop提供了GenericWritable类，允许我们同时操作多个不同的Map输出，输出到一个Reduce中进行处理。

 

技术实现：

我们对HDFS中两个不同的数据源进行处理，数据源如下(hello文件中的内容是通过制表符来分割的，hello2中的内容是通过逗号来分割的)：

技术分享

 

为了同时处理这两个文件，我们要使用不同的Map进行处理。源码如下：

 

[java] view plain copy
 
  1. public class MyGenericWritableTest {  
  2.   
  3.     //定义输入路径  
  4.     private static final String INPUT_PATH = "hdfs://liaozhongmin:9000/files";  
  5.     //定义输出路径  
  6.     private static final String OUT_PATH = "hdfs://liaozhongmin:9000/out";  
  7.       
  8.     public static void main(String[] args) {  
  9.           
  10.         try {  
  11.             //创建配置信息  
  12.             Configuration conf = new Configuration();  
  13.             //创建文件系统  
  14.             FileSystem fileSystem = FileSystem.get(new URI(OUT_PATH), conf);  
  15.             //如果输出目录存在，我们就删除  
  16.             if (fileSystem.exists(new Path(OUT_PATH))){  
  17.                 fileSystem.delete(new Path(OUT_PATH), true);  
  18.             }  
  19.               
  20.             //创建任务  
  21.             Job job = new Job(conf, MyGenericWritableTest.class.getName());  
  22.               
  23.             //1.1 设置输入目录和设置格式化的类(这是使用GenericWritable的关键)  
  24.             MultipleInputs.addInputPath(job, new Path(INPUT_PATH + "/hello"), TextInputFormat.class, MyGenericMapper1.class);  
  25.             MultipleInputs.addInputPath(job, new Path(INPUT_PATH + "/hello2"), TextInputFormat.class, MyGenericMapper2.class);  
  26.               
  27.             //1.2 设置自定义Mapper类和map函数输出数据的key和value的类型  
  28.             job.setMapOutputKeyClass(Text.class);  
  29.             job.setMapOutputValueClass(MyGenericWritable.class);  
  30.               
  31.             //1.3 设置分区和reduce的数量(reduce的数量和分区的数量对应，因为分区的数量为一个所以reduce的数量也为一个)  
  32.             job.setPartitionerClass(HashPartitioner.class);  
  33.             job.setNumReduceTasks(1);  
  34.               
  35.             //1.4 排序 分组  
  36.             //1.5 归约  
  37.             //2.1 Shuffle过程将map端输出的数据拷贝到reduce端  
  38.               
  39.             //2.2 设置自定义的Reducer类和输出的key和value类型  
  40.             job.setReducerClass(MyGenericReducer.class);  
  41.             job.setOutputKeyClass(Text.class);  
  42.             job.setOutputValueClass(LongWritable.class);  
  43.               
  44.             //2.3设置输出的格式化类和指定输出路径  
  45.             job.setOutputFormatClass(TextOutputFormat.class);  
  46.             FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));  
  47.               
  48.             //提交作业  
  49.             System.exit(job.waitForCompletion(true) ? 0 : 1);  
  50.               
  51.         } catch (Exception e) {  
  52.             e.printStackTrace();  
  53.         }  
  54.       
  55.           
  56.     }  
  57.       
  58.     /** 
  59.      * 自定义Mapper类用于处理hello1文件,该文件中的内容是以制表符进行分割的 
  60.      * @author 廖钟民 
  61.      * time : 2015年1月15日下午5:12:18 
  62.      * @version 
  63.      */  
  64.     public static class MyGenericMapper1 extends Mapper<LongWritable, Text, Text, MyGenericWritable>{  
  65.           
  66.         @Override  
  67.         protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, MyGenericWritable>.Context context) throws IOException,  
  68.                 InterruptedException {  
  69.                 //对value按制表符进行切分  
  70.                 String[] splits = value.toString().split("\t");  
  71.                 //遍历字符串数组通过context写出去  
  72.                 for (String str : splits){  
  73.                     context.write(new Text(str), new MyGenericWritable(new LongWritable(1)));  
  74.                 }  
  75.         }  
  76.     }  
  77.       
  78.     /** 
  79.      * 自定义Mapper类用于处理hello2文件，该文件的输入格式化类是TextInputFormat，用逗号给开 
  80.      * @author 廖钟民 
  81.      * time : 2015年1月15日下午5:13:30 
  82.      * @version 
  83.      */  
  84.     public static class MyGenericMapper2 extends Mapper<LongWritable, Text, Text, MyGenericWritable>{  
  85.         protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, MyGenericWritable>.Context context) throws IOException,  
  86.                 InterruptedException {  
  87.             //对value进行切割  
  88.             String[] splits = value.toString().split(",");  
  89.               
  90.             //遍历字符串数组通过context写出去  
  91.             for (String str : splits){  
  92.                 //为了模拟不同的map端输出，我们这里故意设置一个Text类型  
  93.                 context.write(new Text(str), new MyGenericWritable(new Text("1")));  
  94.             }  
  95.               
  96.         }  
  97.     }  
  98.       
  99.     /** 
  100.      * 自定义Reducer类，对不同Map端的输出进行汇总 
  101.      * @author 廖钟民 
  102.      * time : 2015年1月15日下午7:09:27 
  103.      * @version 
  104.      */  
  105.     public static class MyGenericReducer extends Reducer<Text, MyGenericWritable, Text, LongWritable>{  
  106.         @Override  
  107.         protected void reduce(Text key, Iterable<MyGenericWritable> values, Reducer<Text, MyGenericWritable, Text, LongWritable>.Context context)  
  108.                 throws IOException, InterruptedException {  
  109.             //定义单词出现的总次数  
  110.             long count = 0L;  
  111.             //遍历集合进行叠加  
  112.             for (MyGenericWritable time : values){  
  113.                 //获取MyGenericWritable对象  
  114.                 Writable writable = time.get();  
  115.                 //如果当前是LongWritable类型  
  116.                 if (writable instanceof LongWritable){  
  117.                       
  118.                     count += ((LongWritable) writable).get();  
  119.                 }  
  120.                 //如果当前是Text类型  
  121.                 if (writable instanceof Text){  
  122.                     count += Long.parseLong(((Text)writable).toString());  
  123.                 }  
  124.             }  
  125.             //把最后计算所得的结果写出去  
  126.             context.write(key, new LongWritable(count));  
  127.         }  
  128.     }  
  129. }  
  130.   
  131. /** 
  132.  * 继承GenericWritable进行自定义 
  133.  * @author 廖钟民 
  134.  * time : 2015年1月15日下午4:29:50 
  135.  * @version 
  136.  */  
  137. class MyGenericWritable extends GenericWritable{  
  138.   
  139.     //无参构造函数  
  140.     public MyGenericWritable() {  
  141.           
  142.     }  
  143.       
  144.     //有参构造函数  
  145.     public MyGenericWritable(Text text) {  
  146.         super.set(text);  
  147.     }  
  148.       
  149.     //有参构造函数  
  150.     public MyGenericWritable(LongWritable longWritable) {  
  151.         super.set(longWritable);  
  152.     }  
  153.   
  154.       
  155.     @Override  
  156.     protected Class<? extends Writable>[] getTypes() {  
  157.           
  158.         return new Class[]{LongWritable.class,Text.class};  
  159.     }  
  160.       
  161. }  

输出结果：

 

技术分享

Hadoop中的GenericWritable

标签：

原文地址：http://www.cnblogs.com/thinkpad/p/5173725.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！