大数据之HBase MapReduce的实例分析

Posted 2021-04-18 大数据技术工程师

引言
跟Hadoop的无缝集成使得使用MapReduce对HBase的数据进行分布式计算非常方便，本文将以前面的blog示例，介绍HBase下MapReduce开发要点。很好理解本文前提是你对Hadoop MapReduce有一定的了解。
HBase MapReduce核心类介绍
首先一起来回顾下MapReduce的基本编程模型，
 
可以看到最基本的是通过Mapper和Reducer来处理KV对，Mapper的输出经Shuffle及Sort后变为Reducer的输入。除了Mapper和Reducer外，另外两个重要的概念是InputFormat和OutputFormat，定义了Map-Reduce的输入和输出相关的东西。HBase通过对这些类的扩展（继承）来方便MapReduce任务来读写HTable中的数据。
 

实例分析
我们还是以最初的blog例子来进行示例分析，业务需求是这样：找到具有相同兴趣的人，我们简单定义为如果author之间article的tag相同，则认为两者有相同兴趣，将分析结果保存到HBase。除了上面介绍的blog表外，我们新增一张表tag_friend，RowKey为tag，Value为authors,大概就下面这样。

 
我们省略了一些跟分析无关的Column数据，上面的数据按前面描述的业务需求经过MapReduce分析，应该得到下面的结果
 

实际的运算过程分析如下 
代码实现
有了上面的分析，代码实现就比较简单了。只需以下几步
定义Mapper类继承TableMapper，map的输入输出KV跟上面的分析一致。

1. public static class Mapper extends TableMapper <ImmutableBytesWritable, ImmutableBytesWritable> {
   

2. public Mapper() {}
   

3. @Override
   

4. public void map(ImmutableBytesWritable row, Result values,Context context) throws IOException {
   

5. ImmutableBytesWritable value = null;
   

6. String[] tags = null;
   

7. for (KeyValue kv : values.list()) {
   

8. if ("author".equals(Bytes.toString(kv.getFamily()))
   

9. && "nickname".equals(Bytes.toString(kv.getQualifier()))) {
   

10. value = new ImmutableBytesWritable(kv.getValue());
    

11. }
    

12. if ("article".equals(Bytes.toString(kv.getFamily()))
    

13. && "tags".equals(Bytes.toString(kv.getQualifier()))) {
    

14. tags = Bytes.toString(kv.getValue()).split(",");
    

15. }
    

16. }
    

17. for (int i = 0; i < tags.length; i++) {
    

18. ImmutableBytesWritable key = new ImmutableBytesWritable(
    

19. Bytes.toBytes(tags[i].toLowerCase()));
    

20. try {
    

21. context.write(key,value);
    

22. } catch (InterruptedException e) {
    

23. throw new IOException(e);
    

24. }
    

25. }
    

26. }
    

27. }

复制代码

定义Reducer类继承TableReducer，reduce的输入输出KV跟上面分析的一致。

1. public static class Reducer extends TableReducer <ImmutableBytesWritable, ImmutableBytesWritable, ImmutableBytesWritable> {
   

2. @Override
   

3. public void reduce(ImmutableBytesWritable key,Iterable values,
   

4. Context context) throws IOException, InterruptedException {
   

5. String friends="";
   

6. for (ImmutableBytesWritable val : values) {
   

7. friends += (friends.length()>0?",":"")+Bytes.toString(val.get());
   

8. }
   

9. Put put = new Put(key.get());
   

10. put.add(Bytes.toBytes("person"), Bytes.toBytes("nicknames"),
    

11. Bytes.toBytes(friends));
    

12. context.write(key, put);
    

13. }
    

14. }

复制代码

在提交作业时设置inputFormat为TableInputFormat,设置outputFormat为TableOutputFormat，可以借助TableMapReduceUtil类来简化编码。

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

2. Configuration conf = new Configuration();
   

3. conf = HBaseConfiguration.create(conf);
   

4. Job job = new Job(conf, "HBase_FindFriend");
   

5. job.setJarByClass(FindFriend.class);
   

6. Scan scan = new Scan();
   

7. scan.addColumn(Bytes.toBytes("author"),Bytes.toBytes("nickname"));
   

8. scan.addColumn(Bytes.toBytes("article"),Bytes.toBytes("tags"));
   

9. TableMapReduceUtil.initTableMapperJob("blog", scan,FindFriend.Mapper.class,
   

10. ImmutableBytesWritable.class, ImmutableBytesWritable.class, job);
    

11. TableMapReduceUtil.initTableReducerJob("tag_friend",FindFriend.Reducer.class, job);
    

12. System.exit(job.waitForCompletion(true) ? 0 : 1);
    

13. }

复制代码

小结
本文通过实例分析演示了使用MapReduce分析HBase的数据，需要注意的这只是一种常规的方式（分析表中的数据存到另外的表中），实际上不局限于此，不过其他方式跟此类似。如果你进行到这里，你肯定想要马上运行它看看结果，