支付宝红包
京东盲盒抽奖
幸运转盘
秒杀
自营热卖
支付宝红包

使用spark 对hbase 进行先删除,后插入 ,部分数据没有插入成功,

雨后、云初霁 1年前   阅读数 128 0

问题描述, 我们的业务需要我们对hbase 中的数据进行修改操作, 因此需要对数据先删除,后插入,  期间发现部分hbase  rowkey  确实删除了, 但是没有插入数据?

问题产生的原因,   delete  和put 默认使用的  timestamp 插入当前hbase 服务器最后的时间, 如果  put的时间<=delete 的时间,当前rowkey  对应的列 都保存这删除的时间,  我们查询数据, 默认是获取时间戳最新的数据,  因此查询不到当前插入的数据

问题解决方案:put  和delete  都支持自己设置 timestamp  通过设置put时间戳大于delete的时间戳的方式,来避免这个问题

重现问题,这里我就不写spark 了 ,直接用hbase 的api 操作单条记录

 private static Logger logger = LoggerFactory.getLogger(SparkSolrByLocal.class);
    static Configuration conf;

    static Connection connection;

    static {
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "192.168.1.38,192.168.1.39,192.168.1.40");
        try {
            connection = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

测试  先往hbase 表中添加数据

重现问题,使得 put 的时间戳小于或者等于 delete的时间戳

 public static void main(String[] args) throws IOException, InterruptedException {
        String tableName = "two";
        Table table = connection.getTable(TableName.valueOf(tableName));// 获取表
        String str = "A0000";
        int s = 0000;
        Date date=new Date();
        long time = date.getTime();
        System.out.println(time);
        Delete delete = new Delete(Bytes.toBytes("r2"),time);
        table.delete(delete);
        Put put = new Put(Bytes.toBytes("r2"), time);

        for (int i = 0; i < 4; i++) {
            s++;
            String s1 = s + "";
            int length = s1.length();
            String substring = str.substring(0, str.length() - length);
            str = substring + s1;
            put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes(str), Bytes.toBytes("1"));

        }
        table.put(put);
    }

执行main后再去hbase shell中查看  two 表中的r2

发现 当put 和delete 时间相等的时候, 再次查询 就没有结果了 ,重现了问题

使用手动设置时间戳的方式,将put 时间戳设置为 delete  之后, 

 public static void main(String[] args) throws IOException, InterruptedException {
        String tableName = "two";
        Table table = connection.getTable(TableName.valueOf(tableName));// 获取表
        String str = "A0000";
        int s = 0000;
        Date date=new Date();
        long time = date.getTime();
        System.out.println(time);
        Delete delete = new Delete(Bytes.toBytes("r2"),time);
        table.delete(delete);
        Put put = new Put(Bytes.toBytes("r2"), time+1);

        for (int i = 0; i < 4; i++) {
            s++;
            String s1 = s + "";
            int length = s1.length();
            String substring = str.substring(0, str.length() - length);
            str = substring + s1;
            put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes(str), Bytes.toBytes("1"));

        }
        table.put(put);
    }

这样就能确保 delete之后put 数据能够被查到, 

解决了使用spark 程序 对hbase 修改操作的问题

 

 


注意:本文归作者所有,未经作者允许,不得转载

全部评论: 0

    我有话说: