图文精华

About云»专题 › 技术学习(版主发帖区) › 大数据学习 › Mapreduce › 基于HIVE文件格式的map reduce代码编写
分享

基于HIVE文件格式的map reduce代码编写

yuwenge 发表于 2013-12-13 17:56:48 [显示全部楼层] 只看大图 回帖奖励 阅读模式 关闭右栏 2 8896

我们的数据绝大多数都是在HIVE上，对HIVE的SEQUENCEFILE和RCFILE的存储格式都有利用，为了满足HIVE的数据开放，hive client的方式就比较单一，直接访问HIVE生成的HDFS数据也是一种必要途径，所以本文整理测试了如何编写基于TEXTFILE、SEQUENCEFILE、RCFILE的数据的map reduce的代码。以wordcount的逻辑展示3种MR的代码。


其实只要知道MAP的输入格式是什么，就知道如何在MAP中处理数据；只要知道REDUCE(也可能只有MAP)的输出格式，就知道如何把处理结果转成输出格式。

表1：
0713372425.png
如下代码片段是运行一个MR的最简单的配置：

  1. //map/reduce的job配置类，向hadoop框架描述map-reduce执行的工作 

  2. JobConf conf = new JobConf(WordCountRC.class);

  3. //设置一个用户定义的job名称

  4. conf.setJobName("WordCountRC");

  5. 

  6. //为job的输出数据设置Key类

  7. conf.setOutputKeyClass(Text.class);

  8. //为job输出设置value类 

  9. conf.setOutputValueClass(IntWritable.class);

  10. 

  11. //为job设置Mapper类

  12. conf.setMapperClass(MapClass.class);

  13. //为job设置Combiner类

  14. conf.setCombinerClass(Reduce.class);

  15. //为job设置Reduce类

  16. conf.setReducerClass(Reduce.class);

  17. 

  18. //为map-reduce任务设置InputFormat实现类

  19. conf.setInputFormat(RCFileInputFormat.class);

  20. //为map-reduce任务设置OutputFormat实现类

  21. conf.setOutputFormat(TextOutputFormat.class);

  22. 

  23. //为map-reduce job设置路径数组作为输入列表

  24. FileInputFormat.setInputPaths(conf, new Path(args[0]));

  25. //为map-reduce job设置路径数组作为输出列表

  26. FileOutputFormat.setOutputPath(conf, new Path(args[1]));

  27. 

  28. //运行一个job

  29. JobClient.runJob(conf);</font></font>
复制代码

而此刻，我们更多的是关注配置InputFormat和OutputFormat的setInputFormat和setOutputFormat。根据我们不同的输入输出做相应的配置，可以选择表1的任何格式。
当我们确定了输入输出格式，接下来就是来在实现map和reduce函数时首选对输入格式做相应的处理，然后处理具体的业务逻辑，最后把处理后的数据转成既定的输出格式。


如下是处理textfile、sequencefile、rcfile输入文件的wordcount代码，大家可以比较一下具体区别，应该就能处理更多其它输入文件或者输出文件格式的数据。
代码1：textfile版wordcount

  1. import java.io.IOException;

  2. import java.util.Iterator;

  3. import java.util.StringTokenizer;

  4. 

  5. import org.apache.hadoop.fs.Path;

  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.mapred.FileInputFormat;

  10. import org.apache.hadoop.mapred.FileOutputFormat;

  11. import org.apache.hadoop.mapred.JobClient;

  12. import org.apache.hadoop.mapred.JobConf;

  13. import org.apache.hadoop.mapred.MapReduceBase;

  14. import org.apache.hadoop.mapred.Mapper;

  15. import org.apache.hadoop.mapred.OutputCollector;

  16. import org.apache.hadoop.mapred.Reducer;

  17. import org.apache.hadoop.mapred.Reporter;

  18. 

  19. 

  20. public class WordCountTxt{

  21.  

  22.   public static class MapClass extends MapReduceBase

  23.     implements Mapper<LongWritable, Text, Text, IntWritable> {

  24.    

  25.     private final static IntWritable one = new IntWritable(1);

  26.     private Text word = new Text();

  27.    

  28.        @Override

  29.        public void map(LongWritable key, Text value,

  30.                      OutputCollector<Text, IntWritable> output,

  31.             Reporter reporter) throws IOException {

  32.               String line = value.toString();

  33.               StringTokenizer itr = new StringTokenizer(line);

  34.               while (itr.hasMoreTokens()) {

  35.                      word.set(itr.nextToken());

  36.                      output.collect(word, one);

  37.               }

  38.   }

  39.   }

  40. 

  41.   public static class Reduce extends MapReduceBase

  42.     implements Reducer<Text, IntWritable, Text, IntWritable> {

  43.    

  44.        @Override

  45.     public void reduce(Text key, Iterator<IntWritable> values,

  46.                        OutputCollector<Text, IntWritable> output,

  47.                        Reporter reporter) throws IOException {

  48.       int sum = 0;

  49.       while (values.hasNext()) {

  50.         sum += values.next().get();

  51.       }

  52.       output.collect(key, new IntWritable(sum));

  53.     }

  54.   }

  55.  

  56.   public static void main(String[] args) throws Exception {

  57.          JobConf conf = new JobConf(WordCountTxt.class);

  58.          conf.setJobName("wordcounttxt");

  59.         

  60.          conf.setOutputKeyClass(Text.class);

  61.          conf.setOutputValueClass(IntWritable.class);

  62.         

  63.          conf.setMapperClass(MapClass.class);

  64.          conf.setCombinerClass(Reduce.class);

  65.          conf.setReducerClass(Reduce.class);

  66.         

  67.          FileInputFormat.setInputPaths(conf, new Path(args[0]));

  68.          FileOutputFormat.setOutputPath(conf, new Path(args[1]));

  69.               

  70.          JobClient.runJob(conf);   

  71.   }

  72.   

  73. }

  74. 

  75. 

  76. 代码2：sequencefile版wordcount

  77. import java.io.IOException;

  78. import java.util.Iterator;

  79. import java.util.StringTokenizer;

  80. 

  81. import org.apache.hadoop.fs.Path;

  82. import org.apache.hadoop.io.IntWritable;

  83. import org.apache.hadoop.io.Text;

  84. import org.apache.hadoop.mapred.FileInputFormat;

  85. import org.apache.hadoop.mapred.FileOutputFormat;

  86. import org.apache.hadoop.mapred.JobClient;

  87. import org.apache.hadoop.mapred.JobConf;

  88. import org.apache.hadoop.mapred.MapReduceBase;

  89. import org.apache.hadoop.mapred.Mapper;

  90. import org.apache.hadoop.mapred.OutputCollector;

  91. import org.apache.hadoop.mapred.Reducer;

  92. import org.apache.hadoop.mapred.Reporter;

  93. import org.apache.hadoop.mapred.SequenceFileAsTextInputFormat;

  94. import org.apache.hadoop.mapred.TextOutputFormat;

  95. 

  96. 

  97. public class WordCountSeq {

  98. 

  99.          public static class MapClass extends MapReduceBase

  100.            implements Mapper<Text, Text, Text, IntWritable> {

  101.           

  102.            private final static IntWritable one = new IntWritable(1);

  103.            private Text word = new Text();

  104.           

  105.               @Override

  106.               public void map(Text key, Text value,

  107.                            OutputCollector<Text, IntWritable> output,

  108.                    Reporter reporter) throws IOException {

  109.                      String line = value.toString();

  110.                      StringTokenizer itr = new StringTokenizer(line);

  111.                      while (itr.hasMoreTokens()) {

  112.                            word.set(itr.nextToken());

  113.                            output.collect(word, one);

  114.                      }

  115.          }

  116.          }

  117. 

  118.          public static class Reduce extends MapReduceBase

  119.            implements Reducer<Text, IntWritable, Text, IntWritable> {

  120.           

  121.               @Override

  122.            public void reduce(Text key, Iterator<IntWritable> values,

  123.                               OutputCollector<Text, IntWritable> output,

  124.                               Reporter reporter) throws IOException {

  125.              int sum = 0;

  126.              while (values.hasNext()) {

  127.                sum += values.next().get();

  128.              }

  129.              output.collect(key, new IntWritable(sum));

  130.            }

  131.          }

  132.          /**

  133.           * @param args

  134.         * @throws IOException

  135.           */

  136.          public static void main(String[] args) throws IOException {

  137.               // TODO Auto-generated method stub

  138.                 JobConf conf = new JobConf(WordCountSeq.class);

  139.                 conf.setJobName("wordcountseq");

  140.                

  141.                 conf.setOutputKeyClass(Text.class);

  142.                 conf.setOutputValueClass(IntWritable.class);

  143.                

  144.                 conf.setMapperClass(MapClass.class);

  145.                 conf.setCombinerClass(Reduce.class);

  146.                 conf.setReducerClass(Reduce.class);

  147.                

  148.                 conf.setInputFormat(SequenceFileAsTextInputFormat.class);

  149.                 conf.setOutputFormat(TextOutputFormat.class);

  150.                

  151.                 FileInputFormat.setInputPaths(conf, new Path(args[0]));

  152.                 FileOutputFormat.setOutputPath(conf, new Path(args[1]));

  153.                      

  154.                 JobClient.runJob(conf);

  155.          }

  156. 

  157. }

  158. 

  159. 

  160. 代码3：rcfile版wordcount

  161. import java.io.IOException;

  162. import java.util.Iterator;

  163. 

  164. import org.apache.hadoop.fs.Path;

  165. import org.apache.hadoop.hive.ql.io.RCFileInputFormat;

  166. import org.apache.hadoop.hive.serde2.columnar.BytesRefArrayWritable;

  167. import org.apache.hadoop.io.IntWritable;

  168. import org.apache.hadoop.io.LongWritable;

  169. import org.apache.hadoop.io.Text;

  170. import org.apache.hadoop.mapred.FileInputFormat;

  171. import org.apache.hadoop.mapred.FileOutputFormat;

  172. import org.apache.hadoop.mapred.JobClient;

  173. import org.apache.hadoop.mapred.JobConf;

  174. import org.apache.hadoop.mapred.MapReduceBase;

  175. import org.apache.hadoop.mapred.Mapper;

  176. import org.apache.hadoop.mapred.OutputCollector;

  177. import org.apache.hadoop.mapred.Reducer;

  178. import org.apache.hadoop.mapred.Reporter;

  179. import org.apache.hadoop.mapred.TextOutputFormat;

  180. 

  181. public class WordCountRC {

  182.     

  183.      public static class MapClass

  184.           extends MapReduceBase implements Mapper<LongWritable, BytesRefArrayWritable, Text, IntWritable> {

  185.          

  186.           private final static IntWritable one = new IntWritable(1);

  187.           private Text word =new Text();

  188.     

  189.           @Override

  190.           public void map(LongWritable key, BytesRefArrayWritable value,

  191.                     OutputCollector<Text, IntWritable> output, Reporter reporter)

  192.                     throws IOException {

  193.                Text txt = new Text();

  194.                txt.set(value.get(0).getData(), value.get(0).getStart(), value.get(0).getLength());

  195.                String[] result = txt.toString().split("\\s");

  196.                for(int i=0; i < result.length; i++){

  197.                     word.set(result[i]);

  198.                     output.collect(word, one);    

  199.                }

  200.           }         

  201.      }

  202. 

  203.      public static class Reduce

  204.           extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> {

  205.     

  206.           private IntWritable result = new IntWritable();

  207.          

  208.           @Override

  209.           public void reduce(Text key, Iterator<IntWritable> value,

  210.                     OutputCollector<Text, IntWritable> output, Reporter reporter)

  211.                     throws IOException {

  212.                int sum = 0;

  213.                while (value.hasNext()) {

  214.                     sum += value.next().get();

  215.                }

  216.               

  217.                result.set(sum);

  218.                output.collect(key, result);              

  219.           }

  220.          

  221.      }

  222.      /**

  223.      * @param args

  224.      */

  225.      public static void main(String[] args) throws IOException{

  226.           JobConf conf = new JobConf(WordCountRC.class);

  227.           conf.setJobName("WordCountRC");

  228.          

  229.           conf.setOutputKeyClass(Text.class);

  230.           conf.setOutputValueClass(IntWritable.class);

  231.          

  232.           conf.setMapperClass(MapClass.class);

  233.           conf.setCombinerClass(Reduce.class);

  234.           conf.setReducerClass(Reduce.class);

  235.          

  236.           conf.setInputFormat(RCFileInputFormat.class);

  237.           conf.setOutputFormat(TextOutputFormat.class);

  238.          

  239.           FileInputFormat.setInputPaths(conf, new Path(args[0]));

  240.           FileOutputFormat.setOutputPath(conf, new Path(args[1]));

  241.          

  242.           JobClient.runJob(conf);

  243.      }

  244. }

  245. 

  246. 

  247. 原始数据：

  248. hadoop fs -text /group/alidw-dev/seq_input/attempt_201201101606_2339628_m_000000_0

  249. 12/02/13 17:07:57 INFO util.NativeCodeLoader: Loaded the native-hadoop library

  250. 12/02/13 17:07:57 INFO zlib.ZlibFactory: Successfully loaded & initialized native-zlib library

  251. 12/02/13 17:07:57 INFO compress.CodecPool: Got brand-new decompressor

  252. 12/02/13 17:07:57 INFO compress.CodecPool: Got brand-new decompressor

  253. 12/02/13 17:07:57 INFO compress.CodecPool: Got brand-new decompressor

  254. 12/02/13 17:07:57 INFO compress.CodecPool: Got brand-new decompressor

  255.         hello, i am ok. are you?

  256.         i am fine too!</font></font>
复制代码

编译打包完成后执行：

hadoop jarWordCountSeq.jar WordCountSeq /group/alidw-dev/seq_input/ /group/alidw-dev/rc_output



执行完毕就能看到最终结果：

  1. hadoop fs -cat /group/alidw-dev/seq_output/part-00000

  2. am      2

  3. are     1

  4. fine    1

  5. hello,  1

  6. i       2

  7. ok.     1

  8. too!    1

  9. you?    1</font></font>
复制代码
关注公众号，获取大数据、人工智能20套、区块链资源5阶段等资源，随时更新，获取最新技术资源
回复

使用道具 举报

自定义广告语

已有(2)人评论

跳转到指定楼层
正序浏览
qingfeng 发表于 2014-12-3 15:35:22
你好，想请教一个问题：我也是按照你的上面的方法写的，但不知道jar包有哪些？比如RCFileInputFormat和RCFileRecordReader还要将hive源代码中复制到java工程中，但是运行还是错误：Error: org/apache/hadoop/hive/ql/io/RCFile$Reader，谢谢！
回复

使用道具 举报

sstutu 发表于 2014-12-3 16:06:15
qingfeng 发表于 2014-12-3 15:35
你好，想请教一个问题：我也是按照你的上面的方法写的，但不知道jar包有哪些？比如RCFileInputFormat和RCFi ...


需要引用一些jar包，把它们导入工程中
回复

使用道具 举报

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
yuwenge

版主

关注

  • 518

    主题

  • 766

    帖子

  • 59

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2025 About云 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000