HBase Shell操作&Flink写入HBase

一、HBase Shell操作

1、基本操作

1）进入HBase客户端命令行

1.     [root@bigdata1 hbase]$ bin/hbase shell

复制代码

2）查看帮助命令

1.     hbase(main):001:0> help

复制代码

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

1.     hbase(main):002:0> list

复制代码

2、表的操作

1）创建表

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

复制代码

2）插入数据到表

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

复制代码

3）扫描查看表数据

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

复制代码

4）查看表结构

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

复制代码

5）更新指定字段的数据

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

复制代码

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

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

复制代码

7）统计表数据行数

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

复制代码

8）删除数据

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

复制代码

9）清空表数据

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

复制代码

10）删除表

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

复制代码

11）变更表信息

1.     将info列族中的数据存放3个版本：
2.     hbase(main):022:0> alter 'student',{NAME=>'info',VERSIONS=>3}
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的主键，值为ididbigintconsigneestringconsignee_telstringfinal_total_amountdoubleorder_statusstringuser_idbigintdelivery_addressstringorder_commentstringout_trade_nostringtrade_bodystringcreate_timestring转成yyyy-MM-dd hh:mm:ss格式的的字符串operate_timestring转成yyyy-MM-dd hh:mm:ss格式的的字符串expire_timestring转成yyyy-MM-dd hh:mm:ss格式的的字符串tracking_nostringparent_order_idbigintimg_urlstringprovince_idintbenefit_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的实现

1.   class WriteToHBase extends RichSinkFunction[OrderData] {
2.     @transient private var connection: Connection = _
3.     @transient private var table: Table = _
5.     override def open(parameters: Configuration): Unit = {
6.       val config = HBaseConfiguration.create()
7.       // 设置HBase配置, 如Zookeeper地址等
8.        config.set("hbase.zookeeper.quorum", "bigdata1:2181")
10.       connection = ConnectionFactory.createConnection(config)
11.       table = connection.getTable(TableName.valueOf("shtd_result:order_info"))
12.     }
14.     override def invoke(value: OrderData, context: SinkFunction.Context): Unit = {
15.       // 将 id 转换为行键（假设 id 是唯一的）
16.       val rowKey = Bytes.toBytes(value.id.toString)
18.       // 为该行创建一个新的 Put 实例
19.       val put = new Put(rowKey)
21.       // 向 Put 实例中添加列
22.       put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("consignee"), Bytes.toBytes(value.consignee))
23.       put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("consignee_tel"), Bytes.toBytes(value.consignee_tel))
24.       put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("final_total_amount"), Bytes.toBytes(value.final_total_amount))
25.       put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("order_status"), Bytes.toBytes(value.order_status))
26.       put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("user_id"), Bytes.toBytes(value.user_id))
27.       put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("delivery_address"), Bytes.toBytes(value.delivery_address))
28.       put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("order_comment"), Bytes.toBytes(value.order_comment))
29.       put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("out_trade_no"), Bytes.toBytes(value.out_trade_no))
30.       put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("trade_body"), Bytes.toBytes(value.trade_body))
31.       put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("operate_time"), Bytes.toBytes(value.operate_time))
32.       put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("expire_time"), Bytes.toBytes(value.expire_time))
33.       put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("tracking_no"), Bytes.toBytes(value.tracking_no))
34.       put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("parent_order_id"), Bytes.toBytes(value.parent_order_id))
35.       put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("img_url"), Bytes.toBytes(value.img_url))
36.       put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("province_id"), Bytes.toBytes(value.province_id))
37.       put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("benefit_reduce_amount"), Bytes.toBytes(value.benefit_reduce_amount))
39.       table.put(put)
40.     }
42.     override def close(): Unit = {
43.       if (table != null) {
44.         table.close()
45.       }
46.       if (connection != null) {
47.         connection.close()
48.       }
49.     }
50.   }

复制代码

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

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

在 WriteToHBase 类中，connection 和 table 作为 HBase 的连接和表实例，通常不支持序列化，也不应该被序列化。所以，它们被标记为 @transient。
上面的向 Put 实例中添加列过于冗长，可以用反射来代替：

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

复制代码

现在来解释一下这段代码：
addColumnsUsingReflection 函数定义

1. 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 是要插入的数据对象。
  

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

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

复制代码

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

获取类成员并过滤方法

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

复制代码

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

遍历字段并添加到 Put 实例

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

复制代码

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

invoke 方法

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

复制代码

• 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 类

1. dataStream.addSink(new WriteToHBase)

复制代码

二、HBase Shell操作

1. 启动 HBase Shell

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

1. hbase shell

复制代码

2. 创建命名空间

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

1. create_namespace 'shtd_result'

复制代码

3. 创建表

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

1. create 'shtd_result:order_info', 'info'

复制代码

这里，'shtd_resultrder_info' 指定了完整的表名（包括命名空间），而 'info' 是列族名。
4. 验证表创建

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

1. list

复制代码

5. 查询

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

复制代码

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

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！