1,獲取hbase client

/*** Create a new Connection instance using the passed <code>conf</code> instance. Connection* encapsulates all housekeeping for a connection to the cluster. All tables and interfaces* created from returned connection share zookeeper connection, meta cache, and connections* to region servers and masters.* <br>* The caller is responsible for calling {@link Connection#close()} on the returned* connection instance.** Typical usage:* <pre>* Connection connection = ConnectionFactory.createConnection(conf);* Table table = connection.getTable(TableName.valueOf("mytable"));* try {* table.get(...);* ...* } finally {* table.close();* connection.close();* }* </pre>** @param conf configuration* @return Connection object for <code>conf</code>*/public static Connection createConnection(Configuration conf) throws IOException {return createConnection(conf, null, null);}

　　

2,根據前綴匹配進行搜索并分批次獲取結果

private static AtomicLong filterAndDeleRowkeysByYearMonth(String yearMonth,String tableName,String devNo,String[] dayDateRange,String zkIp,String esIp,String esTableName) throws IOException { // Set dates = new HashSet<>(Arrays.asList("20180124")); //需要查詢的日期集合Set dates = new HashSet<>(Arrays.asList(dayDateRange));Table table = HBaseUtil.getTable(tableName, zkIp);TransportClient client = ESConnectFactory.getTransportClientInstance(esIp);Scan s = new Scan();s.setCaching(10000000);s.setFilter(new PrefixFilter(yearMonth.getBytes())); // s.setFilter(new PrefixFilter(yearMonth.getBytes())); // s.setMaxResultSize(100000000);ResultScanner resultScanner = table.getScanner(s);AtomicLong atomicLong1 = new AtomicLong();Result[] next = null;do {next = resultScanner.next(100000); // long l = runDelete(next, devNo, dates, atomicLong, table, client);ThreadPoolManager.getInstance().getProduceThreadPool().execute(new RowKeyProducerThread(next, devNo, dates));ThreadPoolManager.getInstance().getConsumeThreadPool().execute(new RowKeyConsumerThread(table,devNo,atomicLong1,client, esTableName));}while(next != null && next.length > 0 );try {Thread.sleep(10000);} catch (InterruptedException e) {e.printStackTrace();}ThreadPoolManager.getInstance().getProduceThreadPool().shutdown();ThreadPoolManager.getInstance().getConsumeThreadPool().shutdown();return atomicLong1;}

　　

轉載于:https://www.cnblogs.com/yanyuechao/p/8367517.html

總結

以上是生活随笔為你收集整理的hbase，根据前缀匹配进行搜索并分批次获取结果的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。