Hbase代码示例

  <dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-client</artifactId><version>1.2.0-cdh5.14.4</version></dependency><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-server</artifactId><version>1.2.0-cdh5.14.4</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-hdfs</artifactId><version>2.6.0-cdh5.14.4</version></dependency><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-rsgroup</artifactId><version>1.2.0-cdh5.14.4</version></dependency>

HBase的连接是可以复用的，只需要初始化一次，用完关闭table和connection的连接即可，下面将hbase的常用操作封装了工具类中，使用时请根据实际情况进行代码的修改：

public class HBaseUtils {// HBase Connection 创建是一个耗资源的过程// 一般只创建一个 Connection 实例，其它地方共享该实例// 如果往一个表中连续写入数据，建议Table对象也只初始化一次，其他地方共用该实例private static Connection conn = null;private static Table table = null;static{try{String zkHosts = "10.193.222.3,10.193.222.32,10.193.222.31";//请改为配置的方式String zkPort = "2181";//请改为配置的方式String tableName = "testTable"; //请改为配置的方式Configuration conf = ate();conf.set(&#keeper.quorum", zkHosts);conf.set(&#keeper.property.clientPort", zkPort);conf.set(&#dir", "/hbase");conf.set(&#a.replicas.use", "true");conf.set("ies.number", "15");conf.set("hbase.rpc.timeout", "600000");// 使用完需要关闭，先关闭table再关闭connconn = ateConnection(conf, null,new User.ateRemoteUser("11136050")));// 使用完需要关闭，先关闭table再关闭conntable = Table(TableName.valueOf(tableName));} catch(Throwable t) {throw new RuntimeException("HBase connection init failed",t);}}/*** 获取连接* @return*/public static Connection getConnection(){if(null == conn){throw new RuntimeException("HBase connection ");}return conn;}/*** 创建NameSpace* @param nsStr*/public static void createNamespace(String nsStr) {Admin admin = null;try {admin = getConnection().getAdmin();NamespaceDescriptor nd = ate(nsStr).build();ateNamespace(nd);} catch (NamespaceNotFoundException e) {e.printStackTrace();} catch (IOException e1) {e1.printStackTrace();} finally {close(admin);}}/*** 删除NameSpace* @param nsStr*/public static void deleteNamespace(String nsStr) {Admin admin = null;try {admin = getConnection().getAdmin();NamespaceDescriptor(nsStr);admin.deleteNamespace(nsStr);} catch (Exception e) {e.printStackTrace();} finally {close(admin);}}/*** 判断表是否存在* @param tableName* @return*/public static boolean isTableExist(TableName tableName) {boolean flag = false;Admin admin = null;try {admin = getConnection().getAdmin();flag = admin.tableExists(tableName);} catch (IOException e) {e.printStackTrace();} finally {close(admin);}return flag;}/*** 创建表* @param tableName* @param cf*/public static void createTable(TableName tableName, String cf) {HTableDescriptor td = new HTableDescriptor(tableName);HColumnDescriptor hcd = new HColumnDescriptor(cf);td.addFamily(hcd);Admin admin = null;try {admin = getConnection().getAdmin();ateTable(td);} catch (IOException e) {e.printStackTrace();} finally {close(admin);}}/*** 删除表* @param tableName*/public static void deleteTable(TableName tableName) {Admin admin = null;try {admin = getConnection().getAdmin();admin.disableTable(tableName);admin.deleteTable(tableName);} catch (Exception e) {e.printStackTrace();} finally {close(admin);}}/*** 删除指定rowkey的数据* @param tableName* @param row*/public static void deleteRow(TableName tableName, String row) {try {Delete delete = new Bytes(row));table.delete(delete);} catch (Exception e) {e.printStackTrace();}}/*** 获取table* @param tableName* @return* @throws IOException*/public static Table getTable(String tableName) throws IOException{return getTable(TableName.valueOf(tableName));}public static Table getTable(TableName tableName) throws IOException{return getConnection().getTable(tableName);}/*** 插入单条记录，连续写入数据建议使用批量入库接口，异常是抛出还是在这捕获由业务方决定* @param put*/public static void put(Put put) {try {table.put(put);} catch (Exception e) {e.printStackTrace();}}/*** 批量写入的接口，异常是抛出还是在这捕获由业务方决定* @param batch* @return*/public static Object[] put(List<Put> batch) {Object[] result = new Object[batch.size()];try {table.batch(batch, result);} catch (Exception e) {e.printStackTrace();}return result;}/*** 查询单条数据* @param rowKey* @param cf* @param column* @return*/public static String get(String rowKey, String cf, String column) {String value = null;try {Get get = new Bytes(rowKey));Result result = (get);value = Bytes(cf), Bytes(column)));} catch (IOException e) {e.printStackTrace();}return value;}/*** scan查询* @param startRow* @param endRow* @param cf* @param column* @return*/public static List<String> scan(String startRow, String endRow, String cf, String column) {List<String> list = new ArrayList<>();try {Scan scan = new Scan();scan.Bytes(startRow));scan.Bytes(endRow));ResultScanner rs = Scanner(scan);Result result = rs.next();// 获取Row();List<Cell> cells = result.listCells();for (Cell cell : cells) {QualifierArray();}while (result != null) {String value = Bytes(cf), Bytes(column)));list.add(value);result = rs.next();}rs.close();} catch (Exception e) {e.printStackTrace();}return list;}/*** 过滤器扫描（返回过滤器指定条件的记录，即类似于搜索功能）* @throws IOException*/public static void filterScan(String cf, String qualifyName) throws IOException {//查找姓名为 zhangsan 的学生信息SingleColumnValueFilter filter = new Bytes(cf), Bytes(qualifyName), CompareFilter.CompareOp.EQUAL, Bytes("zhangsan"));//注意，在数据量大但情况下，该功能虽然能实现，但性能很差//所以 scan 时最好再指定 startRow 和 stopRowScan scan = new Scan();scan.setFilter(filter);Table table = null;ResultScanner rs = null;try {rs = Scanner(scan);for(Result r : rs) {String id = Row());System.out.println("stu id=" + id);}} finally {HBaseUtil.close(rs);}}/*** 关闭连接*/public static synchronized void close() {if (null != table) {close(table);}if (null == conn || conn.isClosed()) {return;}close(conn);}public static void close(Closeable c) {if(null == c) {return;}try {c.close();} catch (IOException e) {throw new RuntimeException("Error happened while closing " + c, e);}}}