HBase 使用过滤器进行分页代码记录

1. 介绍

在大数据应用场景中，HBase 常被用于存储和管理海量的结构化或半结构化数据。在实际业务中，用户通常不会一次性读取整张表的数据，而是需要按照一定条件对数据进行查询、展示和逐步加载。由于 HBase 中的数据规模通常较大，如果一次性返回全部结果，不仅会增加网络传输开销，还会加重客户端的内存与处理压力，严重时甚至可能引发应用性能下降或内存溢出。因此，在面向结果集展示或分批处理的场景下，分页查询是一种常见且必要的实现方式。

2. 实现逻辑

以下是使用 HBase 过滤器进行查询分页的逻辑：

1. 首先，我们需要创建一个 Scan 对象，该对象用于描述查询范围，如起始行键（start row）和结束行键（stop row）。
2. 为了实现分页，我们可以使用两个过滤器：PageFilter 和 ColumnPaginationFilter。PageFilter 可以用于指定每页返回的行数，而 ColumnPaginationFilter 可以用于限制每行返回的列数。
3. 在创建 Scan 对象之后，我们需要为其添加过滤器。首先，使用 PageFilter 设置每页的行数，然后使用 ColumnPaginationFilter 设置每行的列数。这两个过滤器可以通过 FilterList 组合在一起，以便同时应用于 Scan 对象。
4. 使用完善的 Scan 对象执行查询操作，获取 ResultScanner。遍历 ResultScanner，将结果保存到一个结果集合中。
5. 分页查询的关键在于跟踪上一次查询的最后一个行键。当用户请求下一页时，我们可以使用这个行键作为新的起始行键，然后重复上述过程来获取下一页的数据。
6. 当 ResultScanner 中没有更多的数据时，表示已经到达查询结果的末尾。此时，应停止查询并返回结果。

通过上述逻辑，我们可以在 HBase 中实现基于过滤器的分页查询。需要注意的是，这种方法在数据量较大时可能会导致性能问题，因为 HBase 需要扫描所有满足条件的行。为了提高性能，可以考虑优化数据模型、行键设计或使用更高效的查询策略。

3. 过滤器介绍

以下给出一种基于 Scan 与过滤器的实验性分页思路，该思路在特定场景下可以实现 HBase 查询结果的分批获取，但更准确地说，它属于基于 rowkey 游标的翻页方案，并不等同于关系型数据库中的传统页码分页。

1. 首先，需要创建 Scan 对象，用于描述本次扫描范围，例如起始行键（startRow）和结束行键（stopRow）。
2. 在分页场景中，通常可以使用 PageFilter 控制本次扫描返回的行数上限。如果业务上还需要限制每一行返回的列数量，可以额外使用 ColumnPaginationFilter；但需要注意，ColumnPaginationFilter 解决的是列分页问题，并不是结果集“按页取行”的核心手段。
3. 创建好 Scan 对象后，可以根据业务需要为其设置过滤器。若同时存在多种过滤条件，可通过 FilterList 将多个过滤器组合后再应用到 Scan 对象上。
4. 使用配置完成的 Scan 执行查询，获取 ResultScanner，再遍历扫描结果，将当前批次数据封装到结果集合中。
5. 这种翻页方式的关键在于记录当前批次最后一条记录对应的 rowkey。当需要获取下一批数据时，可以将该 rowkey 作为下一次扫描的起点，继续执行查询，从而实现类似“下一页”的效果。
6. 当本次扫描结果数量不足预期页大小，或者已经没有更多记录可返回时，通常可以认为已经到达结果集末尾。

通过上述方式，可以在 HBase 中实现一种实验性的分页查询方案。需要说明的是，这种方式本质上更接近“顺序翻页”或“游标翻页”，其效果依赖于 rowkey 的设计与扫描顺序。在数据量较大或并发写入频繁的场景下，还需要额外关注结果重复、遗漏以及扫描性能等问题。因此，在实际项目中，应结合具体业务需求，对 rowkey 设计、过滤条件和查询策略进行进一步优化。

4. 代码示例

4.1 HBase 配置工具类

/**
 * @author hnbian
 * @Createdate 2016-4-18下午4:51:08
 */ 
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;

import com.hnbian.record.log.hadoop.utils.ProP;

/**
 * HBase 配置工具类
 *
 * 功能：
 * 1. 初始化 HBase 配置
 * 2. 统一提供 Configuration 对象
 */
public class HBaseConfig {

    /**
     * 获取 HBase 配置对象
     *
     * @return Configuration
     */
    public static Configuration getConfiguration() {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.rootdir", ProP.getPath("rootDir"));
        conf.set("hbase.zookeeper.quorum", ProP.getPath("zkServer"));
        conf.set("hbase.zookeeper.property.clientPort", ProP.getPath("port"));
        return conf;
    }
}

4.2 HBase 表管理类

/**
 * @author hnbian
 * @Createdate 2016-4-18下午5:17:31
 */ 
import java.io.IOException;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;

/**
 * HBase 表管理类
 *
 * 功能：
 * 1. 创建表
 * 2. 创建列族描述
 */
public class HBaseTableManager {

    private Configuration conf;

    public HBaseTableManager() {
        this.conf = HBaseConfig.getConfiguration();
    }

    /**
     * 创建 HBase 表
     *
     * @param tablename 表名
     * @param cols      列族列表
     * @throws IOException
     */
    public void createTable(String tablename, List<String> cols) throws IOException {
        HBaseAdmin admin = null;
        try {
            admin = new HBaseAdmin(conf);

            if (admin.tableExists(tablename)) {
                throw new IOException("table exists");
            }

            HTableDescriptor tableDesc = new HTableDescriptor(tablename);

            for (String col : cols) {
                HColumnDescriptor colDesc = createColumnDescriptor(col);
                tableDesc.addFamily(colDesc);
            }

            admin.createTable(tableDesc);
        } catch (IOException e) {
            e.printStackTrace();
            throw e;
        } finally {
            if (admin != null) {
                admin.close();
            }
        }
    }

    /**
     * 创建列族描述对象
     *
     * @param col 列族名
     * @return HColumnDescriptor
     */
    private HColumnDescriptor createColumnDescriptor(String col) {
        HColumnDescriptor colDesc = new HColumnDescriptor(col);
        colDesc.setCompressionType(Algorithm.GZ);
        colDesc.setDataBlockEncoding(DataBlockEncoding.DIFF);
        return colDesc;
    }
}

4.3 HBase 数据操作类

/**
 * @author hnbian
 * @Createdate 2016-4-18下午5:36:01
 */ 
import java.io.IOException;
import java.util.List;

import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.coprocessor.AggregationClient;
import org.apache.hadoop.hbase.client.coprocessor.LongColumnInterpreter;
import org.apache.hadoop.hbase.util.Bytes;

import com.hnbian.record.log.hadoop.utils.HBaseConnection;

/**
 * HBase 数据操作类
 *
 * 功能：
 * 1. 批量写入数据
 * 2. 查询记录总数
 */
public class HBaseDataManager {

    /**
     * 批量保存数据
     *
     * @param tablename 表名
     * @param puts      Put 集合
     */
    public void saveData(String tablename, List<Put> puts) {
        HConnection hConn = null;
        HTableInterface table = null;

        try {
            hConn = HBaseConnection.GetHConnection();
            table = hConn.getTable(tablename);

            // 关闭自动提交，提高批量写入效率
            table.setAutoFlush(false);

            // 批量写入
            table.put(puts);

            // 手动提交
            table.flushCommits();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            close(table, hConn);
        }
    }

    /**
     * 获取满足条件的总记录数
     *
     * @param tablename 表名
     * @param nowTime   结束 rowkey
     * @return 记录数
     */
    public int getCount(String tablename, String nowTime) {
        HConnection hConn = null;
        HTableInterface table = null;
        AggregationClient aggr = null;

        try {
            hConn = HBaseConnection.GetHConnection();
            table = hConn.getTable(tablename);
            aggr = HBaseConnection.GetAggregationClient();

            Scan scan = new Scan();

            if (nowTime != null && !"".equals(nowTime)) {
                scan.setStopRow(Bytes.toBytes(nowTime));
            }

            Long rowCount = aggr.rowCount(table.getTableName(), new LongColumnInterpreter(), scan);
            return rowCount.intValue();
        } catch (Throwable e) {
            e.printStackTrace();
        } finally {
            if (aggr != null) {
                try {
                    aggr.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            close(table, hConn);
        }

        return 0;
    }

    /**
     * 关闭资源
     *
     * @param table HBase 表对象
     * @param hConn HBase 连接对象
     */
    private void close(HTableInterface table, HConnection hConn) {
        try {
            if (table != null) {
                table.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        try {
            if (hConn != null) {
                hConn.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

4.4 HBase 分页查询类

/**
 * @author hnbian
 * @Createdate 2016-4-18下午6:31:14
 */ 
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.BinaryPrefixComparator;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import com.hnbian.core.utils.JacksonUtils;
import com.hnbian.record.log.hadoop.utils.HBaseConnection;
import com.hnbian.record.log.hadoop.utils.ProP;
import com.hnbian.record.log.persistences.UserTracks;

/**
 * HBase 分页查询类
 *
 * 功能：
 * 1. 基于 Scan + FilterList + PageFilter 执行分页查询
 * 2. 支持 datetime 和 machine_no 条件过滤
 * 3. 返回 data、pageSize、pageNow、nextStart 等分页信息
 *
 * 说明：
 * 该实现属于基于 rowkey 的顺序翻页，不是严格意义上的页码分页。
 */
public class HBasePageQueryManager {

    /**
     * 对外分页查询入口
     *
     * @param tableName  表名
     * @param nowTime    结束 rowkey
     * @param startRow   开始 rowkey
     * @param pageNow    当前页
     * @param pageSize   每页条数
     * @param datetime   时间过滤条件
     * @param machine_no 设备号过滤条件
     * @return 分页结果
     */
    public Map<String, String> getData(String tableName,
                                       String nowTime,
                                       String startRow,
                                       Integer pageNow,
                                       Integer pageSize,
                                       String datetime,
                                       String machine_no) {
        HConnection hConn = null;
        HTableInterface table = null;

        try {
            hConn = HBaseConnection.GetHConnection();
            table = hConn.getTable(tableName);

            Map<String, String> result = getIndexTableInfo(
                    table, nowTime, startRow, pageNow, pageSize, datetime, machine_no
            );

            System.out.println("getData map.size() => " + result.size());
            return result;
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            close(table, hConn);
        }

        return null;
    }

    /**
     * 核心分页查询方法
     *
     * @param table      HBase 表对象
     * @param nowTime    结束 rowkey
     * @param startRow   起始 rowkey
     * @param pageNow    当前页
     * @param pageSize   每页大小
     * @param datetime   datetime 过滤条件
     * @param machine_no machine_no 过滤条件
     * @return 分页结果
     */
    private Map<String, String> getIndexTableInfo(HTableInterface table,
                                                  String nowTime,
                                                  String startRow,
                                                  Integer pageNow,
                                                  Integer pageSize,
                                                  String datetime,
                                                  String machine_no) {
        Map<String, String> resultMap = new TreeMap<String, String>();
        ResultScanner scanner = null;
        List<UserTracks> list = new ArrayList<UserTracks>();

        try {
            if (pageSize == null || pageSize == 0) {
                pageSize = Integer.parseInt(ProP.getPath("pageSize"));
            }

            if (pageNow == null || pageNow == 0) {
                pageNow = 1;
            }

            // 多查一条，用于判断是否还有下一页
            int querySize = pageSize + 1;

            Scan scan = buildScan(nowTime, startRow, querySize, datetime, machine_no);

            scanner = table.getScanner(scan);

            int rowIndex = 0;
            for (Result result : scanner) {
                rowIndex++;

                String row = Bytes.toString(result.getRow());

                // 第 pageSize + 1 条作为下一页起点
                if (rowIndex == querySize) {
                    resultMap.put("nextStart", row);
                    break;
                }

                UserTracks userTracks = buildUserTracks(result);
                if (userTracks.getAccount_id() != null) {
                    list.add(userTracks);
                }
            }

            resultMap.put("data", JacksonUtils.toJSONString(list));
            resultMap.put("pageSize", String.valueOf(pageSize));
            resultMap.put("pageNow", String.valueOf(pageNow));

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (scanner != null) {
                scanner.close();
            }
        }

        return resultMap;
    }

    /**
     * 构建 Scan 对象
     *
     * @param nowTime    结束 rowkey
     * @param startRow   起始 rowkey
     * @param querySize  查询条数
     * @param datetime   时间过滤条件
     * @param machine_no 设备编号过滤条件
     * @return Scan
     */
    private Scan buildScan(String nowTime,
                           String startRow,
                           int querySize,
                           String datetime,
                           String machine_no) {
        Scan scan = new Scan();

        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);

        // 行分页过滤器
        Filter pageFilter = new PageFilter(querySize);
        filterList.addFilter(pageFilter);

        // rowkey 前缀过滤器
        if (nowTime != null && !"".equals(nowTime)) {
            String[] arr = nowTime.split("-");
            if (arr.length > 0) {
                Filter prefixFilter = new PrefixFilter(Bytes.toBytes(arr[0]));
                filterList.addFilter(prefixFilter);
            }
        }

        // datetime 过滤器
        if (datetime != null && !"".equals(datetime)) {
            Filter datetimeFilter = new SingleColumnValueFilter(
                    Bytes.toBytes("info"),
                    Bytes.toBytes("datetime"),
                    CompareFilter.CompareOp.EQUAL,
                    new BinaryPrefixComparator(Bytes.toBytes(datetime))
            );
            filterList.addFilter(datetimeFilter);
        }

        // machine_no 过滤器
        if (machine_no != null && !"".equals(machine_no)) {
            Filter machineNoFilter = new SingleColumnValueFilter(
                    Bytes.toBytes("info"),
                    Bytes.toBytes("machine_no"),
                    CompareFilter.CompareOp.EQUAL,
                    Bytes.toBytes(machine_no)
            );
            filterList.addFilter(machineNoFilter);
        }

        scan.setFilter(filterList);

        if (startRow != null && !"".equals(startRow)) {
            scan.setStartRow(Bytes.toBytes(startRow));
        }

        if (nowTime != null && !"".equals(nowTime)) {
            scan.setStopRow(Bytes.toBytes(nowTime));
        }

        return scan;
    }

    /**
     * 将一行 Result 转换成 UserTracks 对象
     *
     * @param result 一行查询结果
     * @return UserTracks
     */
    private UserTracks buildUserTracks(Result result) {
        UserTracks ut = new UserTracks();

        for (Cell cell : result.rawCells()) {
            String k = Bytes.toString(CellUtil.cloneQualifier(cell));
            String v = Bytes.toString(CellUtil.cloneValue(cell));

            if ("account_id".equals(k)) {
                ut.setAccount_id(v);
            } else if ("activityName".equals(k)) {
                ut.setActivityName(v);
            } else if ("duration".equals(k)) {
                ut.setDuration(v);
            } else if ("end".equals(k)) {
                ut.setEnd(v);
            } else if ("ip".equals(k)) {
                ut.setIp(v);
            } else if ("machine_no".equals(k)) {
                ut.setMachine_no(v);
            } else if ("start".equals(k)) {
                ut.setStart(v);
            } else if ("time".equals(k)) {
                ut.setTime(v);
            }
        }

        return ut;
    }

    /**
     * 关闭资源
     *
     * @param table HBase 表对象
     * @param hConn HBase 连接对象
     */
    private void close(HTableInterface table, HConnection hConn) {
        try {
            if (table != null) {
                table.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        try {
            if (hConn != null) {
                hConn.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}