实现HBase增量入库（HBase删除自定义时间戳行数据）

小程序：扫一扫查出行
【扫一扫了解最新限行尾号】
复制小程序

1. 背景描述
2. 问题描述
3. 解决方案
- 3.1 直接删除行数据
- 3.2 删除自定义时间戳的行数据

1. 背景描述

目前在做音乐推荐项目，前期做排序模型优化，任务是使用模型对用户的历史音乐进行排序，有6800多万个用户，约40G的用户数据，使用HBase作为数据仓库。

利用HBase可以存储多个版本数据的特性，数据运算完后入库时，将用户id作为rowkey，songInfo:songid的值为歌曲id，使用自定义时间戳，将排序模型输出的歌曲得分（分值越高）作为时间戳。因为HBase数据存储默按照时间戳降序存储，这样只要取出用户的songInfo:songid的所有版本的值就能获取该用户的历史音乐经过模型排序后的顺序。

2. 问题描述

每次数据入库时，使用oozie进行任务调度，先使用truncate命令将原表数据清空，然后根据用户日志进行计算后入库。也就是不管用户的历史音乐数据是否有发生改变，都将HBase中的用户数据删除，根据数据统计，每天的活动用户约为20w占总用户的1/340，换句话说，对339/340的用户进行了没有必要的重复计算，因为他们的历史数据没有发生改变。如果能使用增量入库的方式，只获取活跃用户的日志数据进行计算并入库，可以大大节约资源。

3. 解决方案

3.1 直接删除行数据

最直接的想法是，直接在入库前增加删除语句就行，拿到日志有变化的用户后，删除用户数据，然后入库。所以直接新增了以下语句

Delete delete = new Delete(Bytes.toBytes(StringUtils.trimWhite(userid)));table.delete(delete);

查看最后的入库结果，入库失败，只入了几百条数据，而且根本不是想要的效果。

3.2 删除自定义时间戳的行数据

Delete对象在删除前没限制删除的列簇和时间戳，就是删除了该rowkey所对应的所有列簇的所有字段，所以删除数据的操作没问题，可是入库的代码没有改动，之前入库没有问题。继续了解HBase删除机制后找到了原因。

其实问题出在了自定义时间戳上。在入库的时间戳是根据模型计算出来的，远小于当前的时间的时间戳，而删除的时候没有指定时间戳，HBase会默认使用服务器生成的当前时间的时间戳。而HBase的删除操作并不是真正的删除，可以看成是含有Delete标记的特殊put，只是先给数据打上标记，时间戳小于这个删除时间的数据在下一次major compaction的时候才被真正的删除。由于删除后入库的数据使用的是自定义时间戳且远小于当前时间的时间戳，所以导致了入库的数据被HBase删除了。

自定义时间戳远小于删除时自动生成的时间戳，按理来说最后入库结果应该一条数据都没有（因为小于删除时间戳的数据都被删除了），但为什么最后还是入了几百条的数据呢？推测原因是因为在最后数据的put过程中发生了major compaction，HBase进行了真正的删除，删除数据后Delete标记也失效了，所以后面的put操作才会真正生效。

既然问题定位到了时间戳，那就在删除的时候指定时间戳，还是使用自定的时间戳，但是不直接使用模型生成的分值，而是把歌曲按分值升序排序后从1开始编号，将删除时的时间戳+歌曲编号作为入库时的时间戳。在删除的时候直接指定删除时间戳为当前的时间戳，这样之前的用户数据就会被删除，而新入库的数据时间戳大于被删除的时间戳，就会被保留。关键代码如下：

long offset = new Date().getTime(); //判断该用户是否存在 Get get = new Get(Bytes.toBytes(StringUtils.trimWhite(userid))); get.addColumn(Bytes.toBytes(columnFamily),Bytes.toBytes("songID")); get.setMaxVersions(1); Result result = table.get(get); //System.out.println("这是result.rawCells()前的result对象"+result.toString()); if (!result.isEmpty()) { //KeyValue[] kV = result.raw(); //offset = kV[0].getTimestamp(); //System.out.println("获取的最新时间戳为： "+offset); //offset = result.rawCells()[0].getTimestamp(); //报错找不到该方法，可能是hbase版本太低 //删除之前的数据 Delete delete = new Delete(Bytes.toBytes(StringUtils.trimWhite(userid))); delete.deleteFamily(Bytes.toBytes(columnFamily),offset); table.delete(delete); }