banner.webp

1.代码编写

导入依赖：

  1.         <dependency>
  2.             <groupId>org.springframework.boot</groupId>
  3.             <artifactId>spring-boot-starter</artifactId>
  4.         </dependency>
  5.         <dependency>
  6.             <groupId>org.springframework.boot</groupId>
  7.             <artifactId>spring-boot-starter-test</artifactId>
  8.             <scope>test</scope>
  9.             <exclusions>
  10.                 <exclusion>
  11.                     <groupId>org.junit.vintage</groupId>
  12.                     <artifactId>junit-vintage-engine</artifactId>
  13.                 </exclusion>
  14.             </exclusions>
  15.         </dependency>
  16.         <!--HBase -->
  17.         <dependency>
  18.             <groupId>org.apache.hbase</groupId>
  19.             <artifactId>hbase-client</artifactId>
  20.             <version>2.3.0</version>
  21.         </dependency>
  22.         <dependency>
  23.             <groupId>org.projectlombok</groupId>
  24.             <artifactId>lombok</artifactId>
  25.             <version>1.18.4</version>
  26.         </dependency>

配置文件

  1. ## HBase 配置
  2. hbase:
  3.   config:
  4.     hbase:
  5.       zookeeper:
  6.         quorum: hadoop0
  7.         port: 2181
  8.         znode: /hbase
  9.       master: hadoop:16000
  10.       client:
  11.         keyvalue:
  12.           maxsize: 1572864000

HBaseProperties.java

  1. @Data
  2. @ConfigurationProperties(prefix = "hbase")
  3. public class HBaseProperties {
  4.     private Map<String, String> config;
  5. }

HBaseConfig.java

  1. @Configuration
  2. @EnableConfigurationProperties(HBaseProperties.class)
  3. public class HBaseConfig {
  5.     private final HBaseProperties properties;
  7.     public HBaseConfig(HBaseProperties properties) {
  8.         this.properties = properties;
  9.     }
  11.     @Bean
  12.     public org.apache.hadoop.conf.Configuration configuration() {
  14.         org.apache.hadoop.conf.Configuration configuration = HBaseConfiguration.create();
  16.         Map<String, String> config = properties.getConfig();
  18.         Set<String> keySet = config.keySet();
  19.         for (String key : keySet) {
  20.             configuration.set(key, config.get(key));
  21.         }
  22.         return configuration;
  23.     }
  25. }

HBaseService.java

  1. @Slf4j
  2. @Component
  3. @DependsOn("HBaseConfig")
  4. public class HBaseService {
  6.     @Resource
  7.     private HBaseConfig config;
  9.     private static Admin admin = null;
  11.     public static Configuration conf=null;
  13.     private static Connection connection = null;
  15.     private final ThreadLocal<List<Put>> threadLocal = new ThreadLocal<>();
  17.     private static final int CACHE_LIST_SIZE = 1000;
  21.     @PostConstruct
  22.     private void init() {
  23.         if (connection != null) {
  24.             return;
  25.         }
  27.         try {
  28.             connection = ConnectionFactory.createConnection(config.configuration());
  29.             admin = connection.getAdmin();
  30.         } catch (IOException e) {
  31.             log.error("HBase create connection failed: {}","异常");
  32.         }
  33.     }
  35.     /**
  36.      * create 'tableName','[Column Family 1]','[Column Family 2]'
  37.      * @param tableName
  38.      * @param columnFamilies 列族名
  39.      * @throws IOException
  40.      */
  41.     public void createTable(String tableName, String... columnFamilies) throws IOException {
  42.         TableName name = TableName.valueOf(tableName);
  43.         boolean isExists = this.tableExists(tableName);
  44.         if (isExists) {
  45.             throw new TableExistsException(tableName + "is exists!");
  46.         }
  47.         TableDescriptorBuilder descriptorBuilder = TableDescriptorBuilder.newBuilder(name);
  48.         List<ColumnFamilyDescriptor> columnFamilyList = new ArrayList<>();
  49.         for (String columnFamily : columnFamilies) {
  50.             ColumnFamilyDescriptor columnFamilyDescriptor = ColumnFamilyDescriptorBuilder
  51.                     .newBuilder(columnFamily.getBytes()).build();
  52.             columnFamilyList.add(columnFamilyDescriptor);
  53.         }
  54.         descriptorBuilder.setColumnFamilies(columnFamilyList);
  55.         TableDescriptor tableDescriptor = descriptorBuilder.build();
  56.         admin.createTable(tableDescriptor);
  57.     }
  59.     /**
  60.      * put <tableName>,<rowKey>,<family:column>,<value>,<timestamp>
  61.      * @param tableName
  62.      * @param rowKey
  63.      * @param columnFamily
  64.      * @param column
  65.      * @param value
  66.      * @throws IOException
  67.      */
  68.     public void insertOrUpdate(String tableName, String rowKey, String columnFamily, String column, String value)
  69.             throws IOException {
  70.         this.insertOrUpdate(tableName, rowKey, columnFamily, new String[]{column}, new String[]{value});
  71.     }
  73.     /**
  74.      * put <tableName>,<rowKey>,<family:column>,<value>,<timestamp>
  75.      * @param tableName
  76.      * @param rowKey
  77.      * @param columnFamily
  78.      * @param columns
  79.      * @param values
  80.      * @throws IOException
  81.      */
  82.     public void insertOrUpdate(String tableName, String rowKey, String columnFamily, String[] columns, String[] values)
  83.             throws IOException {
  84.         Table table = connection.getTable(TableName.valueOf(tableName));
  85.         Put put = new Put(Bytes.toBytes(rowKey));
  86.         for (int i = 0; i < columns.length; i++) {
  87.             put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columns[i]), Bytes.toBytes(values[i]));
  88.             table.put(put);
  89.         }
  90.     }
  92.     /**
  93.      * @param tableName
  94.      * @param rowKey
  95.      * @throws IOException
  96.      */
  97.     public void deleteRow(String tableName, String rowKey) throws IOException {
  98.         Table table = connection.getTable(TableName.valueOf(tableName));
  99.         Delete delete = new Delete(rowKey.getBytes());
  100.         table.delete(delete);
  101.     }
  103.     /**
  104.      * @param tableName
  105.      * @param rowKey
  106.      * @param columnFamily
  107.      * @throws IOException
  108.      */
  109.     public void deleteColumnFamily(String tableName, String rowKey, String columnFamily) throws IOException {
  110.         Table table = connection.getTable(TableName.valueOf(tableName));
  111.         Delete delete = new Delete(rowKey.getBytes());
  112.         delete.addFamily(Bytes.toBytes(columnFamily));
  113.         table.delete(delete);
  114.     }
  116.     /**
  117.      * delete 'tableName','rowKey','columnFamily:column'
  118.      * @param tableName
  119.      * @param rowKey
  120.      * @param columnFamily
  121.      * @param column
  122.      * @throws IOException
  123.      */
  124.     public void deleteColumn(String tableName, String rowKey, String columnFamily, String column) throws IOException {
  125.         Table table = connection.getTable(TableName.valueOf(tableName));
  126.         Delete delete = new Delete(rowKey.getBytes());
  127.         delete.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
  128.         table.delete(delete);
  129.     }
  131.     /**
  132.      * disable 'tableName' 之后 drop 'tableName'
  133.      * @param tableName
  134.      * @throws IOException
  135.      */
  136.     public void deleteTable(String tableName) throws IOException {
  137.         boolean isExists = this.tableExists(tableName);
  138.         if (!isExists) {
  139.             return;
  140.         }
  141.         TableName name = TableName.valueOf(tableName);
  142.         admin.disableTable(name);
  143.         admin.deleteTable(name);
  144.     }
  146.     /**
  147.      * get 'tableName','rowkey','family:column'
  148.      * @param tableName
  149.      * @param rowkey
  150.      * @param family
  151.      * @param column
  152.      * @return
  153.      */
  154.     public String getValue(String tableName, String rowkey, String family, String column) {
  155.         Table table = null;
  156.         String value = "";
  157.         if (StringUtils.isBlank(tableName) || StringUtils.isBlank(family) || StringUtils.isBlank(rowkey) || StringUtils
  158.                 .isBlank(column)) {
  159.             return null;
  160.         }
  161.         try {
  162.             table = connection.getTable(TableName.valueOf(tableName));
  163.             Get g = new Get(rowkey.getBytes());
  164.             g.addColumn(family.getBytes(), column.getBytes());
  165.             Result result = table.get(g);
  166.             List<Cell> ceList = result.listCells();
  167.             if (ceList != null && ceList.size() > 0) {
  168.                 for (Cell cell : ceList) {
  169.                     value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
  170.                 }
  171.             }
  172.         } catch (IOException e) {
  173.             e.printStackTrace();
  174.         } finally {
  175.             try {
  176.                 table.close();
  177.                 connection.close();
  178.             } catch (IOException e) {
  179.                 e.printStackTrace();
  180.             }
  181.         }
  182.         return value;
  183.     }
  185.     /**
  186.      * get 'tableName','rowKey'
  187.      * @param tableName
  188.      * @param rowKey
  189.      * @return
  190.      * @throws IOException
  191.      */
  192.     public String selectOneRow(String tableName, String rowKey) throws IOException {
  193.         Table table = connection.getTable(TableName.valueOf(tableName));
  194.         Get get = new Get(rowKey.getBytes());
  195.         Result result = table.get(get);
  196.         NavigableMap<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> map = result.getMap();
  197.         for (Cell cell : result.rawCells()) {
  198.             String row = Bytes.toString(cell.getRowArray());
  199.             String columnFamily = Bytes.toString(cell.getFamilyArray());
  200.             String column = Bytes.toString(cell.getQualifierArray());
  201.             String value = Bytes.toString(cell.getValueArray());
  202.             // 可以通过反射封装成对象(列名和Java属性保持一致)
  203.             System.out.println(row);
  204.             System.out.println(columnFamily);
  205.             System.out.println(column);
  206.             System.out.println(value);
  207.         }
  208.         return null;
  209.     }
  211.     /**
  212.      * scan 't1',{FILTER=>"PrefixFilter('2015')"}
  213.      * @param tableName
  214.      * @param rowKeyFilter
  215.      * @return
  216.      * @throws IOException
  217.      */
  218.     public String scanTable(String tableName, String rowKeyFilter) throws IOException {
  219.         Table table = connection.getTable(TableName.valueOf(tableName));
  220.         Scan scan = new Scan();
  221.         if (!StringUtils.isEmpty(rowKeyFilter)) {
  222.             RowFilter rowFilter = new RowFilter(CompareOperator.EQUAL, new SubstringComparator(rowKeyFilter));
  223.             scan.setFilter(rowFilter);
  224.         }
  225.         ResultScanner scanner = table.getScanner(scan);
  226.         try {
  227.             for (Result result : scanner) {
  228.                 System.out.println(Bytes.toString(result.getRow()));
  229.                 for (Cell cell : result.rawCells()) {
  230.                     System.out.println(cell);
  231.                 }
  232.             }
  233.         } finally {
  234.             if (scanner != null) {
  235.                 scanner.close();
  236.             }
  237.         }
  238.         return null;
  239.     }
  241.     /**
  242.      * 判断表是否已经存在，这里使用间接的方式来实现
  243.      *
  244.      * admin.tableExists() 会报NoSuchColumnFamilyException， 有人说是hbase-client版本问题
  245.      * @param tableName
  246.      * @return
  247.      * @throws IOException
  248.      */
  249.     public boolean tableExists(String tableName) throws IOException {
  250.         TableName[] tableNames = admin.listTableNames();
  251.         if (tableNames != null && tableNames.length > 0) {
  252.             for (int i = 0; i < tableNames.length; i++) {
  253.                 if (tableName.equals(tableNames[i].getNameAsString())) {
  254.                     return true;
  255.                 }
  256.             }
  257.         }
  258.         return false;
  259.     }
  260. }

2.单元测试

  1. @SpringBootTest
  2. @Log4j2
  3. class HBaseApplicationTests {
  5.     @Resource
  6.     HBaseService hBaseService;
  8.     @Test
  9.     void createTable() throws IOException {
  10.         hBaseService.createTable("scores","name","grad","course");
  11.         hBaseService.insertOrUpdate("scores","xiapi","grad","grad","1");
  12.         hBaseService.insertOrUpdate("scores","xiapi","course","Chinese","97");
  13.         hBaseService.insertOrUpdate("scores","xiapi","course","Math","128");
  14.         hBaseService.insertOrUpdate("scores","xiapi","course","English","85");
  15.     }
  17.     @Test
  18.     public void selectOneRow() throws IOException {
  19.         hBaseService.selectOneRow("scores", "xiapi");
  20.     }
  22.     @Test
  23.     public void getValue() {
  24.         String result = hBaseService.getValue("scores", "xiapi", "grad", "grad");
  25.         System.out.println(result);
  27.     }
  29.     @Test
  30.     public void deleteTable() throws IOException {
  31.         hBaseService.deleteTable("scores");
  32.     }
  34. }

这里只简单测试几个方法，方法根据业务需求自己改造~

参考链接：https://blog.csdn.net/weixin_42685328/article/details/119040622