Flink电商指标分析项目(3)_HBaseUtil工具类开发

冷不防 2022-11-06 01:53 181阅读 0赞

### 目录 ###

*   *  HBaseUtil工具类
     *   *  API介绍
         *  获取表
         *  存储数据
         *  获取数据
         *  批量存储数据
         *  批量获取数据
         *  删除数据
         *  最终代码

## HBaseUtil工具类 ##

前面我们实现了Flink整合Kafka，可以从Kafka中获取数据进行分析，分析之后我们要把结果存入HBase中，为了方便操作，我们先提前编写一个操作HBase的工具类。HBase作为一个数据库，我们肯定要进行数据的增删改查，那么我们就围绕这几个操作进行开发。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]

### API介绍 ###

先来看下我们要编写的工具类的所有方法，我们先明确了目标，再进行开发。

<table> 
 <thead> 
  <tr> 
   <th>方法名</th> 
   <th>用途</th> 
   <th>参数说明</th> 
   <th>返回值</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>getTable</td> 
   <td>创建/获取表</td> 
   <td>tableNameStr:表名<br>columnFamily:列族名</td> 
   <td>HBase Table对象</td> 
  </tr> 
  <tr> 
   <td>putData</td> 
   <td>插入/更新一列数据</td> 
   <td>tableNameStr: 表名<br>rowkey:String rowkey<br>columnFamily:列族名<br>column:String 列名<br>data:String 列值</td> 
   <td>空</td> 
  </tr> 
  <tr> 
   <td>putMapData</td> 
   <td>插入/更新多个列数据</td> 
   <td>tableNameStr: 表名<br>rowkey:String rowkey<br>columnFamily:列族名<br>mapData:列名/列值</td> 
   <td>空</td> 
  </tr> 
  <tr> 
   <td>getData</td> 
   <td>根据rowkey，列族+列名<br>获取单列数据</td> 
   <td>tableNameStr: 表名<br>rowkey:String rowkey<br>columnFamily:列族名<br>column:列名</td> 
   <td>列对应的数据<br>String类型</td> 
  </tr> 
  <tr> 
   <td>getMapData</td> 
   <td>根据rowkey，列族+列名集合<br>获取多列数据</td> 
   <td>tableNameStr: 表名<br>rowkey:String rowkey<br>columnFamily:列族名<br>column:列名集合</td> 
   <td>列对应的数据<br>Map[列名, 列值]</td> 
  </tr> 
  <tr> 
   <td>deleteData</td> 
   <td>根据rowkey删除一条数据</td> 
   <td>tableNameStr: 表名<br>rowkey:rowkey<br>columnFamily: 列族名</td> 
   <td>空</td> 
  </tr> 
 </tbody> 
</table>

**HBase操作基本类**

<table> 
 <thead> 
  <tr> 
   <th>类名</th> 
   <th>用途</th> 
   <th>获取方式</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Configuration</td> 
   <td>HBase的配置类</td> 
   <td>HBaseConfiguration.create</td> 
  </tr> 
  <tr> 
   <td>Connection</td> 
   <td>连接</td> 
   <td>ConnectionFactory.createConnection(conf)</td> 
  </tr> 
  <tr> 
   <td>Admin</td> 
   <td>HBase的操作API</td> 
   <td>Connection.getAdmin</td> 
  </tr> 
  <tr> 
   <td>Table</td> 
   <td>用来链接HBase的单表</td> 
   <td>Connection.getTable()</td> 
  </tr> 
  <tr> 
   <td>Get</td> 
   <td>用来查询HBase的单行数据</td> 
   <td>new Get(rowkey.getBytes())</td> 
  </tr> 
  <tr> 
   <td>Put</td> 
   <td>保存单行数据</td> 
   <td>new Put(rowkey.getBytes())</td> 
  </tr> 
  <tr> 
   <td>Delete</td> 
   <td>删除单行数据</td> 
   <td>new Delete(rowkey.getBytes())</td> 
  </tr> 
 </tbody> 
</table>

### 获取表 ###

**开发步骤：**

1.  将导入`hbase-site.xml`配  
    置文件到`resources`目录
2.  在`util`包中添加`HBaseUtil`
    
     *  使用`HBaseConfiguration.create`获取配置对象`Configuration`，该配置对象会自动加载hbase-site.xml
     *  使用`ConnectionFactory.createConnection`获取hbase连接
     *  使用`Connection.getAdmin`获取与master的连接
3.  创建`getTable`方法
    
     *  构建`TableName`
     *  构建`TableDescriptorBuilder`
     *  构建`ColumnFamilyDescriptor`
     *  添加列族
     *  检查表是否存在，若不存在，则创建表  
        **源码解析：**  
        我们可以围绕Admin来展开源码的查看，请观察下图，通过源码的注释，我们可以很简单的找到该编写的代码。  
        ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]  
        **示例代码：**

import org.apache.hadoop.conf.Configuration
    import org.apache.hadoop.hbase.{ HBaseConfiguration, TableName}
    import org.apache.hadoop.hbase.client._
    
    object HBaseUtil { 
    
      // 读取配置文件
      val conf: Configuration = HBaseConfiguration.create
    
      // 创建连接
      val conn: Connection = ConnectionFactory.createConnection(conf)
    
      // 创建HBase管理类
      val admin: Admin = conn.getAdmin
    
      /** * 获得表,如果表不存在则创建 * * @param tableNameStr 表名 * @param columnFamilyName 列族名 * @return */
      def getTable(tableNameStr: String, columnFamilyName: String): Table = { 
    
        // 获取 TableName
        val tableName: TableName = TableName.valueOf(tableNameStr)
    
        //判断表是否可用
        if (!admin.tableExists(tableName)) { 
          // 构建TableDescriptorBuilder
          val tableDescriptor: TableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName)
    
          // 构建列族 ColumnFamilyDescriptor
          val columnFamily: ColumnFamilyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(columnFamilyName.getBytes).build()
          tableDescriptor.setColumnFamily(columnFamily)
          // 创建表
          admin.createTable(tableDescriptor.build())
        }
    
        // 返回表
        conn.getTable(tableName)
      }

### 存储数据 ###

创建`putData`方法

*  调用getTable获取表
 *  构建`Put`对象
 *  添加列、列值
 *  对table执行put操作
 *  启动编写main进行测试

/** * 保存数据 * * @param tableNameStr 表名 * @param rowkey rowkey * @param columnFamilyName 列族名 * @param columnName 列名 * @param columnValue 列值 */
    def putData(tableNameStr: String,
                rowkey: String,
                columnFamilyName: String,
                columnName: String,
                columnValue: String
               ) = { 
        val table = getTable(tableNameStr, columnFamilyName)
    
        try { 
            // Put
            val put = new Put(rowkey.getBytes())
            // 参数1: 列族 参数2: 列名 参数3: 列值
            put.addColumn(columnFamilyName.getBytes, columnName.getBytes, columnValue.getBytes)
            // 保存数据
            table.put(put)
        } catch { 
            case ex: Exception => { 
                ex.printStackTrace()
            }
        } finally { 
            // 关闭table
            table.close()
        }
    }

### 获取数据 ###

1.  使用`Connection`获取表
2.  创建`getData`方法
    
     *  调用getTable获取表
     *  构建`Get`对象
     *  对table执行get操作，获取result
     *  使用Result.getValue获取列族列对应的值
     *  捕获异常
     *  关闭表
3.  启动`hhbase`
4.  启动编写main进行测试

/** * 查询数据 * * @param tableNameStr 表名 * @param rowkey rowkey * @param columnFamilyName 列族名 * @param columnName 列名 */
    def getData(tableNameStr: String,
                rowkey: String,
                columnFamilyName: String,
                columnName: String) = { 
    
        val table: Table = getTable(tableNameStr, columnFamilyName)
    
        try { 
    
            // Get对象
            val get: Get = new Get(rowkey.getBytes)
            // 查询结果
            val result = table.get(get)
            // 结果不为空, 并且结果包含我们要查询的列
            if (result != null && result.containsColumn(columnFamilyName.getBytes, columnName.getBytes)) { 
                // - 使用Result.getValue获取列族列对应的值
                val colValueBytes: Array[Byte] = result.getValue(columnFamilyName.getBytes, columnName.getBytes)
                Bytes.toString(colValueBytes)
            }else { 
            ""
          	}
        } catch { 
            case ex: Exception =>
            ex.printStackTrace()
            ""
        } finally { 
            // - 关闭表
            table.close()
        }
    }

### 批量存储数据 ###

创建`putMapData`方法

*  调用`getTable`获取表
 *  构建`Put`对象
 *  添加Map中的列、列值
 *  对table执行put操作
 *  捕获异常
 *  关闭表
 *  启动编写main进行测试

/** * 批量存储数据 * * @param tableNameStr 表名 * @param rowkey rowkey * @param columnFamilyName 列族名 * @param map key/value集合 */
    def putMapData(tableNameStr: String,
                   rowkey: String,
                   columnFamilyName: String,
                   map: Map[String, Any]
                  ) = { 
        val table = getTable(tableNameStr, columnFamilyName)
    
        try { 
            //- 添加Map中的列、列值
            for ((columnName, columnValue) <- map) { 
                // Put
                val put = new Put(rowkey.getBytes())
                put.addColumn(columnFamilyName.getBytes, columnName.getBytes, columnValue.toString.getBytes)
                table.put(put)
            }
    
        } catch { 
            case ex: Exception => { 
                ex.printStackTrace()
            }
        } finally { 
            table.close()
        }
    }

### 批量获取数据 ###

创建`putMapData`方法

*  调用`getTable`获取表
 *  构建`Get`对象
 *  根据Get对象查询表
 *  构建可变Map
 *  遍历查询各个列的列值
 *  过滤掉不符合的结果
 *  把结果转换为Map返回
 *  捕获异常
 *  关闭表
 *  启动编写main进行测试

/** * 批量获取列的数据 * * @param tableNameStr 表名 * @param rowkey rowkey * @param columnFamily 列族 * @param columnList 列的名字列表 * @return */
    def getMapData(tableNameStr: String, rowkey: String, columnFamily: String, columnList: List[String]): Map[String, String] = { 
        // 根据表名和列族名获取表
        val table: Table = getTable(tableNameStr, columnFamily)
    
        try { 
            // 创建Get对象
            val get: Get = new Get(rowkey.getBytes)
    
            // 根据Get对象查询表
            val result: Result = table.get(get)
    
            // 构建可变Map
            val valueMap = collection.mutable.Map[String, String]()
    
            // 遍历要查询的列集合
            columnList.map { 
                col =>
                // 从表查询结果中,根据列族,列名提取出列值
                val values: Array[Byte] = result.getValue(columnFamily.getBytes, col.getBytes)
    
                // 列值不为空 , 则设置 k-> v
                if (values != null && values.size > 0) { 
                    col -> Bytes.toString(values)
                }
                else { 
                    "" -> ""
                }
            }.filter(_._1 != "").toMap  // 过滤掉key为""的内容,并转换为Map类型
        } catch { 
            case e: Exception =>
            e.printStackTrace()
            Map[String, String]()
        } finally { 
            table.close()
        }
    }

### 删除数据 ###

创建`deleteData`方法

*  调用`getTable`获取表
 *  构建`Delete`对象
 *  对table执行delete操作
 *  捕获异常
 *  关闭表
 *  启动编写main进行测试

/** * 删除数据 * * @param tableNameStr 表名 * @param columnFamilyName 列族名 * @param rowkey rowkey */
      def deleteData(tableNameStr: String,
                     rowkey: String,
                     columnFamilyName: String
                    ) = { 
        // 获取表
        val table = getTable(tableNameStr, columnFamilyName)
        try { 
          // 创建Delete对象
          val delete = new Delete(rowkey.getBytes)
          table.delete(delete)
        } catch { 
          case ex: Exception =>
            ex.printStackTrace()
        } finally { 
          table.close()
        }
      }

**main方法测试代码**

object HBaseUtil { 
        
        def main(args: Array[String]): Unit = { 
        // 测试存入单列数据
        putData("test", "123", "info", "t1", "hello world")
    
        // 测试存入多列数据
        val map = Map(
          "t2" -> "scala",
          "t3" -> "hive",
          "t4" -> "sqoop"
        )
        putMapData("test", "123", "info", map)
    
        println(getData("test", "123", "info", "t1"))
        println(getData("test", "123", "info", "t2"))
        println(getData("test", "123", "info", "t3"))
        println(getData("test", "123", "info", "t4"))
            
        println(getMapData("test", "123", "info", List("t1", "t2","t3","t4")))
    
        deleteData("test", "123", "info")
      }
    
    }

### 最终代码 ###

import org.apache.hadoop.conf.Configuration
    import org.apache.hadoop.hbase.{ HBaseConfiguration, TableName}
    import org.apache.hadoop.hbase.client.{ Admin, ColumnFamilyDescriptorBuilder, Connection, ConnectionFactory, Delete, Get, Put, Result, Table, TableDescriptor, TableDescriptorBuilder}
    import org.apache.hadoop.hbase.util.Bytes
    
    
    /** * HBase工具类 * * 获取Table * 保存单列数据 * 查询单列数据 * 保存多列数据 * 查询多列数据 * 删除数据 */
    object HBaseUtil { 
    
      //HBase的配置类，不需要指定配置文件名，文件名要求是hbase-site.xml
      val conf:Configuration = HBaseConfiguration.create()
    
      //hbase的连接
      val conn:Connection = ConnectionFactory.createConnection()
    
      //hbase的api
      val admin:Admin = conn.getAdmin
    
      /** * 返回table，如果不存在，则创建表 * @param tableNameStr * @param columnFamilyName * @return */
      def getTable(tableNameStr:String,columnFamilyName:String):Table={ 
    
        //获取TableName
        val tableName = TableName.valueOf(tableNameStr)
    
        //如果表不存在，则创建表
        if (!admin.tableExists(tableName)){ 
    
          //构建出 表的描述的建造者
          val descBuilder = TableDescriptorBuilder.newBuilder(tableName)
    
          val familyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(columnFamilyName.getBytes).build()
          //给表去添加列族
          descBuilder.setColumnFamily(familyDescriptor)
    
          //创建表
          admin.createTable(descBuilder.build())
        }
        conn.getTable(tableName)
      }
    
      /** *存储单列数据 * @param tableNameStr 表名 * @param rowkey rowkey * @param columnFamilyName 列族名 * @param columnName 列名 * @param columnValue 列值 */
      def putData(tableNameStr:String,rowkey:String,columnFamilyName:String,columnName:String,columnValue:String): Unit ={ 
    
        //获取表
        val table = getTable(tableNameStr, columnFamilyName)
    
        try{ 
          //Put
          val put = new Put(rowkey.getBytes())
          put.addColumn(columnFamilyName.getBytes(),columnName.getBytes(),columnValue.getBytes())
    
          //保存数据
          table.put(put)
        }catch { 
          case ex:Exception=>{ 
            ex.printStackTrace()
          }
        }finally { 
          table.close()
        }
      }
    
      /** * 通过单列获取列值 * @param tableNameStr 表名 * @param rowkey rowkey * @param columnFamilyName 列族名 * @param columnName 列明 * @return 列值 */
      def getData(tableNameStr: String,
                  rowkey: String,
                  columnFamilyName: String,
                  columnName: String) = { 
    
        val table: Table = getTable(tableNameStr, columnFamilyName)
    
        try { 
    
          // Get对象
          val get: Get = new Get(rowkey.getBytes)
          // 查询结果
          val result = table.get(get)
          // 结果不为空, 并且结果包含我们要查询的列
          if (result != null && result.containsColumn(columnFamilyName.getBytes, columnName.getBytes)) { 
            // - 使用Result.getValue获取列族列对应的值
            val colValueBytes: Array[Byte] = result.getValue(columnFamilyName.getBytes, columnName.getBytes)
            Bytes.toString(colValueBytes)
          }else { 
            ""
          }
        } catch { 
          case ex: Exception =>
            ex.printStackTrace()
            ""
        } finally { 
          // - 关闭表
          table.close()
        }
      }
    
      /** * 批量存储数据 * * @param tableNameStr 表名 * @param rowkey rowkey * @param columnFamilyName 列族名 * @param map key/value集合 */
      def putMapData(tableNameStr: String,
                     rowkey: String,
                     columnFamilyName: String,
                     map: Map[String, Any]
                    ) = { 
        val table = getTable(tableNameStr, columnFamilyName)
    
        try { 
          //- 添加Map中的列、列值
          for ((columnName, columnValue) <- map) { 
            // Put
            val put = new Put(rowkey.getBytes())
            put.addColumn(columnFamilyName.getBytes, columnName.getBytes, columnValue.toString.getBytes)
            table.put(put)
          }
    
        } catch { 
          case ex: Exception => { 
            ex.printStackTrace()
          }
        } finally { 
          table.close()
        }
      }
    
      /** * 删除数据 * * @param tableNameStr 表名 * @param columnFamilyName 列族名 * @param rowkey rowkey */
      def deleteData(tableNameStr: String,
                     rowkey: String,
                     columnFamilyName: String
                    ) = { 
        // 获取表
        val table = getTable(tableNameStr, columnFamilyName)
        try { 
          // 创建Delete对象
          val delete = new Delete(rowkey.getBytes)
          table.delete(delete)
        } catch { 
          case ex: Exception =>
            ex.printStackTrace()
        } finally { 
          table.close()
        }
      }
    
      /** * 批量获取列的数据 * * @param tableNameStr 表名 * @param rowkey rowkey * @param columnFamily 列族 * @param columnList 列的名字列表 * @return */
      def getMapData(tableNameStr: String, rowkey: String, columnFamily: String, columnList: List[String]): Map[String, String] = { 
        // 根据表名和列族名获取表
        val table: Table = getTable(tableNameStr, columnFamily)
    
        try { 
          // 创建Get对象
          val get: Get = new Get(rowkey.getBytes)
    
          // 根据Get对象查询表
          val result: Result = table.get(get)
    
          // 构建可变Map
          val valueMap = collection.mutable.Map[String, String]()
    
          // 遍历要查询的列集合
          columnList.map { 
            col =>
              // 从表查询结果中,根据列族,列名提取出列值
              val values: Array[Byte] = result.getValue(columnFamily.getBytes, col.getBytes)
    
              // 列值不为空 , 则设置 k-> v
              if (values != null && values.size > 0) { 
                col -> Bytes.toString(values)
              }
              else { 
                "" -> ""
              }
          }.filter(_._1 != "").toMap  // 过滤掉key为""的内容,并转换为Map类型
        } catch { 
          case e: Exception =>
            e.printStackTrace()
            Map[String, String]()
        } finally { 
          table.close()
        }
      }
    
      def main(args: Array[String]): Unit = { 
       // println(getTable("test", "info"))
       // putData("test","1","info","t1","hello")
       // println(getData("test", "1", "info", "t1"))
      /* val map = Map( "t2" -> "scala", "t3" -> "hive", "t4" -> "sqoop" ) putMapData("test", "123", "info", map)*/
    
        deleteData("test", "123", "info")
        println(getMapData("test", "123", "info", List("t1", "t2","t3","t4")))
      }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]: /images/20221023/a3c8596d82714beeafd9c51134201050.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221023/0aae2efd507d40b78b5a28dba1035fb2.png