Hadoop 之 Hbase 配置与使用（四）

àì夳堔傛蜴生んèń 2023-10-13 19:46 123阅读 0赞

Hadoop 之 Hbase 配置与使用

  • 一.Hbase 下载
    • 1.Hbase 下载
  • 二.Hbase 配置
    • 1.单机部署
    • 2.伪集群部署（基于单机配置）
    • 3.集群部署
      • 1.启动 hadoop 集群
      • 2.启动 zookeeper 集群
      • 3.启动 hbase 集群
      • 4.集群启停脚本
  • 三.测试
    • 1.Pom 配置
    • 2.Yml 配置
    • 3.Hbase 配置类
    • 4.Hbase 连接池配置
    • 5.测试类
    • 6.启动类
    • 7.测试

一.Hbase 下载

HBase 是一个分布式的、面向列的开源数据库：Hbase API

1.Hbase 下载

Hbase 下载

在这里插入图片描述

跳转到下载链接

在这里插入图片描述

二.Hbase 配置

1.单机部署

  1. ## 1.创建安装目录
  2. mkdir -p /usr/local/hbase
  3. ## 2.将压缩包拷贝到虚拟机并解压缩
  4. tar zxvf hbase-3.0.0-alpha-4-bin.tar.gz -C /usr/local/hbase/
  5. ## 3.添加环境变量
  6. echo 'export HBASE_HOME=/usr/local/hbase/hbase-3.0.0-alpha-4' >> /etc/profile
  7. echo 'export PATH=${HBASE_HOME}/bin:${PATH}' >> /etc/profile
  8. source /etc/profile
  9. ## 4.指定 JDK 版本
  10. echo 'export JAVA_HOME=/usr/local/java/jdk-11.0.19' >> $HBASE_HOME/conf/hbase-env.sh
  11. ## 5.创建 hbase 存储目录
  12. mkdir -p /home/hbase/data
  13. ## 6.修改配置
  14. vim $HBASE_HOME/conf/hbase-site.xml
  15.      添加如下信息
  16.      <property>
  17.         <name>hbase.rootdir</name>
  18.         <value>file:///home/hbase/data</value>
  19.      </property>

在这里插入图片描述

  1. ## 1.进入安装目录
  2. cd $HBASE_HOME
  3. ## 2.启动服务
  4. ./bin/start-hbase.sh

在这里插入图片描述

  1. ## 1.进入安装目录
  2. cd $HBASE_HOME
  3. ## 2.关闭服务
  4. ./bin/stop-hbase.sh

在这里插入图片描述

2.伪集群部署（基于单机配置）

  1. ## 1.修改 hbase-env.sh 
  2. echo 'export JAVA_HOME=/usr/local/java/jdk-11.0.19' >> $HBASE_HOME/conf/hbase-env.sh
  3. echo 'export HBASE_MANAGES_ZK=true' >> $HBASE_HOME/conf/hbase-env.sh
  4. ## 2.修改 hbase_site.xml
  5. vim $HBASE_HOME/conf/hbase-site.xml
  6.      <!-- 将 hbase 数据保存到 hdfs -->
  7.      <property>
  8.         <name>hbase.rootdir</name>
  9.         <value>hdfs://nn/hbase</value>
  10.      </property>
  11.      <!-- 分布式配置 -->
  12.      <property>
  13.         <name>hbase.cluster.distributed</name>
  14.         <value>true</value>
  15.      </property>
  16.      <!-- 配置 ZK 地址 -->
  17.      <property>
  18.         <name>hbase.zookeeper.quorum</name>
  19.         <value>nn</value>
  20.      </property>
  21.      <!-- 配置 JK 地址 -->
  22.      <property>
  23.         <name>dfs.replication</name>
  24.         <value>1</value>
  25.      </property>
  26. ## 3.修改 regionservers 的 localhost 为 nn
  27. echo nn > $HBASE_HOME/conf/regionservers

在这里插入图片描述

  1. ## 1.进入安装目录
  2. cd $HADOOP_HOME
  3. ## 2.启动 hadoop 服务
  4. ./sbin/start-all.sh

在这里插入图片描述

  1. ## 1.进入安装目录
  2. cd $HBASE_HOME
  3. ## 2.启动服务
  4. ./bin/start-hbase.sh

在这里插入图片描述

  1. ## 1.进入安装目录
  2. cd $HBASE_HOME
  3. ## 2.关闭主节点服务（直接关服务是关不掉的，如图）
  4. . bin/hbase-daemon.sh stop master
  5. ## 3.关闭服务
  6. ./bin/stop-hbase.sh

在这里插入图片描述

3.集群部署

  1. ## 1.创建 zookeeper 数据目录
  2. mkdir -p $HBASE_HOME/zookeeper/data
  3. ## 2.进入安装目录
  4. cd $HBASE_HOME/conf
  5. ## 3.修改环境配置
  6. vim hbase-env.sh
  7. ## 添加 JDK / 启动外置 Zookeeper
  8.     # JDK
  9.     export JAVA_HOME=/usr/local/java/jdk-11.0.19
  10.     # Disable Zookeeper
  11.     export HBASE_MANAGES_ZK=false
  12. ## 4.修改 hbase-site.xml
  13. vim hbase-site.xml
  14. ## 配置如下信息
  15.     <!--允许的最大同步时钟偏移-->
  16.     <property>
  17.       <name>hbase.master.maxclockskew</name>`
  18.       <value>6000</value>
  19.     </property>
  20.     <!--配置 HDFS 存储实例-->
  21.     <property>
  22.       <name>hbase.rootdir</name>
  23.       <value>hdfs://nn:9000/hbase</value>
  24.     </property>
  25.     <!--启用分布式配置-->
  26.     <property>
  27.       <name>hbase.cluster.distributed</name>
  28.       <value>true</value>
  29.     </property>
  30.     <!--配置 zookeeper 集群节点-->
  31.     <property>
  32.       <name>hbase.zookeeper.quorum</name>
  33.       <value>zk1,zk2,zk3</value>
  34.     </property>
  35.     <!--配置 zookeeper 数据目录-->
  36.     <property>
  37.       <name>hbase.zookeeper.property.dataDir</name>
  38.       <value>/usr/local/hbase/hbase-3.0.0-alpha-4/zookeeper/data</value>
  39.     </property>
  40.     <!-- Server is not running yet -->
  41.     <property>
  42.       <name>hbase.wal.provider</name>
  43.       <value>filesystem</value>
  44.     </property>
  45. ## 5.清空 regionservers 并添加集群节点域名
  46. echo '' > regionservers
  47. echo 'nn' >> regionservers
  48. echo 'nd1' >> regionservers
  49. echo 'nd2' >> regionservers
  50. ## 6.分别为 nd1 / nd2 创建 hbase 目录
  51. mkdir -p /usr/local/hbase
  52. ## 7.分发 hbase 配置到另外两台虚拟机 nd1 / nd2
  53. scp -r /usr/local/hbase/hbase-3.0.0-alpha-4 root@nd1:/usr/local/hbase
  54. scp -r /usr/local/hbase/hbase-3.0.0-alpha-4 root@nd2:/usr/local/hbase
  55. ## 8.分发环境变量配置
  56. scp /etc/profile root@nd1:/etc/profile
  57. scp /etc/profile root@nd2:/etc/profile

1.启动 hadoop 集群

Hadoop 集群搭建参考：Hadoop 搭建

  1. ## 1.启动 hadoop
  2. cd $HADOOP_HOME
  3. . sbin/start-all.sh

在这里插入图片描述

  1. ## 1.关闭 hadoop 安全模式
  2. hadoop dfsadmin -safemode leave

在这里插入图片描述

2.启动 zookeeper 集群

ZOOKEEPER 集群搭建说明

  1. ## 1.启动 zookeeper 集群
  2. zkServer.sh start && ssh root@zk2 "source /etc/profile && zkServer.sh start && exit" && ssh root@zk3 "source /etc/profile && zkServer.sh start && exit"
  3. ## 2.查看状态
  4. zkServer.sh status && ssh root@zk2 "source /etc/profile && zkServer.sh status && exit" && ssh root@zk3 "source /etc/profile && zkServer.sh status && exit"

在这里插入图片描述

3.启动 hbase 集群

  1. ## 1.分别为 nn /nd1 / nd2 配置 zookeeper 域名解析
  2. echo '192.168.1.100 zk1' >> /etc/hosts
  3. echo '192.168.1.101 zk2' >> /etc/hosts
  4. echo '192.168.1.102 zk3' >> /etc/hosts
  5. ## 2.启动 habase
  6. cd $HBASE_HOME
  7. . bin/start-hbase.sh
  8. ## 3.停止服务
  9. . bin/hbase-daemon.sh stop master
  10. . bin/hbase-daemon.sh stop regionserver
  11. . bin/stop-hbase.sh

在这里插入图片描述

查看 UI 监控：http://192.168.1.6:16010/master-status

在这里插入图片描述

4.集群启停脚本

  1. #!/bin/bash
  3. case $1 in
  4. "start")
  5.     ## start hadoop
  6.     start-all.sh
  7.     ## start zookeeper (先配置免密登录)
  8.     zkServer.sh start && ssh root@zk2 "source /etc/profile && zkServer.sh start && exit" && ssh root@zk3 "source /etc/profile && zkServer.sh start && exit"
  9.     ## start hbase
  10.     start-hbase.sh
  11.     ;;
  12. "stop")
  13.     ## stop hbase
  14.     ssh root@nd1 "source /etc/profile && hbase-daemon.sh stop regionserver && stop-hbase.sh && exit"
  15.     ssh root@nd2 "source /etc/profile && hbase-daemon.sh stop regionserver && stop-hbase.sh && exit"
  16.     hbase-daemon.sh stop master  && hbase-daemon.sh stop regionserver && stop-hbase.sh
  17.     ## stop zookeeper
  18.     zkServer.sh stop && ssh root@zk2 "source /etc/profile && zkServer.sh stop && exit" && ssh root@zk3 "source /etc/profile && zkServer.sh stop && exit"
  19.     ## stop hadoop
  20.     stop-all.sh
  21.     ;;
  22. *)
  23.     echo "pls inout start|stop"
  24.     ;;
  25. esac

三.测试

  1. ## 1.为 Windows 增加 Hosts 配置，添加 Hbase 集群域名解析 编辑如下文件
  2. C:\Windows\System32\drivers\etc\hosts
  3. ## 2.增加如下信息
  4. 192.168.1.6 nn
  5. 192.168.1.7 nd1
  6. 192.168.1.8 nd2

测试配置效果

在这里插入图片描述

JDK 版本

在这里插入图片描述

工程结构

在这里插入图片描述

1.Pom 配置

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <modelVersion>4.0.0</modelVersion>
  7.     <groupId>org.example</groupId>
  8.     <artifactId>hbase-demo</artifactId>
  9.     <version>1.0-SNAPSHOT</version>
  11.     <properties>
  12.         <maven.compiler.source>11</maven.compiler.source>
  13.         <maven.compiler.target>11</maven.compiler.target>
  14.         <spring.version>2.7.8</spring.version>
  15.         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  16.     </properties>
  18.     <dependencies>
  19.         <dependency>
  20.             <groupId>org.springframework.boot</groupId>
  21.             <artifactId>spring-boot-starter-web</artifactId>
  22.             <version>${spring.version}</version>
  23.         </dependency>
  24.         <dependency>
  25.             <groupId>org.projectlombok</groupId>
  26.             <artifactId>lombok</artifactId>
  27.             <version>1.18.28</version>
  28.         </dependency>
  29.         <dependency>
  30.             <groupId>com.alibaba</groupId>
  31.             <artifactId>fastjson</artifactId>
  32.             <version>2.0.32</version>
  33.         </dependency>
  34.         <dependency>
  35.             <groupId>org.apache.hbase</groupId>
  36.             <artifactId>hbase-client</artifactId>
  37.             <version>3.0.0-alpha-4</version>
  38.         </dependency>
  41.     </dependencies>
  43. </project>

2.Yml 配置

  1. hbase:
  2.   zookeeper:
  3.     quorum: 192.168.1.100,192.168.1.101,192.168.1.102
  4.     property:
  5.       clientPort: 2181
  6.   master:
  7.     ip: 192.168.1.6
  8.     port: 16000

3.Hbase 配置类

  1. package org.example.config;
  4. import org.apache.hadoop.hbase.HBaseConfiguration;
  5. import org.springframework.beans.factory.annotation.Value;
  6. import org.springframework.context.annotation.Bean;
  7. import org.springframework.context.annotation.Configuration;
  9. /**
  10.  * @author Administrator
  11.  * @Description
  12.  * @create 2023-07-25 0:26
  13.  */
  14. @Configuration
  15. public class HbaseConfig {
  18.     @Value("${hbase.zookeeper.quorum}")
  19.     private String zookeeperQuorum;
  21.     @Value("${hbase.zookeeper.property.clientPort}")
  22.     private String clientPort;
  24.     @Value("${hbase.master.ip}")
  25.     private String ip;
  27.     @Value("${hbase.master.port}")
  28.     private int masterPort;
  30.     @Bean
  31.     public org.apache.hadoop.conf.Configuration hbaseConfiguration(){
  33.         org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
  34.         conf.set("hbase.zookeeper.quorum",zookeeperQuorum);
  35.         conf.set("hbase.zookeeper.property.clientPort",clientPort);
  36.         conf.set("hbase.masters", ip + ":" + masterPort);
  37.         conf.set("hbase.client.keyvalue.maxsize","20971520");
  38.         return HBaseConfiguration.create(conf);
  39.     }
  41. }

4.Hbase 连接池配置

  1. package org.example.config;
  3. import lombok.Data;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.apache.hadoop.conf.Configuration;
  6. import org.apache.hadoop.hbase.client.Connection;
  7. import org.apache.hadoop.hbase.client.ConnectionFactory;
  8. import org.springframework.stereotype.Component;
  10. import javax.annotation.PostConstruct;
  11. import javax.annotation.Resource;
  12. import java.util.Enumeration;
  13. import java.util.Vector;
  15. /**
  16.  * @author Administrator
  17.  * @Description
  18.  * @create 2023-07-25 22:39
  19.  */
  20. @Slf4j
  21. @Component
  22. public class HbaseConnectionPool {
  25.     /**
  26.      * 连接池的初始大小
  27.      * 连接池的创建步长
  28.      * 连接池最大的大小
  29.      */
  30.     private int nInitConnectionAmount = 3;
  31.     private int nIncrConnectionAmount = 3;
  32.     private int nMaxConnections = 20;
  34.     /**
  35.      * 存放连接池中数据库连接的向量
  36.      */
  37.     private Vector vcConnections = new Vector();
  39.     /**
  40.      * 注入连接配置
  41.      */
  42.     @Resource
  43.     private Configuration hbaseConfiguration;
  45.     /**
  46.      * 初始化连接
  47.      */
  48.     @PostConstruct
  49.     public void init() {
  51.         createConnections(nInitConnectionAmount);
  52.     }
  54.     /**
  55.      * 获取可用连接
  56.      * @return
  57.      */
  58.     public synchronized Connection getConnection() {
  60.         Connection conn;
  61.         while (null == (conn =getFreeConnection())){
  63.             try {
  65.                 wait(1000);
  66.             } catch (InterruptedException e) {
  68.                 e.printStackTrace();
  69.             }
  70.         }
  71.         // 返回获得的可用的连接
  72.         return conn;
  73.     }
  75.     /**
  76.      * 释放连接
  77.      * @param conn
  78.      */
  79.     public synchronized void releaseConnection(Connection conn) {
  81.         ConnectionWrapper connWrapper;
  82.         Enumeration enumerate = this.vcConnections.elements();
  83.         while(enumerate.hasMoreElements()) {
  85.             connWrapper = (ConnectionWrapper) enumerate.nextElement();
  86.             if (conn == connWrapper.getConnection()) {
  88.                 connWrapper.setBusy(false);
  89.                 break;
  90.             }
  91.         }
  92.     }
  94.     /**
  95.      * 获取可用连接 当前无可用连接则创建 如果已达到最大连接数则返回 null 阻塞后重试获取
  96.      * @return
  97.      */
  98.     private Connection getFreeConnection() {
  100.         Connection conn;
  101.         if (null == (conn = findFreeConnection())) {
  103.             // 创建新连接
  104.             createConnections(nIncrConnectionAmount);
  105.             // 查看是否有可用连接
  106.             if (null == (conn = findFreeConnection())) {
  108.                 return null;
  109.             }
  110.         }
  111.         return conn;
  112.     }
  114.     /**
  115.      * 查找可用连接
  116.      * @return
  117.      */
  118.     private Connection findFreeConnection() {
  120.         ConnectionWrapper connWrapper;
  121.         //遍历向量内连接对象
  122.         Enumeration enumerate = vcConnections.elements();
  123.         while (enumerate.hasMoreElements()) {
  125.             connWrapper = (ConnectionWrapper) enumerate.nextElement();
  126.             //判断当前连接是否被占用
  127.             if (!connWrapper.isBusy()) {
  129.                 connWrapper.setBusy(true);
  130.                 return connWrapper.getConnection();
  131.             }
  132.         }
  133.         // 返回 NULL
  134.         return null;
  135.     }
  137.     /**
  138.      * 创建新连接
  139.      * @param counts
  140.      */
  141.     private void createConnections(int counts) {
  143.         // 循环创建指定数目的数据库连接
  144.         try {
  146.             for (int i = 0; i < counts; i++) {
  148.                 if (this.nMaxConnections > 0  && this.vcConnections.size() >= this.nMaxConnections) {
  150.                     log.warn("已达到最大连接数...");
  151.                     break;
  152.                 }
  153.                 // 创建一个新连接并加到向量
  154.                 vcConnections.addElement(new ConnectionWrapper(newConnection()));
  155.             }
  156.         } catch (Exception e) {
  158.             log.error("创建连接失败...");
  159.         }
  160.     }
  162.     /**
  163.      * 创建新连接
  164.      * @return
  165.      */
  166.     private Connection newConnection() {
  168.         /** hbase 连接 */
  169.         Connection conn = null;
  170.         // 创建一个数据库连接
  171.         try {
  173.             conn = ConnectionFactory.createConnection(hbaseConfiguration);
  174.         } catch (Exception e) {
  176.             log.error("HBase 连接失败...");
  177.         }
  179.         // 返回创建的新的数据库连接
  180.         return conn;
  181.     }
  183.     /**
  184.      * 封装连接对象
  185.      */
  186.     @Data
  187.     class ConnectionWrapper {
  190.         /**
  191.          * 数据库连接
  192.          */
  193.         private Connection connection;
  195.         /**
  196.          * 此连接是否正在使用的标志，默认没有正在使用
  197.          */
  198.         private boolean busy = false;
  200.         /**
  201.          * 构造函数，根据一个 Connection 构告一个 PooledConnection 对象
  202.          */
  203.         public ConnectionWrapper(Connection connection) {
  205.             this.connection = connection;
  206.         }
  207.     }
  209. }

5.测试类

  1. package org.example.controller;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.apache.hadoop.hbase.Cell;
  5. import org.apache.hadoop.hbase.CompareOperator;
  6. import org.apache.hadoop.hbase.TableName;
  7. import org.apache.hadoop.hbase.client.*;
  8. import org.apache.hadoop.hbase.filter.ColumnValueFilter;
  9. import org.apache.hadoop.hbase.filter.Filter;
  10. import org.apache.hadoop.hbase.util.Bytes;
  11. import org.example.config.HbaseConnectionPool;
  12. import org.springframework.web.bind.annotation.GetMapping;
  13. import org.springframework.web.bind.annotation.RequestMapping;
  14. import org.springframework.web.bind.annotation.RestController;
  16. import javax.annotation.Resource;
  17. import java.io.IOException;
  18. import java.util.*;
  20. /**
  21.  * @author Administrator
  22.  *
  23.  * 可利用 aop 进行连接获取和释放处理
  24.  *
  25.  * @Description
  26.  * @create 2023-07-25 23:06
  27.  */
  28. @Slf4j
  29. @RestController
  30. @RequestMapping("/hbase")
  31. public class HbaseController {
  34.     @Resource
  35.     private HbaseConnectionPool pool;
  37.     /**
  38.      * 表名
  39.      */
  40.     private String tbl_user = "tbl_user";
  42.     /**
  43.      * 创建表(不允许重复创建)
  44.      */
  45.     @GetMapping("/create")
  46.     public void createTable(){
  48.         Connection conn = null;
  49.         //获取连接
  50.         try {
  52.             conn = pool.getConnection();
  53.             Admin admin = conn.getAdmin();
  54.             TableName tableName = TableName.valueOf(tbl_user);
  55.             if (!admin.tableExists(tableName)){
  57.                 //指定表名
  58.                 TableDescriptorBuilder tdb_user = TableDescriptorBuilder.newBuilder(tableName);
  59.                 //添加列族(info,data)
  60.                 ColumnFamilyDescriptor hcd_info = ColumnFamilyDescriptorBuilder.of("name");
  61.                 ColumnFamilyDescriptor hcd_data = ColumnFamilyDescriptorBuilder.of("age");
  62.                 tdb_user.setColumnFamily(hcd_info);
  63.                 tdb_user.setColumnFamily(hcd_data);
  64.                 //创建表
  65.                 TableDescriptor td = tdb_user.build();
  66.                 admin.createTable(td);
  67.             }
  68.         } catch (IOException e) {
  70.             throw new RuntimeException(e);
  71.         } finally {
  73.             if (null != conn){
  75.                 pool.releaseConnection(conn);
  76.             }
  77.         }
  78.     }
  80.     /**
  81.      * 删除表(不允许删除不存在的表)
  82.      */
  83.     @GetMapping("/drop")
  84.     public void dropTable(){
  86.         Connection conn = null;
  87.         try {
  89.             conn = pool.getConnection();
  90.             Admin admin = conn.getAdmin();
  91.             TableName tableName = TableName.valueOf(tbl_user);
  92.             if (admin.tableExists(tableName)){
  94.                 admin.disableTable(tableName);
  95.                 admin.deleteTable(tableName);
  96.             }
  97.         } catch (IOException e) {
  99.             throw new RuntimeException(e);
  100.         } finally {
  102.             if (null != conn){
  104.                 pool.releaseConnection(conn);
  105.             }
  106.         }
  107.     }
  109.     /**
  110.      * 插入测试
  111.      */
  112.     @GetMapping("/insert")
  113.     public void insert(){
  115.         log.info("---插入一列数据---1");
  116.         putData(tbl_user, "row1", "name", "a", "zhangSan");
  117.         putData(tbl_user, "row1", "age", "a", "18");
  118.         log.info("---插入多列数据---2");
  119.         putData(tbl_user, "row2", "name",
  120.                 Arrays.asList("a", "b", "c"),  Arrays.asList("liSi", "wangWu", "zhaoLiu"));
  121.         log.info("---插入多列数据---3");
  122.         putData(tbl_user, "row3", "age",
  123.                 Arrays.asList("a", "b", "c"),  Arrays.asList("18","19","20"));
  124.         log.info("---插入多列数据---4");
  125.         putData(tbl_user, "row4", "age",
  126.                 Arrays.asList("a", "b", "c"),  Arrays.asList("30","19","20"));
  128.     }
  130.     /**
  131.      * 插入数据（单条）
  132.      * @param tableName 表名
  133.      * @param rowKey rowKey
  134.      * @param columnFamily 列族
  135.      * @param column 列
  136.      * @param value 值
  137.      * @return true/false
  138.      */
  139.     public boolean putData(String tableName, String rowKey, String columnFamily, String column,
  140.                            String value) {
  142.         return putData(tableName, rowKey, columnFamily, Arrays.asList(column),
  143.                 Arrays.asList(value));
  144.     }
  146.     /**
  147.      * 插入数据（批量）
  148.      * @param tableName 表名
  149.      * @param rowKey rowKey
  150.      * @param columnFamily 列族
  151.      * @param columns 列
  152.      * @param values 值
  153.      * @return true/false
  154.      */
  155.     public boolean putData(String tableName, String rowKey, String columnFamily,
  156.                            List<String> columns, List<String> values) {
  158.         Connection conn = null;
  159.         try {
  161.             conn = pool.getConnection();
  162.             Table table = conn.getTable(TableName.valueOf(tableName));
  163.             Put put = new Put(Bytes.toBytes(rowKey));
  164.             for (int i=0; i<columns.size(); i++) {
  166.                 put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columns.get(i)), Bytes.toBytes(values.get(i)));
  167.             }
  168.             table.put(put);
  169.             table.close();
  170.             return true;
  171.         } catch (IOException e) {
  173.             e.printStackTrace();
  174.             return false;
  175.         } finally {
  177.             if (null != conn){
  179.                 pool.releaseConnection(conn);
  180.             }
  181.         }
  182.     }
  184.     /**
  185.      * 查询测试
  186.      */
  187.     @GetMapping("/query")
  188.     public void getResultScanner(){
  190.         log.info("全表数据:{}",getData(tbl_user));
  191.         log.info("过滤器,按年龄 [18]:{}",getData(tbl_user,new ColumnValueFilter(Bytes.toBytes("age"), Bytes.toBytes("a"), CompareOperator.EQUAL, Bytes.toBytes("18"))));
  192.         log.info("根据 rowKey [row1]:{}",getData(tbl_user,"row1"));
  193.         log.info("根据 rowKey 列族 列 [row2 name a]:{}",getData(tbl_user,"row2","name","a"));
  194.     }
  196.     /**
  197.      * 获取数据（全表数据）
  198.      * @param tableName 表名
  199.      * @return map
  200.      */
  201.     public List<Map<String, String>> getData(String tableName) {
  203.         List<Map<String, String>> list = new ArrayList<>();
  204.         Connection conn = null;
  205.         try {
  207.             conn = pool.getConnection();
  208.             Table table = conn.getTable(TableName.valueOf(tableName));
  209.             Scan scan = new Scan();
  210.             ResultScanner resultScanner = table.getScanner(scan);
  211.             for(Result result : resultScanner) {
  213.                 HashMap<String, String> map = new HashMap<>(result.listCells().size());
  214.                 map.put("row", Bytes.toString(result.getRow()));
  215.                 for (Cell cell : result.listCells()) {
  217.                     //列族
  218.                     String family = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
  219.                     //列
  220.                     String qualifier = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
  221.                     //值
  222.                     String data = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
  223.                     map.put(family + ":" + qualifier, data);
  224.                 }
  225.                 list.add(map);
  226.             }
  227.             table.close();
  228.         } catch (IOException e) {
  230.             e.printStackTrace();
  231.         } finally {
  233.             if (null != conn){
  235.                 pool.releaseConnection(conn);
  236.             }
  237.         }
  238.         return list;
  239.     }
  241.     /**
  242.      * 获取数据（根据 filter）
  243.      * @param tableName 表名
  244.      * @param filter 过滤器
  245.      * @return map
  246.      */
  247.     public List<Map<String, String>> getData(String tableName, Filter filter) {
  249.         List<Map<String, String>> list = new ArrayList<>();
  250.         Connection conn = null;
  251.         try {
  253.             conn = pool.getConnection();
  254.             Table table = conn.getTable(TableName.valueOf(tableName));
  255.             Scan scan = new Scan();
  256.             // 添加过滤器
  257.             scan.setFilter(filter);
  258.             ResultScanner resultScanner = table.getScanner(scan);
  259.             for(Result result : resultScanner) {
  261.                 HashMap<String, String> map = new HashMap<>(result.listCells().size());
  262.                 map.put("row", Bytes.toString(result.getRow()));
  263.                 for (Cell cell : result.listCells()) {
  265.                     String family = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
  266.                     String qualifier = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
  267.                     String data = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
  268.                     map.put(family + ":" + qualifier, data);
  269.                 }
  270.                 list.add(map);
  271.             }
  272.             table.close();
  273.         } catch (IOException e) {
  275.             e.printStackTrace();
  276.         } finally {
  278.             if (null != conn){
  280.                 pool.releaseConnection(conn);
  281.             }
  282.         }
  283.         return list;
  284.     }
  286.     /**
  287.      * 获取数据（根据 rowKey）
  288.      * @param tableName 表名
  289.      * @param rowKey rowKey
  290.      * @return map
  291.      */
  292.     public Map<String, String> getData(String tableName, String rowKey) {
  294.         HashMap<String, String> map = new HashMap<>();
  295.         Connection conn = null;
  296.         try {
  298.             conn = pool.getConnection();
  299.             Table table = conn.getTable(TableName.valueOf(tableName));
  300.             Get get = new Get(Bytes.toBytes(rowKey));
  301.             Result result = table.get(get);
  302.             if (result != null && !result.isEmpty()) {
  304.                 for (Cell cell : result.listCells()) {
  306.                     String family = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
  307.                     String qualifier = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
  308.                     String data = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
  309.                     map.put(family + ":" + qualifier, data);
  310.                 }
  311.             }
  312.             table.close();
  313.         } catch (IOException e) {
  315.             e.printStackTrace();
  316.         } finally {
  318.             if (null != conn){
  320.                 pool.releaseConnection(conn);
  321.             }
  322.         }
  323.         return map;
  324.     }
  326.     /**
  327.      * 获取数据（根据 rowKey 列族 列）
  328.      * @param tableName 表名
  329.      * @param rowKey rowKey
  330.      * @param columnFamily 列族
  331.      * @param columnQualifier 列
  332.      * @return map
  333.      */
  334.     public String getData(String tableName, String rowKey, String columnFamily,
  335.                           String columnQualifier) {
  337.         String data = "";
  338.         Connection conn = null;
  339.         try {
  341.             conn = pool.getConnection();
  342.             Table table = conn.getTable(TableName.valueOf(tableName));
  343.             Get get = new Get(Bytes.toBytes(rowKey));
  344.             get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columnQualifier));
  345.             Result result = table.get(get);
  346.             if (result != null && !result.isEmpty()) {
  348.                 Cell cell = result.listCells().get(0);
  349.                 data = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
  350.             }
  351.             table.close();
  352.         } catch (IOException e) {
  354.             e.printStackTrace();
  355.         } finally {
  357.             if (null != conn){
  359.                 pool.releaseConnection(conn);
  360.             }
  361.         }
  362.         return data;
  363.     }
  365.     /**
  366.      * 删除数据
  367.      */
  368.     @GetMapping("/delete")
  369.     public void delete(){
  371.         log.info("---删除 rowKey --- row1 ");
  372.         deleteData(tbl_user,"row1");
  373.         log.info("---删除 rowKey 列族 --- row2 age ");
  374.         deleteData(tbl_user,"row2","age");
  375.     }
  377.     /**
  378.      * 删除数据（根据 rowKey）
  379.      * @param tableName 表名
  380.      * @param rowKey rowKey
  381.      */
  382.     public void deleteData(String tableName, String rowKey) {
  384.         Connection conn = null;
  385.         try {
  387.             conn = pool.getConnection();
  388.             Table table = conn.getTable(TableName.valueOf(tableName));
  389.             Delete delete = new Delete(Bytes.toBytes(rowKey));
  390.             table.delete(delete);
  391.             table.close();
  392.         } catch (IOException e) {
  394.             e.printStackTrace();
  395.         } finally {
  397.             if (null != conn){
  399.                 pool.releaseConnection(conn);
  400.             }
  401.         }
  402.     }
  404.     /**
  405.      * 删除数据（根据 row key，列族）
  406.      * @param tableName 表名
  407.      * @param rowKey rowKey
  408.      * @param columnFamily 列族
  409.      */
  410.     public void deleteData(String tableName, String rowKey, String columnFamily) {
  412.         Connection conn = null;
  413.         try {
  415.             conn = pool.getConnection();
  416.             Table table = conn.getTable(TableName.valueOf(tableName));
  417.             Delete delete = new Delete(Bytes.toBytes(rowKey));
  418.             delete.addFamily(columnFamily.getBytes());
  419.             table.delete(delete);
  420.             table.close();
  421.         } catch (IOException e) {
  423.             e.printStackTrace();
  424.         } finally {
  426.             if (null != conn){
  428.                 pool.releaseConnection(conn);
  429.             }
  430.         }
  431.     }
  433. }

6.启动类

  1. package org.example;
  4. import org.springframework.boot.SpringApplication;
  5. import org.springframework.boot.autoconfigure.SpringBootApplication;
  7. /**
  8.  * @author Administrator
  9.  */
  10. @SpringBootApplication
  11. public class HbaseDemo {
  13.     public static void main(String[] args) {
  15.         SpringApplication.run(HbaseDemo.class,args);
  16.     }
  17. }

7.测试

  1. 创建表：http://127.0.0.1:8080/hbase/create
  2. 插入：http://127.0.0.1:8080/hbase/insert
  3. 查询：http://127.0.0.1:8080/hbase/query
  4. 删除：http://127.0.0.1:8080/hbase/delete
  5. 删除表：http://127.0.0.1:8080/hbase/drop

查看 UI

在这里插入图片描述

测试输出日志

在这里插入图片描述

发表评论

表情：
评论列表 （有 0 条评论，123人围观）

还没有评论，来说两句吧...

相关阅读