分享

hbase分页应用场景及分页思路与代码实现

pig2 2014-3-1 23:35:11 发表于 推荐型 [显示全部楼层] 回帖奖励 阅读模式 关闭右栏 13 63837
本帖最后由 pig2 于 2014-3-2 00:52 编辑

可以带着下面问题来阅读
1.hbasef分页什么情况下比较有用?
2.hbase分页的思路是什么?
3.hbase分页代码与sql分页代码有什么区别和联系?

一、hbase分页应用场景:


一、应用场景
hbase到底需不需要分页,hbase的数据量肯定不少,肯定是需要分页的。很多人认为数量量如此大,怎么会展示。这个从客户角度来讲,我们做的系统,不可能会给机器看的。这里面我们可以对其进行统计分析,这样利于我们决策。
比如我们:平台中有一个场景是要做用户历史订单数据的查询,并且需要支持分页。这里只是举了一个场景,后面大家可以根据自己的经验。下面给大家讲讲分页的思路。





二、hbase分页思路:

hbase通过scan来扫描表,通过startKey,stopKey来确定范围,hbase官方提供了一个PageFilter来支持一次scan可以返回多少条数据即每页的行数。假如一页是10条,这样是第一页还好,但是第二页呢,如果不改变PageFilter的pageSize,那返回的还是第一页的数据,如果改变pageSize为20,则返回了第一页10多余的数据,在客户端要过滤掉,性能不好。那怎么办呢,方法就是在查询下一页时,指定下一页的startKey,这样PageFilter每次就不会返回多余的记录,stopKey可以不用变,那现在问题是,怎么得到下一页的startKey(即下一页第一行的rowkey)呢?,有两种方法来取每一页的startKey

1.  上一页的最后一行记录的rowkey作为下一页的startKey。
2.   在每次scan时多取一条记录,即把下一页第一条行页取出来,把该行的rowkey做为下一页的startKey。

这两种方法,都要注意,hbase scan时是包含startKey的,如果是采用第一种,则要在记录多取一条,排除第一条。第二种页是多取一条,但是排除最后一条,用来做下一页的startKey。还有需要注意的是在计算是否有下一页时,可以根据返回的条数来判断。

startKey怎么取没有问题了。但是怎么存储呢,有同学可能会想到存到session,但是如果你的服务是rest api型的,就没有session的概念了。那还有两种选择:
1. 是存到客户端,让客户端每次请求时把startKey再传回来,这样需要依赖客户端,如果客户端是远程,或者是开放平台的情况下,可能不合适。
2. 存在服务端,存在服务端需要注意并发访问的情况。比如scan同一个表,一个访问第2页,一个访问第3页,服务端就需要对每一个table的scan 存每一页的startKey,需要为同一个查询条件包含pageSize,因为pageSize不一样,startKey也会不一样,
在服务crash情况下,从起后都从第一页开始。

我自己是采用第二种方案,存在服务端.




----------------------------------------------------------------------------------------------------------------------------------------------------
三、代码实现
  1. import java.io.IOException;
  2. import java.util.LinkedHashMap;
  3. import java.util.LinkedList;
  4. import java.util.List;
  5. import java.util.Map;
  6. import org.apache.commons.lang.StringUtils;
  7. import org.apache.hadoop.conf.Configuration;
  8. import org.apache.hadoop.hbase.HBaseConfiguration;
  9. import org.apache.hadoop.hbase.client.Get;
  10. import org.apache.hadoop.hbase.client.HTableInterface;
  11. import org.apache.hadoop.hbase.client.HTablePool;
  12. import org.apache.hadoop.hbase.client.Result;
  13. import org.apache.hadoop.hbase.client.ResultScanner;
  14. import org.apache.hadoop.hbase.client.Scan;
  15. import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
  16. import org.apache.hadoop.hbase.filter.Filter;
  17. import org.apache.hadoop.hbase.filter.FilterList;
  18. import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
  19. import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
  20. import org.apache.hadoop.hbase.util.Bytes;
  21. public class HBaseUtils {
  22.         private static Configuration config = null;
  23.         private static HTablePool tp = null;
  24.         static {
  25.                 // 加载集群配置
  26.                 config = HBaseConfiguration.create();
  27.                 config.set("hbase.zookeeper.quorum", "xx.xx.xx");
  28.                 config.set("hbase.zookeeper.property.clientPort", "2181");
  29.                 // 创建表池(可伟略提高查询性能,具体说明请百度或官方API)
  30.                 tp = new HTablePool(config, 10);
  31.         }
  32.         /*
  33.          * 获取hbase的表
  34.          */
  35.         public static HTableInterface getTable(String tableName) {
  36.                 if (StringUtils.isEmpty(tableName))
  37.                         return null;
  38.                 return tp.getTable(getBytes(tableName));
  39.         }
  40.         /* 转换byte数组 */
  41.         public static byte[] getBytes(String str) {
  42.                 if (str == null)
  43.                         str = "";
  44.                 return Bytes.toBytes(str);
  45.         }
  46.         /**
  47.          * 查询数据
  48.          * @param tableKey 表标识
  49.          * @param queryKey 查询标识
  50.          * @param startRow 开始行
  51.          * @param paramsMap 参数集合
  52.          * @return 结果集
  53.          */
  54.         public static TBData getDataMap(String tableName, String startRow,
  55.                         String stopRow, Integer currentPage, Integer pageSize)
  56.                         throws IOException {
  57.                 List<Map<String, String>> mapList = null;
  58.                 mapList = new LinkedList<Map<String, String>>();
  59.                 ResultScanner scanner = null;
  60.                 // 为分页创建的封装类对象,下面有给出具体属性
  61.                 TBData tbData = null;
  62.                 try {
  63.                         // 获取最大返回结果数量
  64.                         if (pageSize == null || pageSize == 0L)
  65.                                 pageSize = 100;
  66.                         if (currentPage == null || currentPage == 0)
  67.                                 currentPage = 1;
  68.                         // 计算起始页和结束页
  69.                         Integer firstPage = (currentPage - 1) * pageSize;
  70.                         Integer endPage = firstPage + pageSize;
  71.                         // 从表池中取出HBASE表对象
  72.                         HTableInterface table = getTable(tableName);
  73.                         // 获取筛选对象
  74.                         Scan scan = getScan(startRow, stopRow);
  75.                         // 给筛选对象放入过滤器(true标识分页,具体方法在下面)
  76.                         scan.setFilter(packageFilters(true));
  77.                         // 缓存1000条数据
  78.                         scan.setCaching(1000);
  79.                         scan.setCacheBlocks(false);
  80.                         scanner = table.getScanner(scan);
  81.                         int i = 0;
  82.                         List<byte[]> rowList = new LinkedList<byte[]>();
  83.                         // 遍历扫描器对象, 并将需要查询出来的数据row key取出
  84.                         for (Result result : scanner) {
  85.                                 String row = toStr(result.getRow());
  86.                                 if (i >= firstPage && i < endPage) {
  87.                                         rowList.add(getBytes(row));
  88.                                 }
  89.                                 i++;
  90.                         }
  91.                         // 获取取出的row key的GET对象
  92.                         List<Get> getList = getList(rowList);
  93.                         Result[] results = table.get(getList);
  94.                         // 遍历结果
  95.                         for (Result result : results) {
  96.                                 Map<byte[], byte[]> fmap = packFamilyMap(result);
  97.                                 Map<String, String> rmap = packRowMap(fmap);
  98.                                 mapList.add(rmap);
  99.                         }
  100.                         // 封装分页对象
  101.                         tbData = new TBData();
  102.                         tbData.setCurrentPage(currentPage);
  103.                         tbData.setPageSize(pageSize);
  104.                         tbData.setTotalCount(i);
  105.                         tbData.setTotalPage(getTotalPage(pageSize, i));
  106.                         tbData.setResultList(mapList);
  107.                 } catch (IOException e) {
  108.                         e.printStackTrace();
  109.                 } finally {
  110.                         closeScanner(scanner);
  111.                 }
  112.                 return tbData;
  113.         }
  114.         private static int getTotalPage(int pageSize, int totalCount) {
  115.                 int n = totalCount / pageSize;
  116.                 if (totalCount % pageSize == 0) {
  117.                         return n;
  118.                 } else {
  119.                         return ((int) n) + 1;
  120.                 }
  121.         }
  122.         // 获取扫描器对象
  123.         private static Scan getScan(String startRow, String stopRow) {
  124.                 Scan scan = new Scan();
  125.                 scan.setStartRow(getBytes(startRow));
  126.                 scan.setStopRow(getBytes(stopRow));
  127.                 return scan;
  128.         }
  129.         /**
  130.          * 封装查询条件
  131.          */
  132.         private static FilterList packageFilters(boolean isPage) {
  133.                 FilterList filterList = null;
  134.                 // MUST_PASS_ALL(条件 AND) MUST_PASS_ONE(条件OR)
  135.                 filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
  136.                 Filter filter1 = null;
  137.                 Filter filter2 = null;
  138.                 filter1 = newFilter(getBytes("family1"), getBytes("column1"),
  139.                                 CompareOp.EQUAL, getBytes("condition1"));
  140.                 filter2 = newFilter(getBytes("family2"), getBytes("column1"),
  141.                                 CompareOp.LESS, getBytes("condition2"));
  142.                 filterList.addFilter(filter1);
  143.                 filterList.addFilter(filter2);
  144.                 if (isPage) {
  145.                         filterList.addFilter(new FirstKeyOnlyFilter());
  146.                 }
  147.                 return filterList;
  148.         }
  149.         private static Filter newFilter(byte[] f, byte[] c, CompareOp op, byte[] v) {
  150.                 return new SingleColumnValueFilter(f, c, op, v);
  151.         }
  152.         private static void closeScanner(ResultScanner scanner) {
  153.                 if (scanner != null)
  154.                         scanner.close();
  155.         }
  156.         /**
  157.          * 封装每行数据
  158.          */
  159.         private static Map<String, String> packRowMap(Map<byte[], byte[]> dataMap) {
  160.                 Map<String, String> map = new LinkedHashMap<String, String>();
  161.                 for (byte[] key : dataMap.keySet()) {
  162.                         byte[] value = dataMap.get(key);
  163.                         map.put(toStr(key), toStr(value));
  164.                 }
  165.                 return map;
  166.         }
  167.         /* 根据ROW KEY集合获取GET对象集合 */
  168.         private static List<Get> getList(List<byte[]> rowList) {
  169.                 List<Get> list = new LinkedList<Get>();
  170.                 for (byte[] row : rowList) {
  171.                         Get get = new Get(row);
  172.                         get.addColumn(getBytes("family1"), getBytes("column1"));
  173.                         get.addColumn(getBytes("family1"), getBytes("column2"));
  174.                         get.addColumn(getBytes("family2"), getBytes("column1"));
  175.                         list.add(get);
  176.                 }
  177.                 return list;
  178.         }
  179.         /**
  180.          * 封装配置的所有字段列族
  181.          */
  182.         private static Map<byte[], byte[]> packFamilyMap(Result result) {
  183.                 Map<byte[], byte[]> dataMap = null;
  184.                 dataMap = new LinkedHashMap<byte[], byte[]>();
  185.                 dataMap.putAll(result.getFamilyMap(getBytes("family1")));
  186.                 dataMap.putAll(result.getFamilyMap(getBytes("family2")));
  187.                 return dataMap;
  188.         }
  189.         private static String toStr(byte[] bt) {
  190.                 return Bytes.toString(bt);
  191.         }
  192.         public static void main(String[] args) throws IOException {
  193.                 // 拿出row key的起始行和结束行
  194.                 // #<0<9<:
  195.                 String startRow = "aaaa#";
  196.                 String stopRow = "aaaa:";
  197.                 int currentPage = 1;
  198.                 int pageSize = 20;
  199.                 // 执行hbase查询
  200.                 getDataMap("table", startRow, stopRow, currentPage, pageSize);
  201.         }
  202. }
  203. class TBData {
  204.         private Integer currentPage;
  205.         private Integer pageSize;
  206.         private Integer totalCount;
  207.         private Integer totalPage;
  208.         private List<Map<String, String>> resultList;
  209.         public Integer getCurrentPage() {
  210.                 return currentPage;
  211.         }
  212.         public void setCurrentPage(Integer currentPage) {
  213.                 this.currentPage = currentPage;
  214.         }
  215.         public Integer getPageSize() {
  216.                 return pageSize;
  217.         }
  218.         public void setPageSize(Integer pageSize) {
  219.                 this.pageSize = pageSize;
  220.         }
  221.         public Integer getTotalCount() {
  222.                 return totalCount;
  223.         }
  224.         public void setTotalCount(Integer totalCount) {
  225.                 this.totalCount = totalCount;
  226.         }
  227.         public Integer getTotalPage() {
  228.                 return totalPage;
  229.         }
  230.         public void setTotalPage(Integer totalPage) {
  231.                 this.totalPage = totalPage;
  232.         }
  233.         public List<Map<String, String>> getResultList() {
  234.                 return resultList;
  235.         }
  236.         public void setResultList(List<Map<String, String>> resultList) {
  237.                 this.resultList = resultList;
  238.         }
  239. }
复制代码

已有(13)人评论

跳转到指定楼层
sunshine_junge 发表于 2014-7-27 15:32:48
回复

使用道具 举报

ascentzhen 发表于 2014-8-9 19:20:10
不错,学习了,谢谢分享
回复

使用道具 举报

XiaoLuo 发表于 2014-10-22 14:29:33
感谢分享,写的很好,启发很大,但是你代码里有些方法过时了向HtablePool被HConnection代替了。
回复

使用道具 举报

thomaschen 发表于 2016-3-9 16:15:01
分页总数及总条数 怎么获取呢
回复

使用道具 举报

雷夫23 发表于 2016-7-1 15:00:43
不太明白,这个程序的最终结果应该什么样 的
回复

使用道具 举报

Rommy.Yang 发表于 2016-7-12 18:53:37
如果查询的时候,需要排序怎么处理的?
回复

使用道具 举报

hx252502115 发表于 2016-7-28 19:04:27
Rommy.Yang 发表于 2016-7-12 18:53
如果查询的时候,需要排序怎么处理的?

已经按照rowkey的字典顺序排好序了,如果需要其他列排序的话结合hive就可以做到了

回复

使用道具 举报

12下一页
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关闭

推荐上一条 /2 下一条