HBase Shell操作&Flink写入HBase

一、HBase Shell操作1、基本操作

1）进入HBase客户端命令行

    [root@bigdata1 hbase]$ bin/hbase shell

2）查看帮助命令

    hbase(main):001:0> help

3）查看当前数据库中有哪些表

    hbase(main):002:0> list

2、表的操作

1）创建表

    hbase(main):002:0> create 'student','info'

2）插入数据到表

    hbase(main):003:0> put 'student','1001','info:sex','male'    hbase(main):004:0> put 'student','1001','info:age','18'    hbase(main):005:0> put 'student','1002','info:name','Janna'    hbase(main):006:0> put 'student','1002','info:sex','female'    hbase(main):007:0> put 'student','1002','info:age','20'

3）扫描查看表数据

    hbase(main):008:0> scan 'student'    hbase(main):009:0> scan 'student',{STARTROW => '1001', STOPROW  => '1001'}    hbase(main):010:0> scan 'student',{STARTROW => '1001'}

4）查看表结构

    hbase(main):011:0> describe 'student'

5）更新指定字段的数据

    hbase(main):012:0> put 'student','1001','info:name','Nick'    hbase(main):013:0> put 'student','1001','info:age','100'

6）查看“指定行”或“指定列族:列”的数据

    hbase(main):014:0> get 'student','1001'    hbase(main):015:0> get 'student','1001','info:name'

7）统计表数据行数

    hbase(main):021:0> count 'student'

8）删除数据

    删除某rowkey的全部数据：    hbase(main):016:0> deleteall 'student','1001'    删除某rowkey的某一列数据：    hbase(main):017:0> delete 'student','1002','info:sex'

9）清空表数据

    hbase(main):018:0> truncate 'student'    提示：清空表的操作顺序为先disable，然后再truncate。

10）删除表

    首先需要先让该表为disable状态：    hbase(main):019:0> disable 'student'    然后才能drop这个表：    hbase(main):020:0> drop 'student'    提示：如果直接drop表，会报错：ERROR: Table student is enabled. Disable it first.

11）变更表信息

    将info列族中的数据存放3个版本：    hbase(main):022:0> alter 'student',{NAME=>'info',VERSIONS=>3}    hbase(main):022:0> get 'student','1001',{COLUMN=>'info:name',VERSIONS=>3}

二、Flink整合HBase写入操作

现在需要将Flink处理的数据存入HBase数据库(namespace)shtd_result的order_info表中，rowkey为id的值，然后在Linux的HBase shell命令行中查询列consignee，并查询出任意5条

表空间为：shtd_result，表为order_info，列族为：info
表结构为：

我们需要写一个WriteToHBase类，集成自RichSinkFunction，RichSinkFunction 是一个抽象类，提供了一个更为丰富的接口，用于实现自定义的 Sink（接收器）功能。

在Scala api中RichSinkFunction的主要方法有open，invoke以及close。

• open(Configuration parameters):

  这个方法在 Sink 函数初始化时被调用，通常用于一次性的设置工作，例如打开数据库连接或初始化状态。
  参数 parameters 提供了访问 Flink 配置的能力。

• invoke(value: T, context: SinkFunction.Context):

  这是核心方法，用于处理每条流入的数据。
  value 参数代表当前的数据元素。
  context 提供了此元素的上下文信息，如当前的处理时间或事件时间。

• close():

  当 Sink 不再接收数据时调用此方法，用于执行清理工作，如关闭数据库连接。
  这个方法是在最后一次调用 invoke 方法后执行。

了解了这些方法后，我们来写一下WriteToHBase

一、WriteToHBase的实现

  class WriteToHBase extends RichSinkFunction[OrderData] {    @transient private var connection: Connection = _    @transient private var table: Table = _    override def open(parameters: Configuration): Unit = {      val config = HBaseConfiguration.create()      // 设置HBase配置, 如Zookeeper地址等       config.set("hbase.zookeeper.quorum", "bigdata1:2181")      connection = ConnectionFactory.createConnection(config)      table = connection.getTable(TableName.valueOf("shtd_result:order_info"))    }    override def invoke(value: OrderData, context: SinkFunction.Context): Unit = {      // 将 id 转换为行键（假设 id 是唯一的）      val rowKey = Bytes.toBytes(value.id.toString)      // 为该行创建一个新的 Put 实例      val put = new Put(rowKey)      // 向 Put 实例中添加列      put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("consignee"), Bytes.toBytes(value.consignee))      put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("consignee_tel"), Bytes.toBytes(value.consignee_tel))      put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("final_total_amount"), Bytes.toBytes(value.final_total_amount))      put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("order_status"), Bytes.toBytes(value.order_status))      put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("user_id"), Bytes.toBytes(value.user_id))      put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("delivery_address"), Bytes.toBytes(value.delivery_address))      put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("order_comment"), Bytes.toBytes(value.order_comment))      put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("out_trade_no"), Bytes.toBytes(value.out_trade_no))      put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("trade_body"), Bytes.toBytes(value.trade_body))      put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("operate_time"), Bytes.toBytes(value.operate_time))      put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("expire_time"), Bytes.toBytes(value.expire_time))      put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("tracking_no"), Bytes.toBytes(value.tracking_no))      put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("parent_order_id"), Bytes.toBytes(value.parent_order_id))      put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("img_url"), Bytes.toBytes(value.img_url))      put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("province_id"), Bytes.toBytes(value.province_id))      put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("benefit_reduce_amount"), Bytes.toBytes(value.benefit_reduce_amount))      table.put(put)    }    override def close(): Unit = {      if (table != null) {        table.close()      }      if (connection != null) {        connection.close()      }    }  }

在 Scala 和 Java 中，@transient 关键字用于标记一个类的成员变量为“暂时的”（transient），这意味着这个变量不会被默认的序列化过程序列化。

在 Flink中，通常用于：

• 防止序列化问题：当一个对象需要在不同的机器或上下文中传递时，某些属性可能不支持序列化（例如，数据库连接），或者序列化这些属性没有意义（例如，临时缓存）。使用 @transient 可以避免这些字段在对象序列化时引发错误。

• 减少网络开销：对于不需要跨节点传输的字段，使用 @transient 可以减少不必要的网络传输开销。

在 WriteToHBase 类中，connection 和 table 作为 HBase 的连接和表实例，通常不支持序列化，也不应该被序列化。所以，它们被标记为 @transient。

上面的向 Put 实例中添加列过于冗长，可以用反射来代替：

def addColumnsUsingReflection[T: TypeTag](put: Put, cf: String, data: T): Unit = {  //获取运行时镜像和实例镜像  val mirror = runtimeMirror(getClass.getClassLoader)  val instanceMirror = mirror.reflect(data)   //获取类成员并过滤方法  val members = typeOf[T].members.sorted.filterNot(_.isMethod)  //遍历字段并添加到 Put 实例  members.foreach { m =>    val fieldMirror = instanceMirror.reflectField(m.asTerm)    val name = m.name.toString.trim    val value = fieldMirror.get.toString    put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(name), Bytes.toBytes(value))  }}override def invoke(value: OrderData, context: SinkFunction.Context): Unit = {  val rowKey = Bytes.toBytes(value.id.toString)  val put = new Put(rowKey)  val infoCF = "info"  // 使用反射自动添加列  addColumnsUsingReflection(put, infoCF, value)  table.put(put)}

现在来解释一下这段代码：

addColumnsUsingReflection 函数定义

def addColumnsUsingReflection[T: TypeTag](put: Put, cf: String, data: T): Unit = {

• [T: TypeTag]：类型参数 T，带有一个上下文界定 TypeTag，这使得可以在运行时获取类型 T 的信息。
• (put: Put, cf: String, data: T)：函数接受三个参数：put 是 HBase 的 Put 实例，cf 是列族名，data 是要插入的数据对象。

获取运行时镜像和实例镜像

  val mirror = runtimeMirror(getClass.getClassLoader)  val instanceMirror = mirror.reflect(data)

• val mirror：创建一个 Mirror 实例，它是反射 API 的入口点。
• runtimeMirror(getClass.getClassLoader)：获取当前类的类加载器的运行时镜像。
• val instanceMirror：反射 data 对象，得到一个可以用来访问 data 实例成员的 InstanceMirror。

获取类成员并过滤方法

  val members = typeOf[T].members.sorted.filterNot(_.isMethod)

• typeOf[T]：获取类型 T 的类型信息。
• .members：获取类型 T 的所有成员（字段和方法）。
• .sorted：对成员进行排序（默认按名称）。
• .filterNot(_.isMethod)：过滤掉方法成员，只保留字段。

遍历字段并添加到 Put 实例

  members.foreach { m =>    val fieldMirror = instanceMirror.reflectField(m.asTerm)    val name = m.name.toString.trim    val value = fieldMirror.get.toString    put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(name), Bytes.toBytes(value))  }

• 遍历所有字段。
• val fieldMirror：反射每个字段，得到一个可以操作字段的 FieldMirror。
• m.asTerm：将成员 m 转换为一个 term（字段）。
• val name：获取字段名，并去除首尾空格。
• val value：获取字段的值并转换为字符串。
• put.addColumn：向 Put 实例添加列，列族为 cf，列名为字段名 name，值为字段值 value。

invoke 方法

override def invoke(value: OrderData, context: SinkFunction.Context): Unit = {  val rowKey = Bytes.toBytes(value.id.toString)  val put = new Put(rowKey)  val infoCF = "info"  addColumnsUsingReflection(put, infoCF, value)  table.put(put)}

• override def invoke：重写 RichSinkFunction 的 invoke 方法。
• val rowKey：将 OrderData 的 id 字段转换为字节作为行键。
• val put：创建一个新的 Put 实例。
• val infoCF：定义列族名。
• addColumnsUsingReflection：调用之前定义的函数来动态添加列。
• table.put(put)：将 Put 实例写入 HBase 表。

这段代码通过反射自动化了向 HBase Put 实例添加数据的过程，避免了手动为每个字段编写重复代码的需要。

然后我们需要对于dataStream应用刚才写的 WriteToHBase 类

应用 WriteToHBase 类

dataStream.addSink(new WriteToHBase)

二、HBase Shell操作1. 启动 HBase Shell

首先，我们需要进入 HBase Shell。在命令行中输入：

hbase shell

2. 创建命名空间

如果命名空间 shtd_result 还不存在，需要先创建它。在 HBase Shell 中执行以下命令：

create_namespace 'shtd_result'

3. 创建表

接着，创建表 order_info。我们需要定义至少一个列族（在这个示例中，我将使用 info 作为列族名）。在 HBase Shell 中执行以下命令：

create 'shtd_result:order_info', 'info'

这里，'shtd_result:order_info' 指定了完整的表名（包括命名空间），而 'info' 是列族名。

4. 验证表创建

最后，您可以列出所有表来验证新表是否已成功创建：

list

5. 查询

scan 'shtd_result:order_info', {COLUMNS => ['info:consignee'], LIMIT => 5}

这里的 scan 命令用于扫描 shtd_result:order_info 表，COLUMNS 参数指定我们只关心 info:consignee 列（假设 consignee 存储在名为 info 的列族中），而 LIMIT => 5 指定我们只查看 5 条记录。