图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › 实操演练› 用MR(MapReduce)查询hbase数据－用到TableMapper和Scan
分享

用MR(MapReduce)查询hbase数据－用到TableMapper和Scan

howtodown 2014-8-23 18:40:14 发表于 实操演练 [显示全部楼层] 回帖奖励 阅读模式 关闭右栏 1 20848
问题导读：
1.下面例子是如何实现filter的
2.filter的过后的数据，又进行startRow, stopRow相结合查询？






首先，可以设置scan的startRow, stopRow, filter等属性。

于是两种方案：
1.设置scan的filter，然后执行mapper，再reducer成一份结果
2.不用filter过滤，将filter做的事传给mapper做

进行了测试，前者在执行较少量scan记录的时候效率较后者高，但是执行的scan数量多了，便容易导致超时无返回而退出的情况。而为了实现后者，学会了如何向mapper任务中传递参数，走了一点弯路。
最后的一点思考是，用后者效率仍然不高，即便可用前者时效率也不高，因为默认的tablemapper是将对一个region的scan任务放在了一个mapper里，而我一个region有2G多，而我查的数据只占七八个region。于是，想能不能不以region为单位算做mapper，如果不能改，那只有用MR直接操作HBase底层HDFS文件了，这个，…，待研究。
上代码（为了保密，将表名啊，列名列族名啊都改了一下，有改漏的，大家当做没看见啊，另：主要供大家参考下方法，即用mr来查询海量hbase数据，还有如何向mapper传参数）：

  1. package mapreduce.hbase;

  2. 

  3. import java.io.IOException;

  4. 

  5. import mapreduce.HDFS_File;

  6. 

  7. import org.apache.commons.logging.Log;

  8. import org.apache.commons.logging.LogFactory;

  9. import org.apache.hadoop.conf.Configuration;

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

  11. import org.apache.hadoop.hbase.HBaseConfiguration;

  12. import org.apache.hadoop.hbase.client.Result;

  13. import org.apache.hadoop.hbase.client.Scan;

  14. import org.apache.hadoop.hbase.filter.Filter;

  15. import org.apache.hadoop.hbase.filter.FilterList;

  16. import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;

  17. import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;

  18. import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

  19. import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;

  20. import org.apache.hadoop.hbase.mapreduce.TableMapper;

  21. import org.apache.hadoop.hbase.util.Bytes;

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

  23. import org.apache.hadoop.mapreduce.Job;

  24. import org.apache.hadoop.mapreduce.Mapper.Context;

  25. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

  26. 

  27. /**

  28.  * 用MR对HBase进行查找，给出Scan的条件诸如startkey endkey;以及filters用来过滤掉不符合条件的记录 LicenseTable

  29.  * 的 RowKey 201101010000000095\xE5\xAE\x81WDTLBZ

  30.  * 

  31.  * @author Wallace

  32.  * 

  33.  */

  34. @SuppressWarnings("unused")

  35. public class MRSearchAuto {

  36.         private static final Log LOG = LogFactory.getLog(MRSearchAuto.class);

  37. 

  38.         private static String TABLE_NAME = "tablename";

  39.         private static byte[] FAMILY_NAME = Bytes.toBytes("cfname");

  40.         private static byte[][] QUALIFIER_NAME = { Bytes.toBytes("col1"),Bytes.toBytes("col2"), Bytes.toBytes("col3") };

  41. 

  42.         public static class SearchMapper extends

  43.                         TableMapper<ImmutableBytesWritable, Text> {

  44.                 private int numOfFilter = 0;

  45. 

  46.                 private Text word = new Text();

  47.                 String[] strConditionStrings = new String[]{"","",""}/* { "新C87310", "10", "2" } */;

  48. 

  49.                 /*

  50.                  * private void init(Configuration conf) throws IOException,

  51.                  * InterruptedException { strConditionStrings[0] =

  52.                  * conf.get("search.license").trim(); strConditionStrings[1] =

  53.                  * conf.get("search.carColor").trim(); strConditionStrings[2] =

  54.                  * conf.get("search.direction").trim(); LOG.info("license: " +

  55.                  * strConditionStrings[0]); }

  56.                  */

  57.                 protected void setup(Context context) throws IOException,

  58.                                 InterruptedException {

  59.                         strConditionStrings[0] = context.getConfiguration().get("search.license").trim();

  60.                         strConditionStrings[1] = context.getConfiguration().get("search.color").trim();

  61.                         strConditionStrings[2] = context.getConfiguration().get("search.direction").trim();

  62.                 }

  63. 

  64.                 protected void map(ImmutableBytesWritable key, Result value,

  65.                                 Context context) throws InterruptedException, IOException {

  66.                         String string = "";

  67.                         String tempString;

  68. 

  69.                         /**/

  70.                         for (int i = 0; i < 1; i++) {

  71.                                 // /在此map里进行filter的功能

  72.                                 tempString = Text.decode(value.getValue(FAMILY_NAME,QUALIFIER_NAME[i]));

  73.                                 if (tempString.equals(/* strConditionStrings[i] */"新C87310")) {

  74.                                         LOG.info("新C87310. conf: " + strConditionStrings[0]);

  75.                                         if (tempString.equals(strConditionStrings[i])) {

  76.                                                 string = string + tempString + " ";

  77.                                         } else {

  78.                                                 return;

  79.                                         }

  80.                                 }

  81. 

  82.                                 else {

  83.                                         return;

  84.                                 }

  85.                         }

  86. 

  87.                         word.set(string);

  88.                         context.write(null, word);

  89.                 }

  90.         }

  91. 

  92.         public void searchHBase(int numOfDays) throws IOException,

  93.                         InterruptedException, ClassNotFoundException {

  94.                 long startTime;

  95.                 long endTime;

  96. 

  97.                 Configuration conf = HBaseConfiguration.create();

  98.                 conf.set("hbase.zookeeper.quorum", "node2,node3,node4");

  99.                 conf.set("fs.default.name", "hdfs://node1");

  100.                 conf.set("mapred.job.tracker", "node1:54311");

  101.                 /*

  102.                  * 传递参数给map

  103.                  */

  104.                 conf.set("search.license", "新C87310");

  105.                 conf.set("search.color", "10");

  106.                 conf.set("search.direction", "2");

  107. 

  108.                 Job job = new Job(conf, "MRSearchHBase");

  109.                 System.out.println("search.license: " + conf.get("search.license"));

  110.                 job.setNumReduceTasks(0);

  111.                 job.setJarByClass(MRSearchAuto.class);

  112.                 Scan scan = new Scan();

  113.                 scan.addFamily(FAMILY_NAME);

  114.                 byte[] startRow = Bytes.toBytes("2011010100000");

  115.                 byte[] stopRow;

  116.                 switch (numOfDays) {

  117.                 case 1:

  118.                         stopRow = Bytes.toBytes("2011010200000");

  119.                         break;

  120.                 case 10:

  121.                         stopRow = Bytes.toBytes("2011011100000");

  122.                         break;

  123.                 case 30:

  124.                         stopRow = Bytes.toBytes("2011020100000");

  125.                         break;

  126.                 case 365:

  127.                         stopRow = Bytes.toBytes("2012010100000");

  128.                         break;

  129.                 default:

  130.                         stopRow = Bytes.toBytes("2011010101000");

  131.                 }

  132.                 // 设置开始和结束key

  133.                 scan.setStartRow(startRow);

  134.                 scan.setStopRow(stopRow);

  135. 

  136.                 TableMapReduceUtil.initTableMapperJob(TABLE_NAME, scan,

  137.                                 SearchMapper.class, ImmutableBytesWritable.class, Text.class,

  138.                                 job);

  139.                 Path outPath = new Path("searchresult");

  140.                 HDFS_File file = new HDFS_File();

  141.                 file.DelFile(conf, outPath.getName(), true); // 若已存在，则先删除

  142.                 FileOutputFormat.setOutputPath(job, outPath);// 输出结果

  143. 

  144.                 startTime = System.currentTimeMillis();

  145.                 job.waitForCompletion(true);

  146.                 endTime = System.currentTimeMillis();

  147.                 System.out.println("Time used: " + (endTime - startTime));

  148.                 System.out.println("startRow:" + Text.decode(startRow));

  149.                 System.out.println("stopRow: " + Text.decode(stopRow));

  150.         }

  151. 

  152.         public static void main(String args[]) throws IOException,

  153.                         InterruptedException, ClassNotFoundException {

  154.                 MRSearchAuto mrSearchAuto = new MRSearchAuto();

  155.                 int numOfDays = 1;

  156.                 if (args.length == 1)

  157.                         numOfDays = Integer.valueOf(args[0]);

  158.                 System.out.println("Num of days: " + numOfDays);

  159.                 mrSearchAuto.searchHBase(numOfDays);

  160.         }

  161. }

复制代码

开始时，我是在外面conf.set了传入的参数，而在mapper的init(Configuration)里get参数并赋给mapper对象。
将参数传给map运行时结果不对
for (int i = 0; i < 1; i++) {
    // /在此map里进行filter的功能
    tempString = Text.decode(value.getValue(FAMILY_NAME,
      QUALIFIER_NAME));
    if (tempString.equals(/*strConditionStrings*/"新C87310"))
     string = string + tempString + " ";
    else {
     return;
    }
   }
如果用下面的mapper的init获取conf传来的参数，然后在上面map函数里进行调用，结果便不对了。
直接指定值时和参数传过来相同的值时，其output的结果分别为1条和0条。

  private void init(Configuration conf) throws IOException,
    InterruptedException {
   strConditionStrings[0] = conf.get("search.licenseNumber").trim();
   strConditionStrings[1] = conf.get("search.carColor").trim();
   strConditionStrings[2] = conf.get("search.direction").trim();
  }
加了个日志写
private static final Log LOG = LogFactory.getLog(MRSearchAuto.class);
init()函数里：
LOG.info("license: " + strConditionStrings[0]);
map里
if (tempString.equals(/* strConditionStrings */"新C87310")) {
  LOG.info("新C87310. conf: " + strConditionStrings[0]);
然后在网页 namenode:50030上看任务，最终定位到哪台机器执行了那个map，然后看日志
mapreduce.hbase.TestMRHBase: 新C87310. conf: null
在conf.set之后我也写了下，那时正常，但是在map里却是null了，而在map类的init函数打印的却没有打印。
因此，问题应该是：
map类的init()函数没有执行到！
于是init()的获取conf中参数值并赋给map里变量的操作便未执行，同时打印日志也未执行。
OK！看怎么解决
放在setup里获取
  protected void setup(Context context) throws IOException,
    InterruptedException {
  // strConditionStrings[0] = context.getConfiguration().get("search.license").trim();
  // strConditionStrings[1] = context.getConfiguration().get("search.color").trim();
  // strConditionStrings[2] = context.getConfiguration().get("search.direction").trim();
  }
报错
12/01/12 11:21:56 INFO mapred.JobClient:  map 0% reduce 0%
12/01/12 11:22:03 INFO mapred.JobClient: Task Id : attempt_201201100941_0071_m_000000_0, Status : FAILED
java.lang.NullPointerException
at mapreduce.hbase.MRSearchAuto$SearchMapper.setup(MRSearchAuto.java:66)
at org.apache.hadoop.mapreduce.Mapper.run(Mapper.java:142)
at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:656)
at org.apache.hadoop.mapred.MapTask.run(MapTask.java:325)
at org.apache.hadoop.mapred.Child$4.run(Child.java:270)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:396)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1127)
at org.apache.hadoop.mapred.Child.main(Child.java:264)
attempt_201201100941_0071_m_000000_0: log4j:WARN No appenders could be found for logger (org.apache.hadoop.hdfs.DFSClient).
attempt_201201100941_0071_m_000000_0: log4j:WARN Please initialize the log4j system properly.
12/01/12 11:22:09 INFO mapred.JobClient: Task Id : attempt_201201100941_0071_m_000000_1, Status : FAILED
java.lang.NullPointerException
at mapreduce.hbase.MRSearchAuto$SearchMapper.setup(MRSearchAuto.java:66)
at org.apache.hadoop.mapreduce.Mapper.run(Mapper.java:142)
at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:656)
at org.apache.hadoop.mapred.MapTask.run(MapTask.java:325)
at org.apache.hadoop.mapred.Child$4.run(Child.java:270)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:396)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1127)
at org.apache.hadoop.mapred.Child.main(Child.java:264)
然后将setup里的东西注释掉，无错，错误应该在context上，进一步确认，在里面不用context，直接赋值，有结果，好！
说明是context的事了，NullPointerException,应该是context.getConfiguration().get("search.license")这些中有一个是null的。
突然想起来，改了下get时候的属性，而set时候没改，于是不对应，于是context.getConfiguration().get("search.color")及下面的一项都是null，null.trim()报的异常。
  conf.set("search.license", "新C87310");
  conf.set("search.color", "10");
  conf.set("search.direction", "2");
修改后，问题解决。
实现了向map中传参数









关注公众号，获取大数据、人工智能20套、区块链资源5阶段等资源，随时更新，获取最新技术资源
回复

使用道具 举报

自定义广告语

已有(1)人评论

跳转到指定楼层
正序浏览
lyman825726 发表于 2014-8-24 12:15:40
大数据学习的好论坛，感谢
回复

使用道具 举报

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

本版积分规则

发表新帖
howtodown

版主

关注

  • 666

    主题

  • 1249

    帖子

  • 322

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

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

Copyright © 2001-2024 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

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