• SpringBoot 封装 HBase 操作工具类

            最近项目中用到了Hbase相关的操作并封装成工具类，我的Hbase服务器端版本是2.1.0，图示如下：

            特此记录便于日后查阅。

    一、pom.xml 依赖

    1. <dependency>
    2.     <groupId>org.apache.hbase</groupId>
    3.     <artifactId>hbase-shaded-client</artifactId>
    4.     <version>2.1.0</version>
    5. </dependency>
    6. <dependency>
    7.     <groupId>org.apache.hadoop</groupId>
    8.     <artifactId>hadoop-common</artifactId>
    9.     <version>3.0.0</version>
    10. </dependency>

    二、application.yml 项目配置

            此处我是自定义HBase配置，后面会有专门的配置类来加载这个配置

    1. datasource:
    2.   hbase:
    3.     zookeeper:
    4.       port: 2181
    5.       quorum: 10.0.61.12,10.0.61.22,10.0.61.24
    6.       znode:
    7.         parent: ''

    三、HbaseConfig 自定义配置类

    1. import lombok.Data;
    2. import org.apache.hadoop.hbase.HBaseConfiguration;
    3. import org.apache.hadoop.hbase.client.Connection;
    4. import org.apache.hadoop.hbase.client.ConnectionFactory;
    5. import org.springframework.beans.factory.annotation.Value;
    6. import org.springframework.context.annotation.Configuration;
    7. import org.springframework.stereotype.Component;
    8.  
    9. import java.io.IOException;
    10.  
    11. /**
    12.  * @description: Hbase配置类
    13.  * @author: zhangzhixiang
    14.  * @createDate: 2022/11/24
    15.  * @version: 1.0
    16.  */
    17. @Data
    18. @Component
    19. @Configuration
    20. public class HbaseConfig {
    21.  
    22.     @Value("${datasource.hbase.zookeeper.quorum}")
    23.     private String zookeeper;
    24.  
    25.     @Value("${datasource.hbase.zookeeper.znode.parent}")
    26.     private String parent;
    27.  
    28.     @Value("${datasource.hbase.zookeeper.port}")
    29.     private String port;
    30.  
    31.     public Connection getConnection() throws IOException {
    32.         org.apache.hadoop.conf.Configuration config = HBaseConfiguration.create();
    33.         config.set("hbase.zookeeper.quorum", zookeeper);
    34.         config.set("hbase.zookeeper.property.clientPort", port);
    35.         if (parent != null && !"".equals(parent)) {
    36.             config.set("zookeeper.znode.parent", parent);
    37.         }
    38.         Connection connection = ConnectionFactory.createConnection(config);
    39.         return connection;
    40.     }
    41. }

    四、HbaseUtil 工具类

            首先添加 SpringContext 工具类，下面会用到：

    1. import org.springframework.beans.BeansException;
    2. import org.springframework.context.ApplicationContext;
    3. import org.springframework.context.ApplicationContextAware;
    4. import org.springframework.core.env.Environment;
    5. import org.springframework.stereotype.Component;
    6.  
    7. /**
    8.  * @Description: 
    9.  * @Author：zhangzhixiang
    10.  * @Date: 2022/7/25
    11.  * @Version: 1.0
    12.  */
    13. @Component
    14. public class SpringContext implements ApplicationContextAware {
    15.     public static ApplicationContext context;
    16.     public static Environment env;
    17.  
    18.     @Override
    19.     public void setApplicationContext(ApplicationContext context) throws BeansException {
    20.         SpringContext.context = context;
    21.         SpringContext.env = context.getEnvironment();
    22.     }
    23.  
    24.     public static Object getBean(String name) {
    25.         return context.getBean(name);
    26.     }
    27.  
    28.     public static  T getBean(Class clazz) {
    29.         return context.getBean(clazz);
    30.     }
    31.  
    32.     public static ApplicationContext getContext() {
    33.         return context;
    34.     }
    35.  
    36.     public static Environment getEnv() {
    37.         return env;
    38.     }
    39.  
    40.     public static String getProperty(String key) {
    41.         return getProperty(key, "");
    42.     }
    43.  
    44.     public static String getProperty(String key, String defaultValue) {
    45.         return env.getProperty(key, defaultValue);
    46.     }
    47.  
    48.     public static  T getProperty(String key, Class targetType) {
    49.         return env.getProperty(key, targetType);
    50.     }
    51.  
    52.     public static String getActiveProfile() {
    53.         return env.getActiveProfiles()[0];
    54.     }
    55. }

             然后我们来写 HbaseUtil 工具类的代码：

    1. import com.swkj.common.base.context.SpringContext;
    2. import com.swkj.common.base.log.GLog;
    3. import com.swkj.common.base.log.LogFactory;
    4. import com.swkj.common.hbase.config.HbaseConfig;
    5. import org.apache.hadoop.hbase.Cell;
    6. import org.apache.hadoop.hbase.HColumnDescriptor;
    7. import org.apache.hadoop.hbase.HTableDescriptor;
    8. import org.apache.hadoop.hbase.TableName;
    9. import org.apache.hadoop.hbase.client.*;
    10. import org.apache.hadoop.hbase.filter.CompareFilter;
    11. import org.apache.hadoop.hbase.filter.RowFilter;
    12. import org.apache.hadoop.hbase.filter.SubstringComparator;
    13. import org.apache.hadoop.hbase.util.Bytes;
    14. import org.springframework.context.annotation.DependsOn;
    15. import org.springframework.stereotype.Component;
    16.  
    17. import java.io.IOException;
    18. import java.util.ArrayList;
    19. import java.util.List;
    20. import java.util.NavigableMap;
    21.  
    22. /**
    23.  * @description: Hbase工具类
    24.  * @author: zhangzhixiang
    25.  * @createDate: 2022/11/24
    26.  * @version: 1.0
    27.  */
    28. @DependsOn("springContext")
    29. @Component
    30. public class HbaseUtil {
    31.  
    32.     private static final GLog LOG = LogFactory.getLogger(HbaseUtil.class);
    33.     private static HbaseConfig hbaseConfig = (HbaseConfig) SpringContext.getBean("hbaseConfig");
    34.     private static Connection connection = null;
    35.     private static Admin admin = null;
    36.  
    37.     private HbaseUtil() {
    38.         if (connection == null) {
    39.             try {
    40.                 connection = hbaseConfig.getConnection();
    41.                 admin = connection.getAdmin();
    42.             } catch (IOException e) {
    43.                 LOG.error("HbaseUtils实例初始化失败！错误信息为：" + e.getMessage(), e);
    44.             }
    45.         }
    46.     }
    47.  
    48.     /**
    49.      * 创建表
    50.      *
    51.      * @param tableName    表名
    52.      * @param columnFamily 列族（数组）
    53.      */
    54.     public void createTable(String tableName, String[] columnFamily) throws IOException {
    55.         TableName name = TableName.valueOf(tableName);
    56.         //如果存在则删除
    57.         if (admin.tableExists(name)) {
    58.             admin.disableTable(name);
    59.             admin.deleteTable(name);
    60.             LOG.error("create htable error! this table {} already exists!", name);
    61.         } else {
    62.             HTableDescriptor desc = new HTableDescriptor(name);
    63.             for (String cf : columnFamily) {
    64.                 desc.addFamily(new HColumnDescriptor(cf));
    65.             }
    66.             admin.createTable(desc);
    67.         }
    68.     }
    69.  
    70.     /**
    71.      * 插入记录（单行单列族-多列多值）
    72.      *
    73.      * @param tableName     表名
    74.      * @param row           行名
    75.      * @param columnFamilys 列族名
    76.      * @param columns       列名（数组）
    77.      * @param values        值（数组）（且需要和列一一对应）
    78.      */
    79.     public void insertRecords(String tableName, String row, String columnFamilys, String[] columns, String[] values) throws IOException {
    80.         TableName name = TableName.valueOf(tableName);
    81.         Table table = connection.getTable(name);
    82.         Put put = new Put(Bytes.toBytes(row));
    83.         for (int i = 0; i < columns.length; i++) {
    84.             put.addColumn(Bytes.toBytes(columnFamilys), Bytes.toBytes(columns[i]), Bytes.toBytes(values[i]));
    85.             table.put(put);
    86.         }
    87.     }
    88.  
    89.     /**
    90.      * 插入记录（单行单列族-单列单值）
    91.      *
    92.      * @param tableName    表名
    93.      * @param row          行名
    94.      * @param columnFamily 列族名
    95.      * @param column       列名
    96.      * @param value        值
    97.      */
    98.     public void insertOneRecord(String tableName, String row, String columnFamily, String column, String value) throws IOException {
    99.         TableName name = TableName.valueOf(tableName);
    100.         Table table = connection.getTable(name);
    101.         Put put = new Put(Bytes.toBytes(row));
    102.         put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
    103.         table.put(put);
    104.     }
    105.  
    106.     /**
    107.      * 删除一行记录
    108.      *
    109.      * @param tablename 表名
    110.      * @param rowkey    行名
    111.      */
    112.     public void deleteRow(String tablename, String rowkey) throws IOException {
    113.         TableName name = TableName.valueOf(tablename);
    114.         Table table = connection.getTable(name);
    115.         Delete d = new Delete(rowkey.getBytes());
    116.         table.delete(d);
    117.     }
    118.  
    119.     /**
    120.      * 删除单行单列族记录
    121.      *
    122.      * @param tablename    表名
    123.      * @param rowkey       行名
    124.      * @param columnFamily 列族名
    125.      */
    126.     public void deleteColumnFamily(String tablename, String rowkey, String columnFamily) throws IOException {
    127.         TableName name = TableName.valueOf(tablename);
    128.         Table table = connection.getTable(name);
    129.         Delete d = new Delete(rowkey.getBytes()).addFamily(Bytes.toBytes(columnFamily));
    130.         table.delete(d);
    131.     }
    132.  
    133.     /**
    134.      * 删除单行单列族单列记录
    135.      *
    136.      * @param tablename    表名
    137.      * @param rowkey       行名
    138.      * @param columnFamily 列族名
    139.      * @param column       列名
    140.      */
    141.     public void deleteColumn(String tablename, String rowkey, String columnFamily, String column) throws IOException {
    142.         TableName name = TableName.valueOf(tablename);
    143.         Table table = connection.getTable(name);
    144.         Delete d = new Delete(rowkey.getBytes()).addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
    145.         table.delete(d);
    146.     }
    147.  
    148.     /**
    149.      * 查找一行记录
    150.      *
    151.      * @param tablename 表名
    152.      * @param rowKey    行名
    153.      */
    154.     public static String selectRow(String tablename, String rowKey) throws IOException {
    155.         String record = "";
    156.         TableName name = TableName.valueOf(tablename);
    157.         Table table = connection.getTable(name);
    158.         Get g = new Get(rowKey.getBytes());
    159.         Result rs = table.get(g);
    160.         NavigableMap<byte[], NavigableMap<byte[], NavigableMapbyte[]>>> map = rs.getMap();
    161.         for (Cell cell : rs.rawCells()) {
    162.             StringBuffer stringBuffer = new StringBuffer().append(Bytes.toString(cell.getRowArray())).append("\t")
    163.                     .append(Bytes.toString(cell.getFamilyArray())).append("\t")
    164.                     .append(Bytes.toString(cell.getQualifierArray())).append("\t")
    165.                     .append(Bytes.toString(cell.getValueArray())).append("\n");
    166.             String str = stringBuffer.toString();
    167.             record += str;
    168.         }
    169.         return record;
    170.     }
    171.  
    172.     /**
    173.      * 查找单行单列族单列记录
    174.      *
    175.      * @param tablename    表名
    176.      * @param rowKey       行名
    177.      * @param columnFamily 列族名
    178.      * @param column       列名
    179.      * @return
    180.      */
    181.     public static String selectValue(String tablename, String rowKey, String columnFamily, String column) throws IOException {
    182.         TableName name = TableName.valueOf(tablename);
    183.         Table table = connection.getTable(name);
    184.         Get g = new Get(rowKey.getBytes());
    185.         g.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
    186.         Result rs = table.get(g);
    187.         return Bytes.toString(rs.value());
    188.     }
    189.  
    190.     /**
    191.      * 查询表中所有行（Scan方式）
    192.      *
    193.      * @param tablename
    194.      * @return
    195.      */
    196.     public String scanAllRecord(String tablename) throws IOException {
    197.         String record = "";
    198.         TableName name = TableName.valueOf(tablename);
    199.         Table table = connection.getTable(name);
    200.         Scan scan = new Scan();
    201.         ResultScanner scanner = table.getScanner(scan);
    202.         try {
    203.             for (Result result : scanner) {
    204.                 for (Cell cell : result.rawCells()) {
    205.                     StringBuffer stringBuffer = new StringBuffer().append(Bytes.toString(cell.getRowArray())).append("\t")
    206.                             .append(Bytes.toString(cell.getFamilyArray())).append("\t")
    207.                             .append(Bytes.toString(cell.getQualifierArray())).append("\t")
    208.                             .append(Bytes.toString(cell.getValueArray())).append("\n");
    209.                     String str = stringBuffer.toString();
    210.                     record += str;
    211.                 }
    212.             }
    213.         } finally {
    214.             if (scanner != null) {
    215.                 scanner.close();
    216.             }
    217.         }
    218.         return record;
    219.     }
    220.  
    221.     /**
    222.      * 根据rowkey关键字查询报告记录
    223.      *
    224.      * @param tablename
    225.      * @param rowKeyword
    226.      * @return
    227.      */
    228.     public List scanReportDataByRowKeyword(String tablename, String rowKeyword) throws IOException {
    229.         ArrayList list = new ArrayList<>();
    230.  
    231.         Table table = connection.getTable(TableName.valueOf(tablename));
    232.         Scan scan = new Scan();
    233.  
    234.         //添加行键过滤器，根据关键字匹配
    235.         RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator(rowKeyword));
    236.         scan.setFilter(rowFilter);
    237.  
    238.         ResultScanner scanner = table.getScanner(scan);
    239.         try {
    240.             for (Result result : scanner) {
    241.                 //TODO 此处根据业务来自定义实现
    242.                 list.add(null);
    243.             }
    244.         } finally {
    245.             if (scanner != null) {
    246.                 scanner.close();
    247.             }
    248.         }
    249.         return list;
    250.     }
    251.  
    252.     /**
    253.      * 根据rowkey关键字和时间戳范围查询报告记录
    254.      *
    255.      * @param tablename
    256.      * @param rowKeyword
    257.      * @return
    258.      */
    259.     public List scanReportDataByRowKeywordTimestamp(String tablename, String rowKeyword, Long minStamp, Long maxStamp) throws IOException {
    260.         ArrayList list = new ArrayList<>();
    261.  
    262.         Table table = connection.getTable(TableName.valueOf(tablename));
    263.         Scan scan = new Scan();
    264.         //添加scan的时间范围
    265.         scan.setTimeRange(minStamp, maxStamp);
    266.  
    267.         RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator(rowKeyword));
    268.         scan.setFilter(rowFilter);
    269.  
    270.         ResultScanner scanner = table.getScanner(scan);
    271.         try {
    272.             for (Result result : scanner) {
    273.                 //TODO 此处根据业务来自定义实现
    274.                 list.add(null);
    275.             }
    276.         } finally {
    277.             if (scanner != null) {
    278.                 scanner.close();
    279.             }
    280.         }
    281.         return list;
    282.     }
    283.  
    284.     /**
    285.      * 删除表操作
    286.      *
    287.      * @param tablename
    288.      */
    289.     public void deleteTable(String tablename) throws IOException {
    290.         TableName name = TableName.valueOf(tablename);
    291.         if (admin.tableExists(name)) {
    292.             admin.disableTable(name);
    293.             admin.deleteTable(name);
    294.         }
    295.     }
    296. }

    五、使用

            接下来只需要在项目业务类里注入hbaseUtils就可以使用了：

    1. @Autowired
    2. private HbaseUtil hbaseUtil;

            测试方法：

    1. import com.swkj.common.hbase.utils.HbaseUtil;
    2. import org.junit.Test;
    3. import org.junit.runner.RunWith;
    4. import org.springframework.beans.factory.annotation.Autowired;
    5. import org.springframework.boot.test.context.SpringBootTest;
    6. import org.springframework.test.context.ActiveProfiles;
    7. import org.springframework.test.context.junit4.SpringRunner;
    8.  
    9. /**
    10.  * @description: Hbase工具类测试
    11.  * @author: zhangzhixiang
    12.  * @createDate: 2022/11/24
    13.  * @version: 1.0
    14.  */
    15. @RunWith(SpringRunner.class)
    16. @SpringBootTest
    17. @ActiveProfiles(profiles = "local")
    18. public class HbaseServiceTest {
    19.  
    20. @Autowired
    21. private HbaseUtil hbaseUtil;
    22.  
    23.     @Test
    24.     public void testHbase() {
    25.         try {
    26.             hbaseUtil.createTable("Student", new String[]{"StuInfo", "Grades"});
    27.             hbaseUtil.insertOneRecord("Student", "0001", "StuInfo", "name", "Tom Green");
    28.             hbaseUtil.insertOneRecord("Student", "0002", "StuInfo", "Age", "18");
    29.             System.out.println("=================" + hbaseUtil.selectValue("Student", "0001", "StuInfo", "name"));
    30.             System.out.println("=================" + hbaseUtil.selectValue("Student", "0002", "StuInfo", "Age"));
    31.             System.out.println("=================" + hbaseUtil.selectRow("Student", "0001"));
    32.             System.out.println("=================" + hbaseUtil.selectRow("Student", "0002"));
    33.         } catch (Exception e) {
    34.             e.printStackTrace();
    35.         }
    36.     }
    37. }

            到此 SpringBoot 封装 HBase 操作工具类介绍完成。

  • 相关阅读:
    手动从0搭建ABP框架-ABP官方完整解决方案和手动搭建简化解决方案实践
    springboot集成nacos作配置中心，动态配置不生效
    Redis实现微博好友功能微服务（关注，取关，共同关注）
    如何在用pip配置文件设置HTTP爬虫IP
    机器学习与图像识别（二）—— OpenCV环境折腾。。
    华为机试题刷题总结
    怎样做ChatGPT应用开发？
    【开发教程10】疯壳·开源蓝牙智能健康手表-OTA镜像制作及下载技术文档
    记将一个大型客户端应用项目迁移到 dotnet 6 的经验和决策
    【夯实算法基础】最近公共祖先
  • 原文地址：https://blog.csdn.net/qq_19734597/article/details/128080643