HBase分页

Posted 2021-07-10 lijialexiao

HBase 分页

前言

最近接到了一个HBase查询数据分页的需求，了解HBase的小伙伴都知道，HBase做分页想做到与mysql/Oracle那样灵活还是有些困难的，常见的分页所包含的功能一般都有：上一页、下一页、页码跳转、首页、尾页。上一页、下一页、首页这三个HBase实现起来虽然有些麻烦但也还不是不能实现，但是页码跳转与跳转到尾页这个对于HBase来说确实有些尴尬了，做大数据方面也都知道我们做HBase查询一般都是使用Scan配合Filter通过对Rowkey过滤检索的方式实现，当然也有使用Phoenix编写Sql查询的。无论使用哪一种想如同MySql那样通过页码跳转到某一页还是很尴尬的，反正我是没有思路怎么做，当然要是说做内存分页不行吗，可以，非常可以，不仅可以还很快很简单，前提是内存够用吗？？？行了扯了这么多就是想吐槽一下我们不是做线下分析的吗？？？的为何会想到做分页？？？作为小兵我也很无奈啊。。。

分页思路

1.准备两个维护当前页第一条数据rowkey(该次查询的起始rowkey)与当前页最后一条数据rowkey的缓存区
2.查询时从缓存区中先查找起始rowkey,获取到直接返回作为该次的起始rowkey查询条件，未获取到则在另一个缓存区中查找最后一条数据rowkey
（注意：这里的rowkey都是上一次查询记录在缓存区的，后面会说）最后一条数据的rowkey同样未获取到则表示该查询未初始查询，直接根据
传入的rowkey作为初始rowkey查询（有人说我想对某一列的值进行检索，不想通过rowkey检索怎么办，那抱歉了，先撇开这样是否符合HBase存储结构设计初衷不说，这样能不能查的动都还不好说，绝对卡的你分分钟自闭）
3.查询出数据后，判断是向上翻页还是向下翻页，向上分页则过滤删除掉最后一条数据，向下翻页过滤删除掉第一条数据（因每次翻页都是通过上一次
查询时记录的rowkey做初始rowkeya查询，所以查询出的数据中肯定页包含了该rowkey数据不处理掉的话，响应给用户的话不就给用户一种重复数据的感觉了吗）
4.记录本次查询数据的第一条数据的rowkey(或者说是本次查询的初始rowkey)，与本次查询结果最后一条数据的rowkey并清除记录在缓存区中上一页
的最后一条数据的rowkey

以上就是分页的大致思路，下面粘一下核心代码，也许看着更直观一些

分页实现代码

查询参数

/**
 * @Author Lijl
 * @ClassName HbaseModel
 * @Description 查询参数
 * @Date 2021/5/25 15:29
 * @Version 1.0
 */
public class HbaseModel {
    private String rowKey;
    private String tabName;
    private String methodNameType;
    private Configuration conf;
    private String page;
    private String pageSize;
    private String sessionId;
    private boolean reBol;

    ...省略GET\\SET

    public long getPageSize() {
        if (pageSize!=null && !"".equals(pageSize)){
            return Long.parseLong(pageSize)+1;
        }
        return 0;
    }

    private static Cache<String,String> startRowKeyMap = Cache.newHardMemoryCache(1000000,60*30);
    private static Cache<String,String> lastRwoKeyMap = Cache.newHardMemoryCache(1000000,60*30);
    
    /**
     * @Author lijiale
     * @MethodName getQueryStartRowKey
     * @Description 获取查询开始Rowkey
     * @Date 17:11 2021/5/28
     * @Version 1.0
     * @param 
     * @return: java.lang.String
    **/
    public synchronized String getQueryStartRowKey(){
        String page = this.page;
        if (page!=null){
            int pageInt = Integer.parseInt(page);
            String key = this.sessionId+this.rowKey+"a_"+pageInt;
            String s = startRowKeyMap.get(key);
            if (s!=null){
                String s1 = this.sessionId+this.rowKey + "a_" + (pageInt + 1);
                if (startRowKeyMap.get(s1)!=null){
                    startRowKeyMap.remove(s1);
                }
                this.reBol = false;
                //-----------返回上一页-----------
                return s;
            }
            String key1 = this.sessionId+this.rowKey + "a_" + pageInt;
            String s1 = lastRwoKeyMap.get(key1);
            if (s1!=null){
                //-----------翻到下一页-------------
                this.reBol = true;
                return s1;
            }
        }
        return null;
    }

    public synchronized void setStartRowKey(String rowKey,String startRowKey){
        String key = this.sessionId+rowKey + "a_" +this.page;
        startRowKeyMap.remove(key);
        startRowKeyMap.put(key,startRowKey);
    }

    public synchronized void setLastRowKey(String rowKey,String lastRowKey){
        String s = this.sessionId+rowKey + "a_" + this.page;
        lastRwoKeyMap.remove(s);
        String key = this.sessionId+rowKey + "a_" + (Integer.parseInt(this.page)+1);
        lastRwoKeyMap.remove(key);
        lastRwoKeyMap.put(key, lastRowKey);
    }

pageSize加1的意义，在上面提到了因rowkey查询数来的重复数据，加1的意义则是为了保证删除一条后响应给用户/前端的数据
与他们分页要展示的数据数量一致

从缓存中查找rowkey

    ...省略
        long pageSize = hbaseModel.getPageSize();
        int dataCount = hBaseServiceUtil.getTableDataCount(tableName, "", startRowKey, endRowKey);
        Map<String,Object> dataMap = new HashMap<>(2);
        dataMap.put("count",dataCount);
        List<Object> list = new ArrayList<>();
        if (dataCount>0){
            if (pageSize>0){
                String lastRowKey = hbaseModel.getQueryStartRowKey();
                if (lastRowKey!=null && !"".equals(lastRowKey)){
                    startRowKey = lastRowKey;
                }
            }
            List<Map<String,Object>> mapList = hBaseServiceUtil.getNumRegexRow(tableName, startRowKey,endRowKey,pageSize);
    ...省略

这是上面提到提到的第二步，获取本次查询的初始rowkey查询条件

查询Filter参数设置

/**
     * @Author lijiale
     * @MethodName getNumRegexRow
     * @Description 时间区间查询青海HBase数据
     * @Date 9:46 2021/3/12
     * @Version 1.0
     * @param tableName
     * @param startRowKey
     * @param endRowKey
     * @return: java.util.List<java.util.Map<java.lang.String,java.lang.Object>>
    **/
    public List<Map<String,Object>> getNumRegexRow(String tableName,String startRowKey,String endRowKey,long pageSize){
        Table table = getTable(tableName);
        logger.info("组装Rowkey过滤器，startRowKey:{},endRowKey:{},表名:{}",startRowKey,endRowKey,table.getName());
        FilterList fl = new FilterList(FilterList.Operator.MUST_PASS_ALL);
        RegexStringComparator rc = new RegexStringComparator("[^\\\\\\\\\\\\/\\\\^]");
        RowFilter rf = new RowFilter(CompareOperator.EQUAL,rc);
        fl.addFilter(rf);
        if (pageSize>0){
            fl.addFilter(new PageFilter(pageSize));
        }
        Scan scan = new Scan();
        //设置取值范围
        scan.withStartRow(startRowKey.getBytes()).withStopRow(endRowKey.getBytes());
        scan.setFilter(fl);//为查询设置过滤器的list
        scan.setCaching(5000);
        if (pageSize==0){
            scan.setBatch(100);
        }
        return queryData(table,scan);
    }

要注意的是PageFilter这个是设置本次查询输出多少条类似limit，且设置了PageFilter后便不可再
设置批处理，代码中也能看出来做了一个判断

删除多余重复数据

    /**
     * @Author lijiale
     * @MethodName deleteARowOfData
     * @Description 删除与上一页重复的一条
     * @Date 17:26 2021/5/28
     * @Version 1.0
     * @param mapList
     * @param hbaseModel
     * @return: void
    **/
    protected void deleteARowOfData(List<Map<String, Object>> mapList, HbaseModel hbaseModel) {
        if (hbaseModel.getPageSize()>0){
            boolean reBol = hbaseModel.getReBol();
            if (reBol){
                mapList.remove(mapList.size()-1);
            }else{
                mapList.remove(0);
            }
        }
    }

记录rowkey

                super.deleteARowOfData(mapList,hbaseModel);
                if (mapList.size()>0){
                    int count = 0;
                    for (Map<String, Object> map : mapList) {
                        Object obj = super.invoteSetter(ToMEvntNlUserFlux.class, map);
                        if (obj!=null){
                            list.add(obj);
                            ++ count;
                        }
                    }
                    if (pageSize>0){
                        if (mapList.size()-1 == count){
                            Object rk = mapList.get(count-1).get("rowKey");
                            if (rk!=null){
                                hbaseModel.setLastRowKey(rowKey,rk+"");
                            }
                        }
                    }
                    if (pageSize>0){
                        hbaseModel.setStartRowKey(rowKey,startRowKey);
                    }
                }
            }
        }
        dataMap.put("element",list);
        return BaseResult.ok("成功",dataMap);

记录本次查询结果第一条数据的rowkey与最后一条数据的rowkey