一、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
表结构为:

字段类型注释
rowkeystringHBase的主键,值为id
idbigint
consigneestring
consignee_telstring
final_total_amountdouble
order_statusstring
user_idbigint
delivery_addressstring
order_commentstring
out_trade_nostring
trade_bodystring
create_timestring转成yyyy-MM-dd hh:mm:ss格式的的字符串
operate_timestring转成yyyy-MM-dd hh:mm:ss格式的的字符串
expire_timestring转成yyyy-MM-dd hh:mm:ss格式的的字符串
tracking_nostring
parent_order_idbigint
img_urlstring
province_idint
benefit_reduce_amountdouble

我们需要写一个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:重写 RichSinkFunctioninvoke 方法。
  • val rowKey:将 OrderDataid 字段转换为字节作为行键。
  • 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 条记录。