Java SpringBoot 连接 Hbase Demo

package org.example.controller; import lombok.extern.slf4j.Slf4j; import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.CompareOperator; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.client.*; import org.apache.hadoop.hbase.filter.ColumnValueFilter; import org.apache.hadoop.hbase.filter.Filter; import org.apache.hadoop.hbase.util.Bytes; import org.example.config.HbaseConnectionPool; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import javax.annotation.Resource; import java.io.IOException; import java.util.*; /** * @author Administrator * * 可利用 aop 进行连接获取和释放处理 * * @Description * @create 2023-07-25 23:06 */ @Slf4j @RestController @RequestMapping("/hbase") public class HbaseController { @Resource private HbaseConnectionPool pool; /** * 表名 */ private String tbl_user = "tbl_user"; /** * 创建表(不允许重复创建) */ @GetMapping("/create") public void createTable(){ Connection conn = null; //获取连接 try { conn = pool.getConnection(); Admin admin = conn.getAdmin(); TableName tableName = TableName.valueOf(tbl_user); if (!admin.tableExists(tableName)){ //指定表名 TableDescriptorBuilder tdb_user = TableDescriptorBuilder.newBuilder(tableName); //添加列族(info,data) ColumnFamilyDescriptor hcd_info = ColumnFamilyDescriptorBuilder.of("name"); ColumnFamilyDescriptor hcd_data = ColumnFamilyDescriptorBuilder.of("age"); tdb_user.setColumnFamily(hcd_info); tdb_user.setColumnFamily(hcd_data); //创建表 TableDescriptor td = tdb_user.build(); admin.createTable(td); } } catch (IOException e) { throw new RuntimeException(e); } finally { if (null != conn){ pool.releaseConnection(conn); } } } /** * 删除表(不允许删除不存在的表) */ @GetMapping("/drop") public void dropTable(){ Connection conn = null; try { conn = pool.getConnection(); Admin admin = conn.getAdmin(); TableName tableName = TableName.valueOf(tbl_user); if (admin.tableExists(tableName)){ admin.disableTable(tableName); admin.deleteTable(tableName); } } catch (IOException e) { throw new RuntimeException(e); } finally { if (null != conn){ pool.releaseConnection(conn); } } } /** * 插入测试 */ @GetMapping("/insert") public void insert(){ log.info("---插入一列数据---1"); putData(tbl_user, "row1", "name", "a", "zhangSan"); putData(tbl_user, "row1", "age", "a", "18"); log.info("---插入多列数据---2"); putData(tbl_user, "row2", "name", Arrays.asList("a", "b", "c"), Arrays.asList("liSi", "wangWu", "zhaoLiu")); log.info("---插入多列数据---3"); putData(tbl_user, "row3", "age", Arrays.asList("a", "b", "c"), Arrays.asList("18","19","20")); log.info("---插入多列数据---4"); putData(tbl_user, "row4", "age", Arrays.asList("a", "b", "c"), Arrays.asList("30","19","20")); } /** * 插入数据（单条） * @param tableName 表名 * @param rowKey rowKey * @param columnFamily 列族 * @param column 列 * @param value 值 * @return true/false */ public boolean putData(String tableName, String rowKey, String columnFamily, String column, String value) { return putData(tableName, rowKey, columnFamily, Arrays.asList(column), Arrays.asList(value)); } /** * 插入数据（批量） * @param tableName 表名 * @param rowKey rowKey * @param columnFamily 列族 * @param columns 列 * @param values 值 * @return true/false */ public boolean putData(String tableName, String rowKey, String columnFamily, List<String> columns, List<String> values) { Connection conn = null; try { conn = pool.getConnection(); Table table = conn.getTable(TableName.valueOf(tableName)); Put put = new Put(Bytes.toBytes(rowKey)); for (int i=0; i<columns.size(); i++) { put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columns.get(i)), Bytes.toBytes(values.get(i))); } table.put(put); table.close(); return true; } catch (IOException e) { e.printStackTrace(); return false; } finally { if (null != conn){ pool.releaseConnection(conn); } } } /** * 查询测试 */ @GetMapping("/query") public void getResultScanner(){ log.info("全表数据:{}",getData(tbl_user)); log.info("过滤器,按年龄 [18]:{}",getData(tbl_user,new ColumnValueFilter(Bytes.toBytes("age"), Bytes.toBytes("a"), CompareOperator.EQUAL, Bytes.toBytes("18")))); log.info("根据 rowKey [row1]:{}",getData(tbl_user,"row1")); log.info("根据 rowKey 列族 列 [row2 name a]:{}",getData(tbl_user,"row2","name","a")); } /** * 获取数据（全表数据） * @param tableName 表名 * @return map */ public List<Map<String, String>> getData(String tableName) { List<Map<String, String>> list = new ArrayList<>(); Connection conn = null; try { conn = pool.getConnection(); Table table = conn.getTable(TableName.valueOf(tableName)); Scan scan = new Scan(); ResultScanner resultScanner = table.getScanner(scan); for(Result result : resultScanner) { HashMap<String, String> map = new HashMap<>(result.listCells().size()); map.put("row", Bytes.toString(result.getRow())); for (Cell cell : result.listCells()) { //列族 String family = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()); //列 String qualifier = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()); //值 String data = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()); map.put(family + ":" + qualifier, data); } list.add(map); } table.close(); } catch (IOException e) { e.printStackTrace(); } finally { if (null != conn){ pool.releaseConnection(conn); } } return list; } /** * 获取数据（根据 filter） * @param tableName 表名 * @param filter 过滤器 * @return map */ public List<Map<String, String>> getData(String tableName, Filter filter) { List<Map<String, String>> list = new ArrayList<>(); Connection conn = null; try { conn = pool.getConnection(); Table table = conn.getTable(TableName.valueOf(tableName)); Scan scan = new Scan(); // 添加过滤