本文主要介紹 HBase 和 Flink SQL 的結(jié)合使用。HBase 作為 Google 發(fā)表 Big Table 論文的開(kāi)源實(shí)現(xiàn)版本，是一種分布式列式存儲(chǔ)的數(shù)據(jù)庫(kù)，構(gòu)建在 HDFS 之上的 NoSQL 數(shù)據(jù)庫(kù)，非常適合大規(guī)模實(shí)時(shí)查詢(xún)，因此 HBase 在實(shí)時(shí)計(jì)算領(lǐng)域使用非常廣泛。可以實(shí)時(shí)寫(xiě) HBase，也可以利用 buckload 一把把離線 Job 生成 HFile Load 到HBase 表中。而當(dāng)下 Flink SQL 的火熱程度不用多說(shuō)，F(xiàn)link SQL 也為 HBase 提供了 connector，因此 HBase 與 Flink SQL 的結(jié)合非常有必要實(shí)踐實(shí)踐。

當(dāng)然，本文假設(shè)用戶(hù)有一定的 HBase 知識(shí)基礎(chǔ)，不會(huì)詳細(xì)去介紹 HBase 的架構(gòu)和原理，本文著重介紹 HBase 和 Flink 在實(shí)際場(chǎng)景中的結(jié)合使用。主要分為兩種場(chǎng)景，第一種場(chǎng)景：HBase 作為維表與 Flink Kafka table 做 temporal table join 的場(chǎng)景;第二種場(chǎng)景：Flink SQL 做計(jì)算之后的結(jié)果寫(xiě)到 HBase 表，供其他用戶(hù)查詢(xún)的場(chǎng)景。因此，本文介紹的內(nèi)容如下所示：

• HBase 環(huán)境準(zhǔn)備
• 數(shù)據(jù)準(zhǔn)備
• HBase 作為維度表進(jìn)行 temporal table join的場(chǎng)景
• Flink SQL 做計(jì)算寫(xiě) HBase 的場(chǎng)景
• 總結(jié)

01 HBase 環(huán)境準(zhǔn)備

由于沒(méi)有測(cè)試的 HBase 環(huán)境以及為了避免污染線上 Hbase 環(huán)境。因此，自己 build一個(gè) Hbase docker image(大家可以 docker pull guxinglei/myhbase 拉到本地)，是基于官方干凈的 ubuntu imgae 之上安裝了 Hbase 2.2.0 版本以及 JDK1.8 版本。

啟動(dòng)容器，暴露 Hbase web UI 端口以及內(nèi)置 zk 端口，方便我們從 web 頁(yè)面看信息以及創(chuàng)建 Flink Hbase table 需要 zk 的鏈接信息。

docker run -it --network=host -p 2181:2181 -p 60011:60011 docker.io/guxinglei/myhbase:latest bash 

進(jìn)入容器，啟動(dòng) HBase 集群，以及啟動(dòng) rest server，后續(xù)方便我們用 REST API 來(lái)讀取 Flink SQL 寫(xiě)進(jìn) HBase 的數(shù)據(jù)。

# 啟動(dòng)hbase 集群bin/start-hbase.sh# 后臺(tái)啟動(dòng)restServerbin/hbase-daemon.sh start rest -p 8000 

02 數(shù)據(jù)準(zhǔn)備

由于 HBase 環(huán)境是自己臨時(shí)搞的單機(jī)服務(wù)，里面沒(méi)有數(shù)據(jù)，需要往里面寫(xiě)點(diǎn)數(shù)據(jù)供后續(xù)示例用。在 Flink SQL 實(shí)戰(zhàn)系列第二篇中介紹了如何注冊(cè) Flink Mysql table，我們可以將廣告位表抽取到 HBase 表中，用來(lái)做維度表，進(jìn)行 temporal table join。因此，我們需要在 HBase 中創(chuàng)建一張表，同時(shí)還需要?jiǎng)?chuàng)建 Flink HBase table, 這兩張表通過(guò) Flink SQL 的 HBase connector 關(guān)聯(lián)起來(lái)。

在容器中啟動(dòng) HBase shell，創(chuàng)建一張名為 dim_hbase 的 HBase 表，建表語(yǔ)句如下所示：

# 在hbase shell創(chuàng)建 hbase表 
hbase(main):002:0> create 'dim_hbase','cf' 
Created table dim_hbase 
Took 1.3120 seconds 
=> Hbase::Table - dim_hbase 

在 Flink 中創(chuàng)建 Flink HBase table，建表語(yǔ)句如下所示：

# 注冊(cè) Flink Hbase table 
DROP TABLE IF EXISTS flink_rtdw.demo.hbase_dim_table; 
CREATE TABLE flink_rtdw.demo.hbase_dim_table ( 
  rowkey STRING, 
  cf ROW < adspace_name STRING >, 
  PRIMARY KEY (rowkey) NOT ENFORCED 
) WITH ( 
'connector' = 'hbase-1.4', 
'table-name' = 'dim_hbase', 
'sink.buffer-flush.max-rows' = '1000', 
'zookeeper.quorum' = 'localhost:2181' 
); 

Flink MySQL table 和 Flink HBase table 已經(jīng)創(chuàng)建好了，就可以寫(xiě)抽取數(shù)據(jù)到HBase 的 SQL job 了，SQL 語(yǔ)句以及 job 狀態(tài)如下所示：

# 抽取Mysql數(shù)據(jù)到Hbase表中 
 
 
insert into 
  hbase_dim_table 
select 
CAST (ID as VARCHAR), 
ROW(name) 
from 
  mysql_dim_table; 

03 HBase 作為維表與 Kafka做 temporal join 的場(chǎng)景

在 Flink SQL join 中，維度表的 join 一定繞不開(kāi)的，比如訂單金額 join 匯率表，點(diǎn)擊流 join 廣告位的明細(xì)表等等，使用場(chǎng)景非常廣泛。那么作為分布式數(shù)據(jù)庫(kù)的 HBase 比 MySQL 作為維度表用作維度表 join 更有優(yōu)勢(shì)。在 Flink SQL 實(shí)戰(zhàn)系列第二篇中，我們注冊(cè)了廣告的點(diǎn)擊流，將 Kafka topic 注冊(cè) Flink Kafka Table，同時(shí)也介紹了 temporal table join 在 Flink SQL 中的使用;那么本節(jié)中將會(huì)介紹 HBase 作為維度表來(lái)使用，上面小節(jié)中已經(jīng)將數(shù)據(jù)抽取到 Hbase 中了，我們直接寫(xiě) temporal table join 計(jì)算邏輯即可。

作為廣告點(diǎn)擊流的 Flink Kafa table 與 作為廣告位的 Flink HBase table 通過(guò)廣告位 Id 進(jìn)行 temporal table join，輸出廣告位 ID 和廣告位中文名字，SQL join 邏輯如下所示：

select adsdw_dwd_max_click_mobileapp.publisher_adspace_adspaceId as publisher_adspace_adspaceId, 
       hbase_dim_table.cf.adspace_name as publisher_adspace_name 
from adsdw_dwd_max_click_mobileapp 
left join hbase_dim_table FOR SYSTEM_TIME AS OF adsdw_dwd_max_click_mobileapp.procTime 
on cast(adsdw_dwd_max_click_mobileapp.publisher_adspace_adspaceId as string) = hbase_dim_table.rowkey; 

temporal table join job 提交 Flink 集群上的狀態(tài)以及 join 結(jié)果如下所示：

04 計(jì)算結(jié)果 sink 到 HBase 作為結(jié)果的場(chǎng)景

上面小節(jié)中，HBase 作為維度表用作 temporal table join 是非常常見(jiàn)的場(chǎng)景，實(shí)際上 HBase 作為存儲(chǔ)計(jì)算結(jié)果也是非常常見(jiàn)的場(chǎng)景，畢竟 Hbase 作為分布式數(shù)據(jù)庫(kù)，底層存儲(chǔ)是擁有多副本機(jī)制的 HDFS，維護(hù)簡(jiǎn)單，擴(kuò)容方便， 實(shí)時(shí)查詢(xún)快，而且提供各種客戶(hù)端方便下游使用存儲(chǔ)在 HBase 中的數(shù)據(jù)。那么本小節(jié)就介紹 Flink SQL 將計(jì)算結(jié)果寫(xiě)到 HBase，并且通過(guò) REST API 查詢(xún)計(jì)算結(jié)果的場(chǎng)景。

進(jìn)入容器中，在 HBase 中新建一張 HBase 表，一個(gè) column family 就滿(mǎn)足需求，建表語(yǔ)句如下所示：

# 注冊(cè)hbase sink table 
create 'dwa_hbase_click_report','cf' 

建立好 HBase 表之后，我們需要在 Flink SQL 創(chuàng)建一張 Flink HBase table，這個(gè)時(shí)候我們需要明確 cf 這個(gè) column famaly 下面 column 字段，在 Flink SQL實(shí)戰(zhàn)第二篇中，已經(jīng)注冊(cè)好了作為點(diǎn)擊流的 Flink Kafka table，因此本節(jié)中，將會(huì)計(jì)算點(diǎn)擊流的 uv 和點(diǎn)擊數(shù)，因此兩個(gè) column 分別為 uv 和 click_count，建表語(yǔ)句如下所示：

# 注冊(cè) Flink Hbase table 
DROP TABLE IF EXISTS flink_rtdw.demo.dwa_hbase_click_report; 
CREATE TABLE flink_rtdw.demo.dwa_hbase_click_report ( 
  rowkey STRING, 
  cf ROW < uv BIGINT, click_count BIGINT >, 
  PRIMARY KEY (rowkey) NOT ENFORCED 
) WITH ( 
'connector' = 'hbase-1.4', 
'table-name' = 'dwa_hbase_click_report', 
'sink.buffer-flush.max-rows' = '1000', 
'zookeeper.quorum' = 'hostname:2181' 
); 

前面點(diǎn)擊流的 Flink Kafka table 和存儲(chǔ)計(jì)算結(jié)果的 HBase table 和 Flink HBase table 已經(jīng)準(zhǔn)備了，我們將做一個(gè)1分鐘的翻轉(zhuǎn)窗口計(jì)算 uv 和點(diǎn)擊數(shù)，并且將計(jì)算結(jié)果寫(xiě)到 HBase 中。對(duì) HBase 了解的人應(yīng)該知道，rowkey 的設(shè)計(jì)對(duì) hbase regoin 的分布有著非常重要的影響，基于此我們的 rowkey 是使用 Flink SQL 內(nèi)置的 reverse 函數(shù)進(jìn)行廣告位 Id 進(jìn)行反轉(zhuǎn)和窗口啟始時(shí)間做 concat，因此，SQL 邏輯語(yǔ)句如下所示：

INSERT INTO dwa_hbase_click_report 
SELECT 
CONCAT(REVERSE(CAST(publisher_adspace_adspaceId AS STRING)) , 
'_', 
CAST((UNIX_TIMESTAMP(DATE_FORMAT(TUMBLE_START(ets, INTERVAL '1' MINUTE),'yyyy-MM-dd HH:mm:ss')) * 1000) AS STRING) 
  ) as rowkey,  
ROW(COUNT(DISTINCT audience_mvid) , COUNT(audience_behavior_click_creative_impressionId)) as cf 
FROM 
  adsdw_dwd_max_click_mobileapp 
WHERE publisher_adspace_adspaceId IS NOT NULL AND audience_mvid IS NOT NULL AND audience_behavior_click_creative_impressionId IS NOT NULL 
GROUP BY 
  TUMBLE(ets, INTERVAL '1' MINUTE), 
  publisher_adspace_adspaceId; 

SQL job 提交之后的狀態(tài)以及結(jié)果 check 如下所示：

上述 SQL job 已經(jīng)成功的將結(jié)算結(jié)果寫(xiě)到 HBase 中了。對(duì)于線上的 HBase 服務(wù)來(lái)講，很多同事不一定有 HBase 客戶(hù)端的權(quán)限，從而也不能通過(guò) HBase shell 讀取數(shù)據(jù);另外作為線上報(bào)表服務(wù)顯然不可能通過(guò) HBase shell 來(lái)通過(guò)查詢(xún)數(shù)據(jù)。因此，在實(shí)時(shí)報(bào)表場(chǎng)景中，數(shù)據(jù)開(kāi)發(fā)工程師將數(shù)據(jù)寫(xiě)入 HBase, 前端工程師通過(guò) REST API 來(lái)讀取數(shù)據(jù)。前面我們已經(jīng)啟動(dòng)了 HBase rest server 進(jìn)程，我們可以通 rest 服務(wù)提供讀取 HBase 里面的數(shù)據(jù)。

我們先 get 一條剛剛寫(xiě)到 HBase 中的數(shù)據(jù)看看，如下所示：

下面我們開(kāi)始通過(guò) REST API 來(lái)查詢(xún) HBase 中的數(shù)據(jù)，第一步，執(zhí)行如下語(yǔ)句拿到 scannerId;首先需要將要查詢(xún)的 rowkey 進(jìn)行 base64 編碼才能使用，后面需要將結(jié)果進(jìn)行 base64 解碼

rowkey base64 編碼前：0122612_1606295280000 base64 編碼之后：MDEyMjYxMl8xNjA2Mjk1MjgwMDAw

curl -vi -X PUT \ 
         -H "Accept: text/xml" \ 
         -H "Content-Type: text/xml" \ 
         -d '<Scanner startRow="MDEyMjYxMl8xNjA2Mjk1MjgwMDAw" endRow="MDEyMjYxMl8xNjA2Mjk1MjgwMDAw"></Scanner>' \ 
"http://hostname:8000/dwa_hbase_click_report/scanner" 

第二步，執(zhí)行如下語(yǔ)句根據(jù)上條語(yǔ)句返回的 scannerID 查詢(xún)數(shù)據(jù)，可以看到返回的結(jié)果：

curl -vi -X GET \ 
         -H "Accept: application/json" \ 
"http://hostname:8000/dwa_hbase_click_report/scanner/16063768141736ac0a8b5" 

第三步，查詢(xún)完畢之后，執(zhí)行如下語(yǔ)句刪除該 scannerId:

curl -vi -X DELETE \ 
         -H "Accept: text/xml" \ 
"http://hostname:8000/dwa_hbase_click_report/scanner/16063768141736ac0a8b5" 

五. 總結(jié)

在本篇文章中，我們介紹了 HBase 和 Flink SQL 的結(jié)合使用比較廣泛兩種的場(chǎng)景：作為維度表用以及存儲(chǔ)計(jì)算結(jié)果;同時(shí)使用 REST API 對(duì) HBase 中的數(shù)據(jù)進(jìn)行查詢(xún)，對(duì)于查詢(xún)用戶(hù)來(lái)說(shuō)，避免直接暴露 HBase 的 zk，同時(shí)將 rest server 和 HBase 集群解耦。

作者簡(jiǎn)介

余敖，360 數(shù)據(jù)開(kāi)發(fā)高級(jí)工程師，目前專(zhuān)注于基于 Flink 的實(shí)時(shí)數(shù)倉(cāng)建設(shè)與平臺(tái)化工作。對(duì) Flink、Kafka、Hive、Spark 等進(jìn)行數(shù)據(jù) ETL 和數(shù)倉(cāng)開(kāi)發(fā)有豐富的經(jīng)驗(yàn)。