hbase 自动刷写API

绝地灬酷狼 2024-04-17 15:09 79阅读 0赞

1，旧的Htable API

>     storm_wxgz = Resources.hbasePro.getProperty("table_name");
>             wxgz_scenes_data = Resources.hbasePro.getProperty("hbase_log_table");
>     
>             try {
>                 if (connection == null) {
>                     String zkQuorum = Resources.hbasePro.getProperty("hbase_zookeeper_quorum");
>                     String clientPort = Resources.hbasePro.getProperty("hbase_zookeeper_property_clientport");
>     
>                     conf = HBaseConfiguration.create();
>                     conf.set("hbase.zookeeper.quorum", zkQuorum);
>                     conf.set("hbase.zookeeper.property.clientPort", clientPort);
>                     connection = ConnectionFactory.createConnection(conf);
>     
>     
>                     wxgzHtable = new HTable(conf, TableName.valueOf(storm_wxgz));
>                     wxgzHtable.setWriteBufferSize(10 * 1024 * 1024);
>                     wxgzHtable.setAutoFlushTo(false);
>     
>                     scenesHtable = new HTable(conf, TableName.valueOf(wxgz_scenes_data));
>     
>                     //10M
>                     scenesHtable.setWriteBufferSize(10 * 1024 * 1024);
>                     scenesHtable.setAutoFlushTo(false);
>     
>     
>                     storm_wxgz_table = connection.getTable(TableName.valueOf(storm_wxgz));
>                     scenes_data_table = connection.getTable(TableName.valueOf(wxgz_scenes_data));
>     //                scenes_data_table = connection.getTable(TableName.valueOf("hbase_test"));
>     
>     //
>                 }
>     
>             } catch (Exception ex) {
>                 ex.printStackTrace();
>             }

2，新的API

>     hbasePro = Resources.hbasePro;
>     storm_wxgz = hbasePro.getProperty("table_name");
>     wxgz_scenes_data = hbasePro.getProperty("hbase_log_table");
>     
>     try {
>         if (connection == null) {
>             String zkQuorum = hbasePro.getProperty("hbase_zookeeper_quorum");
>             String clientPort = hbasePro.getProperty("hbase_zookeeper_property_clientport");
>     
>             Configuration hbaseConf = HBaseConfiguration.create();
>             hbaseConf.set("hbase.zookeeper.quorum", zkQuorum);
>             hbaseConf.set("hbase.zookeeper.property.clientPort", clientPort);
>             connection = ConnectionFactory.createConnection(hbaseConf);
>     
>     
>             stormTable = connection.getTable(TableName.valueOf(storm_wxgz));
>             scenesTable = connection.getTable(TableName.valueOf(wxgz_scenes_data));
>     
>             //todo Test测试hbase 新API
>             TableName stormTableName = TableName.valueOf(storm_wxgz);
>             BufferedMutatorParams stormBuffSize = new BufferedMutatorParams(stormTableName).writeBufferSize(5 * 1024 * 1024);
>             stormMutator = connection.getBufferedMutator(stormBuffSize);
>     
>             //todo 设置缓存5m，当达到1m时数据会自动刷到hbase
>             TableName scenesTableName = TableName.valueOf(wxgz_scenes_data);
>             BufferedMutatorParams scenesBuffSize = new BufferedMutatorParams(scenesTableName).writeBufferSize(5 * 1024 * 1024);
>             //todo 创建一个批量异步与hbase通信的对象
>             scenesMutator = connection.getBufferedMutator(scenesBuffSize);
>     
>     
>         }
>     } catch (Exception ex) {
>         logger.error("HbaseUtilService初始化-81,抛出异常:", ex);
>     }

使用代码：

>     public static void wxgzHtableFlushCommits(List<Put> hbaseOfValueList) {
>     
>         try {
>             if (hbaseOfValueList.size() >0){
>                 //给全局变量hTable赋值
>                 stormMutator.mutate(hbaseOfValueList);
>                 stormMutator.flush();
>             }
>     
>         } catch (Exception ex) {
>             logger.error("putListWithHTable()方法-160,抛出异常:", ex);
>         } finally {
>             if (stormMutator != null) {
>                 try {
>                     stormMutator.close();
>                 } catch (IOException e) {
>                     e.printStackTrace();
>                 }
>             }
>     
>         }
>     }