【HBase】SpringBoot 整合 HBase 工程实例

落日映苍穹つ 2023-01-12 04:11 197阅读 0赞

> PS：本文是基于曾经我做的项目的部分代码，尽量做到在有限代码的前提下把逻辑表述完整…

首先是 SpringBoot 整合 HBase 需要引入的依赖

<dependency>
        <groupId>com.spring4all</groupId>
        <artifactId>spring-boot-starter-hbase</artifactId>
        <version>1.0.0.RELEASE</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hbase</groupId>
        <artifactId>hbase-client</artifactId>
        <version>2.1.0</version>
    </dependency>

### 1.建表 ###

我们将用户子系统的所有数据都是存在 HBase 中，将建 4 张表：

*  User：所有用户数据
 *  PassTemplate：商户投放的所有优惠券信息
 *  Pass：已经被用户们领取的优惠券
 *  FeedBack：评论反馈

create_namespace 'pb'
    
    create 'pb:user', {NAME => 'b', VERSIONS => '3', TTL => '2147483647', 'BLOOMFILTER' => 'ROW'}, {NAME => 'o', VERSIONS => '3', TTL => '2147483647', 'BLOOMFILTER' => 'ROW'}
    create 'pb:pass', {NAME => 'i', VERSIONS => '3', TTL => '2147483647', 'BLOOMFILTER' => 'ROW'}
    create 'pb:passtemplate', {NAME => 'b', VERSIONS => '3', TTL => '2147483647', 'BLOOMFILTER' => 'ROW'}, {NAME => 'c', VERSIONS => '3', TTL => '2147483647', 'BLOOMFILTER' => 'ROW'}
    create 'pb:feedback', {NAME => 'i', VERSIONS => '3', TTL => '2147483647', 'BLOOMFILTER' => 'ROW'}

*  `create_namespace 'pb'`：创建一个 pb 命名空间
 *  `create 'pb:user'`：在 pb 命名空间下创建一张 user 表
 *  `{NAME => 'b', VERSIONS => '3', TTL => '2147483647', 'BLOOMFILTER' => 'ROW'}`：
    
     *  NAME：创建一个列族 B
     *  VERSIONS：有 3 个版本（可以不写，默认3，改小可以节省空间），就像上面的图，有几个版本就能有几行
     *  TTL：过期时间
     *  BLOOMFILTER：布隆过滤器，用 RowKey 来过滤

其中，VERSION、TTL、BLOOMFILTER 是列族属性，如果我们不设置，HBase 会填充默认值

### 2.记录列族名，定义列名 ###

将表名，列族名，列名定义成常量（constant包下的 Constants 类）；格式是一个表一个内部类，在内部类中定义具体常量。

> 注：关于内部类
> 
>  *  内部类与外部类并无从属关系，所以内部类可以 public
>  *  非静态内部类不可以有静态成员，因为它的创建依赖的是外部类实例，但是非静态内部类可以有静态常量

public class Constants {
        
    
        /** 商户优惠券 Kafka Topic */
        public static final String TEMPLATE_TOPIC = "merchants-template";
    
        /** token 文件存储目录 */
        public static final String TOKEN_DIR = "/tmp/token/";
    
        /** 已使用的 token 文件名后缀 */
        public static final String USED_TOKEN_SUFFIX = "_";
    
        /** 用户数的 redis key */
        public static final String USE_COUNT_REDIS_KEY = "imooc-user-count";
    
        /**
         * <h2>User HBase Table</h2>
         * */
        public class UserTable {
        ...}
        
         /**
         * <h2>PassTemplate HBase Table</h2>
         * */
        public class PassTemplateTable {
        
            /** PassTemplate HBase 表名 */
            public static final String TABLE_NAME = "pb:passtemplate";
    
            /** 基本信息列族 */
            public static final String FAMILY_B = "b";
    
            /** 商户 id */
            public static final String ID = "id";
    
            /** 优惠券标题 */
            public static final String TITLE = "title";
    
            /** 优惠券摘要信息 */
            public static final String SUMMARY = "summary";
    
            /** 优惠券详细信息 */
            public static final String DESC = "desc";
    
            /** 优惠券是否有 token */
            public static final String HAS_TOKEN = "has_token";
    
            /** 优惠券背景色 */
            public static final String BACKGROUND = "background";
    
            /** 约束信息列族 */
            public static final String FAMILY_C = "c";
    
            /** 最大个数限制 */
            public static final String LIMIT = "limit";
    
            /** 优惠券开始时间 */
            public static final String START = "start";
    
            /** 优惠券结束时间 */
            public static final String END = "end";
        }
        
        /**
         * <h2>Pass HBase Table</h2>
         * */
        public class PassTable {
        
    		/** Pass HBase 表名 */
            public static final String TABLE_NAME = "pb:pass";
    
            /** 信息列族 */
            public static final String FAMILY_I = "i";
    
            /** 用户 id */
            public static final String USER_ID = "user_id";
    
            /** 优惠券 id */
            public static final String TEMPLATE_ID = "template_id";
    
            /** 优惠券识别码 */
            public static final String TOKEN = "token";
    
            /** 领取日期 */
            public static final String ASSIGNED_DATE = "assigned_date";
    
            /** 消费日期 */
            public static final String CON_DATE = "con_date";
    	}
        
         /**
         * <h2>Feedback Hbase Table</h2>
         * */
        public class Feedback {
        ...}
    }

上面只列出了 PassTemplate 和 Pass 表的相关列属性，User 表和 Feedback 在下面中的演示用不到所以就不列出了。

### 3.创建实体对象（vo包） ###

跟我们使用 MySQL 一样，使用 HBase 也需要一个实体对象

@Data
    @NoArgsConstructor
    @AllArgsConstructor
    public class PassTemplate {
        
    
        /** 所属商户 id */
        private Integer id;
    
        /** 优惠券标题 */
        private String title;
    
        /** 优惠券摘要 */
        private String summary;
    
        /** 优惠券详细信息 */
        private String desc;
    
        /** 最大个数限制 */
        private Long limit;
    
        /** 优惠券是否有 Token, 用于商户核销 */
        private Boolean hasToken;
    
        /** 优惠券背景色 */
        private Integer background;
    
        /** 优惠券开始时间 */
        private Date start;
    
        /** 优惠券结束时间 */
        private Date end;
    }

### 4.手动 ORM 映射（mapper包） ###

> ORM 就是对象关系映射（Object Relation Mapping），即数据库表字段 <-> Java 对象。对于 mysql 等基本的关系型数据库，Java 提供了 JDBC 去进行操作，但是使用 JDBC 时需要根据查询结果手动封装Java对象（这块可以单独抽象一个方法去做）。所以，后来产生了 MyBatis 和 JPA 等框架，它们就对这块进行了处理，可以直接根据对象类型去自动构建 Java 对象。

> 而这个 HBase 目前用的也是基本 API，所以需要我们像 JDBC 那样，去手动封装查询对象，具体的做法是重写继承 RowMapper（目的是获取到查询结果），然后重写 mapRow 方法构建 Java 对象。

public class PassTemplateRowMapper implements RowMapper<PassTemplate> {
        
    
        // B 列族
       	// 注：所有属性名称都要从 String 转换成 byte 数组
        private static byte[] FAMILY_B = Constants.PassTemplateTable.FAMILY_B.getBytes();
        // B 列族的属性
        private static byte[] ID = Constants.PassTemplateTable.ID.getBytes();
        private static byte[] TITLE = Constants.PassTemplateTable.TITLE.getBytes();
        private static byte[] SUMMARY = Constants.PassTemplateTable.SUMMARY.getBytes();
        private static byte[] DESC = Constants.PassTemplateTable.DESC.getBytes();
        private static byte[] HAS_TOKEN = Constants.PassTemplateTable.HAS_TOKEN.getBytes();
        private static byte[] BACKGROUND = Constants.PassTemplateTable.BACKGROUND.getBytes();
    
        // C 列族
        private static byte[] FAMILY_C = Constants.PassTemplateTable.FAMILY_C.getBytes();
        // C 列族的属性
        private static byte[] LIMIT = Constants.PassTemplateTable.LIMIT.getBytes();
        private static byte[] START = Constants.PassTemplateTable.START.getBytes();
        private static byte[] END = Constants.PassTemplateTable.END.getBytes();
    
        // 根据查询结果构建实体对象
        @Override
        public PassTemplate mapRow(Result result, int rowNum) throws Exception {
        
    
            PassTemplate passTemplate = new PassTemplate();
            
    		// 根据 列族名+属性名 得到具体的结果，然后 set 到结果对象中
            passTemplate.setId(Bytes.toInt(result.getValue(FAMILY_B, ID)));
            passTemplate.setTitle(Bytes.toString(result.getValue(FAMILY_B, TITLE)));
            passTemplate.setSummary(Bytes.toString(result.getValue(FAMILY_B, SUMMARY)));
            passTemplate.setDesc(Bytes.toString(result.getValue(FAMILY_B, DESC)));
            passTemplate.setHasToken(Bytes.toBoolean(result.getValue(FAMILY_B, HAS_TOKEN)));
            passTemplate.setBackground(Bytes.toInt(result.getValue(FAMILY_B, BACKGROUND)));
    
            String[] patterns = new String[] {
        "yyyy-MM-dd"};
    
            passTemplate.setLimit(Bytes.toLong(result.getValue(FAMILY_C, LIMIT)));
            passTemplate.setStart(DateUtils.parseDate(Bytes.toString(result.getValue(FAMILY_C, START)), patterns));
            passTemplate.setEnd(DateUtils.parseDate(Bytes.toString(result.getValue(FAMILY_C, END)), patterns));
    
            return passTemplate;
        }
    }

### 5.操作 HBase（service包） ###

下面我们来看看如何向 HBase 中添加数据…

> 注：在进行具体操作前，首先是要判断 RowKey 是否存在。

@Slf4j
    @Service
    public class HBasePassServiceImpl implements IHBasePassService {
        
    
        /** HBase 客户端 */
        @Autowired
        private final HbaseTemplate hbaseTemplate;
    
        // 将商户投放子系统通过 Kafka 发送过来的优惠券信息落库（HBase）
        @Override
        public boolean dropPassTemplateToHBase(PassTemplate passTemplate) {
        
    
            if (null == passTemplate) {
        
                return false;
            }
    		
            // 根据 PassTemplate(优惠券信息) 生成 RowKey
            String rowKey = RowKeyGenUtil.genPassTemplateRowKey(passTemplate);
    
            try {
        
                // 校验 RowKey 在 PassTemplate 表中是否已经存在
                if (hbaseTemplate.getConnection() // 获取 Hbase 连接
                    .getTable(TableName.valueOf(Constants.PassTemplateTable.TABLE_NAME)) // 获取 Hbase 中的 PassTemplate 表（注：TableName.valueOf 是为了将 String 类型的常量转为 HBase 能处理的 TableName 类型
                        .exists(new Get(Bytes.toBytes(rowKey)))) {
         // 判断表中是否已经有这个 RowKey 了
                    
                    log.warn("RowKey {} is already exist!", rowKey);
                    return false;
                }
            } catch (Exception ex) {
        
                log.error("DropPassTemplateToHBase Error: {}", ex.getMessage());
                return false;
            }
    	
            /**
            * 写入 HBase
            */
            // 1.创建 Put 对象
            // 注：Put 需要 RowKey 作为构造函数的实参
            Put put = new Put(Bytes.toBytes(rowKey));
            
    	    // 2.将具体表的具体列族的具体列的具体值，写入 Put
            // 注：这里都要采用 byte[]
            put.addColumn(
                    Bytes.toBytes(Constants.PassTemplateTable.FAMILY_B), // B 列族
                    Bytes.toBytes(Constants.PassTemplateTable.ID), // ID 列
                    Bytes.toBytes(passTemplate.getId()) // 具体值
            );
            put.addColumn(
                    Bytes.toBytes(Constants.PassTemplateTable.FAMILY_B), // B 列族
                    Bytes.toBytes(Constants.PassTemplateTable.TITLE), // TITLE 列
                    Bytes.toBytes(passTemplate.getTitle()) // 具体值
            );
            //...
            put.addColumn(
                    Bytes.toBytes(Constants.PassTemplateTable.FAMILY_C), // 列族 B
                    Bytes.toBytes(Constants.PassTemplateTable.LIMIT), // LIMIT 列
                    Bytes.toBytes(passTemplate.getLimit()) // 具体值
            );
            put.addColumn(
                    Bytes.toBytes(Constants.PassTemplateTable.FAMILY_C), // 列族 C
                    Bytes.toBytes(Constants.PassTemplateTable.START), // START 列
                    Bytes.toBytes(DateFormatUtils.ISO_DATE_FORMAT.format(passTemplate.getStart())) // 具体值（注：这里使用 Apache的工具类 DateFormatUtils，将Date转换成指定String格式）
            );
            //...
            
            // 3.存入 HBase
            // 入参： TableName，Put
            hbaseTemplate.saveOrUpdate(Constants.PassTemplateTable.TABLE_NAME, put);
        }
    }

这里我们再举几个别的例子，来看看除了添加外，如何进行修改和查询…

**1）修改数据**

这里举两个修改列属性的例子（只截取了片段代码）

示例一：将 Pass 表的 CON\_DATE 设置为当前日期

// 1.创建 Mutation 集合
    List<Mutation> datas = new ArrayList<>();
    // 2.构建 Put 对象
    Put put = new Put(passes.get(0).getRowKey().getBytes());
    // 将要修改的表的列族的列的具体值放入 Put
    put.addColumn(FAMILY_I, CON_DATE,
            Bytes.toBytes(DateFormatUtils.ISO_DATE_FORMAT.format(new Date()))
    );
    // 3.将 Put 加入到 Mutation 集合
    datas.add(put);
    
    // 4.HbaseTemplate#saveOrUpdate()
    // 入参：TableName，List<Mutation>
    hbaseTemplate.saveOrUpdates(Constants.PassTable.TABLE_NAME, datas);

示例二：将 Pass 表记录的优惠券的数量 -1

// 1.1.创建 Mutation 集合
    List<Mutation> datas = new ArrayList<>();
    
    byte[] FAMILY_C = Constants.PassTemplateTable.FAMILY_C.getBytes();
    byte[] LIMIT = Constants.PassTemplateTable.LIMIT.getBytes();
    
    // 2.构建 Put 对象
    Put put = new Put(Bytes.toBytes(passTemplateId));
    // 将要修改的表的列族的列的具体值放入 Put
    put.addColumn(FAMILY_C, LIMIT, Bytes.toBytes(passTemplate.getLimit() - 1));
    
    // 3.将 Put 加入到 Mutation 集合
    datas.add(put);
    
    // 4.HbaseTemplate#saveOrUpdate()
    hbaseTemplate.saveOrUpdates(Constants.PassTemplateTable.TABLE_NAME, datas);

**2）RowKey 多行查询**

多行查询也可以认为是条件查询，是通过过滤条件+扫描器来实现的，有 MySQL 的 where 查询的感觉。

来看个例子，查询评论信息并返回 FeedBack 对象

public Response getFeedback(Long userId) {
        
    	// 拿到 RowKey
        // 注：因为 RowKeyGenUtil 生成 FeedBack 的 RowKey 时进行了翻转，所以这里也要转
        byte[] reverseUserId = new StringBuilder(String.valueOf(userId)).reverse().toString().getBytes();
        
        // 1.扫描器 Scan
        Scan scan = new Scan();
        // 2.设置过滤器（相当于 where 条件）
        // 这里是根据 RowKey 进行前缀过滤（注：最常用的就是前缀过滤，因为 RowKey 为了唯一，结构是数据+随机序列）
        scan.setFilter(new PrefixFilter(reverseUserId));
    
        // 3.HbaseTemplate#find()，返回一个 Java 对象
        // 入参：表名，扫描器，结果转换器（我们自定义的那些 RowMapper）
        List<Feedback> feedbacks = hbaseTemplate.find(Constants.Feedback.TABLE_NAME, scan, new FeedbackRowMapper());
    
        return new Response(feedbacks);
    }

HBase 提供了多种过滤器，我们也可以自己创建过滤器。下面来看一个多过滤器，组成过滤器链的例子

/**
     * <h2>根据优惠券状态获取优惠券信息</h2>
     * @param userId 用户 id
     * @param status {@link PassStatus}
     * @return {@link Response}
     * */
    private Response getPassInfoByStatus(Long userId, PassStatus status) throws Exception {
        
    
        // 根据 userId 构造行键前缀
        byte[] rowPrefix = Bytes.toBytes(new StringBuilder(String.valueOf(userId)).reverse().toString());
    	
        // 根据优惠券状态构建比较器
        // 如果是未使用，就比较是否相等；如果不是未使用，就比较是否不相等
        CompareFilter.CompareOp compareOp =
                status == PassStatus.UNUSED ? 
                        CompareFilter.CompareOp.EQUAL : CompareFilter.CompareOp.NOT_EQUAL;
    
        Scan scan = new Scan();
    
        List<Filter> filters = new ArrayList<>();
    
        // 1. 行键前缀过滤器, 找到特定用户的优惠券
        filters.add(new PrefixFilter(rowPrefix));
        // 2. 基于列单元值的过滤器, 找到未使用的优惠券
        // 注：未使用的优惠券 CON_DATE 为 -1（未设值），使用过的优惠券 CON_DATE 是具体使用日期
        if (status != PassStatus.ALL) {
        
            filters.add(
                    new SingleColumnValueFilter(
                            Constants.PassTable.FAMILY_I.getBytes(), // I 列族
                            Constants.PassTable.CON_DATE.getBytes(),  // CON_DATE 列
                        	compareOp, // 找等于的，还是不等于的
                        	Bytes.toBytes("-1")) // -1
            );
        }
    	// 3.将过滤器链设置到扫描器
        scan.setFilter(new FilterList(filters));
    	
        // 4.HbaseTemplate#find()
        // 在 Pass 表中查找当前用户领取的优惠券们
        List<Pass> passes = hbaseTemplate.find(Constants.PassTable.TABLE_NAME, scan, new PassRowMapper());
        
        //...
    }

上面的过滤器链是必须满足所有的过滤条件（默认），当然，也可以手动设置只满足一个过滤条件就行

/**
     * <h2>获取系统中可用的优惠券</h2>
     * @param excludeIds 需要排除的优惠券 ids
     * @return {@link PassTemplate}
     * */
    private List<PassTemplate> getAvailablePassTemplate(List<String> excludeIds) {
        
    	
        // FilterList.Operator.MUST_PASS_ONE：只满足一个过滤器即可
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ONE);
    
        filterList.addFilter(
                new SingleColumnValueFilter(
                        Bytes.toBytes(Constants.PassTemplateTable.FAMILY_C), // C 列族
                        Bytes.toBytes(Constants.PassTemplateTable.LIMIT),  // LIMIT 列（个数限制->剩余）
                        CompareFilter.CompareOp.GREATER, // 大于
                        new LongComparator(0L) // 0（表示可领取）
                )
        );
        filterList.addFilter(
                new SingleColumnValueFilter(
                        Bytes.toBytes(Constants.PassTemplateTable.FAMILY_C), // C 列族
                        Bytes.toBytes(Constants.PassTemplateTable.LIMIT),  // LIMIT 列（个数限制->剩余）
                        CompareFilter.CompareOp.EQUAL, // 等于
                        Bytes.toBytes("-1") // -1（表示无穷大，即不存在上限）
                )
        );
    
        // 扫描器
        Scan scan = new Scan();
        // 添加过滤器链
        scan.setFilter(filterList);
    
        // HbaseTemplate#find() 
        List<PassTemplate> validTemplates = hbaseTemplate.find(
                Constants.PassTemplateTable.TABLE_NAME, scan, new PassTemplateRowMapper());
    
        //...根据 excludeIds 过滤掉排除的优惠券；以及根据 PassTemplate 中保存的时间，过滤掉过期的优惠券（种类）
    }

**3）RowKey 单行查询**

/**
     * <h2>通过获取的 Passes 对象构造 Map</h2>
    **/
    private Map<String, PassTemplate> buildPassTemplateMap(List<Pass> passes) throws Exception {
        
    
        // PassTemplate 的列族、列
        String[] patterns = new String[] {
        "yyyy-MM-dd"};
    
        byte[] FAMILY_B = Bytes.toBytes(Constants.PassTemplateTable.FAMILY_B);
        byte[] ID = Bytes.toBytes(Constants.PassTemplateTable.ID);
        byte[] TITLE = Bytes.toBytes(Constants.PassTemplateTable.TITLE);
        byte[] SUMMARY = Bytes.toBytes(Constants.PassTemplateTable.SUMMARY);
        byte[] DESC = Bytes.toBytes(Constants.PassTemplateTable.DESC);
        byte[] HAS_TOKEN = Bytes.toBytes(Constants.PassTemplateTable.HAS_TOKEN);
        byte[] BACKGROUND = Bytes.toBytes(Constants.PassTemplateTable.BACKGROUND);
    
        byte[] FAMILY_C = Bytes.toBytes(Constants.PassTemplateTable.FAMILY_C);
        byte[] LIMIT = Bytes.toBytes(Constants.PassTemplateTable.LIMIT);
        byte[] START = Bytes.toBytes(Constants.PassTemplateTable.START);
        byte[] END = Bytes.toBytes(Constants.PassTemplateTable.END);
    	
        // 在每个 Pass 对象中，获取到每个优惠券的 RowKey（templateId）
        // 注：这里是通过 lamada 表达式构建结果集
        List<String> templateIds = passes.stream().map(
                Pass::getTemplateId
        ).collect(Collectors.toList());
    
        
        // 创建保存 Get 对象的容器
        List<Get> templateGets = new ArrayList<>(templateIds.size());
        // 通过 RowKey 构造 Get 对象
        templateIds.forEach(t -> templateGets.add(new Get(Bytes.toBytes(t))));
    	// 借助 Get 对象去 HBase 的 PassTemplate 中查找
        Result[] templateResults = hbaseTemplate.getConnection() // 获取连接
                .getTable(TableName.valueOf(Constants.PassTemplateTable.TABLE_NAME)) // 获取表
                .get(templateGets); // 根据 RowKey 查询，返回所有符合的 PassTemplate
        
    
        // 构造 PassTemplate -> PassTemplate Object 的 Map, 用于构造 PassInfo
        Map<String, PassTemplate> templateId2Object = new HashMap<>();
        for (Result item : templateResults) {
        
            PassTemplate passTemplate = new PassTemplate();
    		
            // 列族 B
            passTemplate.setId(Bytes.toInt(item.getValue(FAMILY_B, ID)));
            passTemplate.setTitle(Bytes.toString(item.getValue(FAMILY_B, TITLE)));
            passTemplate.setSummary(Bytes.toString(item.getValue(FAMILY_B, SUMMARY)));
            passTemplate.setDesc(Bytes.toString(item.getValue(FAMILY_B, DESC)));
            passTemplate.setHasToken(Bytes.toBoolean(item.getValue(FAMILY_B, HAS_TOKEN)));
            passTemplate.setBackground(Bytes.toInt(item.getValue(FAMILY_B, BACKGROUND)));
    		
            // 列族 C
            passTemplate.setLimit(Bytes.toLong(item.getValue(FAMILY_C, LIMIT)));
            passTemplate.setStart(DateUtils.parseDate(
                    Bytes.toString(item.getValue(FAMILY_C, START)), patterns));
            passTemplate.setEnd(DateUtils.parseDate(
                    Bytes.toString(item.getValue(FAMILY_C, END)), patterns
            ));
    		
            // => <RowKey, passTemplate>
            templateId2Object.put(Bytes.toString(item.getRow()), passTemplate);
        }
    
        return templateId2Object;
    }