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

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