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

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

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

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

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

啟動容器，暴露 Hbase web UI 端口以及內(nèi)置 zk 端口，方便我們從 web 頁面看信息以及創(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)入容器，啟動 HBase 集群，以及啟動 rest server，后續(xù)方便我們用 REST API 來讀取 Flink SQL 寫進(jìn) HBase 的數(shù)據(jù)。

# 啟動hbase 集群bin/start-hbase.sh# 后臺啟動restServerbin/hbase-daemon.sh start rest -p 8000 

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

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

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

# 在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，建表語句如下所示：

# 注冊 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)建好了，就可以寫抽取數(shù)據(jù)到HBase 的 SQL job 了，SQL 語句以及 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 的場景

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

作為廣告點擊流的 Flink Kafa table 與 作為廣告位的 Flink HBase table 通過廣告位 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 計算結(jié)果 sink 到 HBase 作為結(jié)果的場景

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

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

# 注冊hbase sink table 
create 'dwa_hbase_click_report','cf' 

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

# 注冊 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' 
); 

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

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é)果寫到 HBase 中了。對于線上的 HBase 服務(wù)來講，很多同事不一定有 HBase 客戶端的權(quán)限，從而也不能通過 HBase shell 讀取數(shù)據(jù);另外作為線上報表服務(wù)顯然不可能通過 HBase shell 來通過查詢數(shù)據(jù)。因此，在實時報表場景中，數(shù)據(jù)開發(fā)工程師將數(shù)據(jù)寫入 HBase, 前端工程師通過 REST API 來讀取數(shù)據(jù)。前面我們已經(jīng)啟動了 HBase rest server 進(jìn)程，我們可以通 rest 服務(wù)提供讀取 HBase 里面的數(shù)據(jù)。

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

下面我們開始通過 REST API 來查詢 HBase 中的數(shù)據(jù)，第一步，執(zhí)行如下語句拿到 scannerId;首先需要將要查詢的 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í)行如下語句根據(jù)上條語句返回的 scannerID 查詢數(shù)據(jù)，可以看到返回的結(jié)果：

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

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

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

五. 總結(jié)

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

作者簡介

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