Flink SQL 實戰：HBase 的結合應用
本文主要介紹 HBase 和 Flink SQL 的結合使用。HBase 作為 Google 發表 Big Table 論文的開源實現版本，是一種分布式列式存儲的數據庫，構建在 HDFS 之上的 NoSQL 數據庫，非常適合大規模實時查詢，因此 HBase 在實時計算領域使用非常廣泛。可以實時寫 HBase，也可以利用 buckload 一把把離線 Job 生成 HFile Load 到HBase 表中。而當下 Flink SQL 的火熱程度不用多說，Flink SQL 也為 HBase 提供了 connector，因此 HBase 與 Flink SQL 的結合非常有必要實踐實踐。
當然，本文假設用戶有一定的 HBase 知識基礎，不會詳細去介紹 HBase 的架構和原理，本文著重介紹 HBase 和 Flink 在實際場景中的結合使用。主要分為兩種場景，第一種場景：HBase 作為維表與 Flink Kafka table 做 temporal table join 的場景;第二種場景：Flink SQL 做計算之后的結果寫到 HBase 表，供其他用戶查詢的場景。因此，本文介紹的內容如下所示：
- HBase 環境準備
- 數據準備
- HBase 作為維度表進行 temporal table join的場景
- Flink SQL 做計算寫 HBase 的場景
- 總結
01 HBase 環境準備
由于沒有測試的 HBase 環境以及為了避免污染線上 Hbase 環境。因此，自己 build一個 Hbase docker image(大家可以 docker pull guxinglei/myhbase 拉到本地)，是基于官方干凈的 ubuntu imgae 之上安裝了 Hbase 2.2.0 版本以及 JDK1.8 版本。
啟動容器，暴露 Hbase web UI 端口以及內置 zk 端口，方便我們從 web 頁面看信息以及創建 Flink Hbase table 需要 zk 的鏈接信息。
- docker run -it --network=host -p 2181:2181 -p 60011:60011 docker.io/guxinglei/myhbase:latest bash
進入容器，啟動 HBase 集群，以及啟動 rest server，后續方便我們用 REST API 來讀取 Flink SQL 寫進 HBase 的數據。
- # 啟動hbase 集群bin/start-hbase.sh# 后臺啟動restServerbin/hbase-daemon.sh start rest -p 8000
02 數據準備
由于 HBase 環境是自己臨時搞的單機服務，里面沒有數據，需要往里面寫點數據供后續示例用。在 Flink SQL 實戰系列第二篇中介紹了如何注冊 Flink Mysql table，我們可以將廣告位表抽取到 HBase 表中，用來做維度表，進行 temporal table join。因此，我們需要在 HBase 中創建一張表，同時還需要創建 Flink HBase table, 這兩張表通過 Flink SQL 的 HBase connector 關聯起來。
在容器中啟動 HBase shell，創建一張名為 dim_hbase 的 HBase 表，建表語句如下所示：
- # 在hbase shell創建 hbase表
- hbase(main):002:0> create 'dim_hbase','cf'
- Created table dim_hbase
- Took 1.3120 seconds
- => Hbase::Table - dim_hbase
在 Flink 中創建 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 已經創建好了，就可以寫抽取數據到HBase 的 SQL job 了，SQL 語句以及 job 狀態如下所示：
- # 抽取Mysql數據到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 廣告位的明細表等等，使用場景非常廣泛。那么作為分布式數據庫的 HBase 比 MySQL 作為維度表用作維度表 join 更有優勢。在 Flink SQL 實戰系列第二篇中，我們注冊了廣告的點擊流，將 Kafka topic 注冊 Flink Kafka Table，同時也介紹了 temporal table join 在 Flink SQL 中的使用;那么本節中將會介紹 HBase 作為維度表來使用，上面小節中已經將數據抽取到 Hbase 中了，我們直接寫 temporal table join 計算邏輯即可。
作為廣告點擊流的 Flink Kafa table 與 作為廣告位的 Flink HBase table 通過廣告位 Id 進行 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 集群上的狀態以及 join 結果如下所示：
04 計算結果 sink 到 HBase 作為結果的場景
上面小節中，HBase 作為維度表用作 temporal table join 是非常常見的場景，實際上 HBase 作為存儲計算結果也是非常常見的場景，畢竟 Hbase 作為分布式數據庫，底層存儲是擁有多副本機制的 HDFS，維護簡單，擴容方便， 實時查詢快，而且提供各種客戶端方便下游使用存儲在 HBase 中的數據。那么本小節就介紹 Flink SQL 將計算結果寫到 HBase，并且通過 REST API 查詢計算結果的場景。
進入容器中，在 HBase 中新建一張 HBase 表，一個 column family 就滿足需求，建表語句如下所示：
- # 注冊hbase sink table
- create 'dwa_hbase_click_report','cf'
建立好 HBase 表之后，我們需要在 Flink SQL 創建一張 Flink HBase table，這個時候我們需要明確 cf 這個 column famaly 下面 column 字段，在 Flink SQL實戰第二篇中，已經注冊好了作為點擊流的 Flink Kafka table，因此本節中，將會計算點擊流的 uv 和點擊數，因此兩個 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 和存儲計算結果的 HBase table 和 Flink HBase table 已經準備了，我們將做一個1分鐘的翻轉窗口計算 uv 和點擊數，并且將計算結果寫到 HBase 中。對 HBase 了解的人應該知道，rowkey 的設計對 hbase regoin 的分布有著非常重要的影響，基于此我們的 rowkey 是使用 Flink SQL 內置的 reverse 函數進行廣告位 Id 進行反轉和窗口啟始時間做 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 提交之后的狀態以及結果 check 如下所示：
上述 SQL job 已經成功的將結算結果寫到 HBase 中了。對于線上的 HBase 服務來講，很多同事不一定有 HBase 客戶端的權限，從而也不能通過 HBase shell 讀取數據;另外作為線上報表服務顯然不可能通過 HBase shell 來通過查詢數據。因此，在實時報表場景中，數據開發工程師將數據寫入 HBase, 前端工程師通過 REST API 來讀取數據。前面我們已經啟動了 HBase rest server 進程，我們可以通 rest 服務提供讀取 HBase 里面的數據。
我們先 get 一條剛剛寫到 HBase 中的數據看看，如下所示：
下面我們開始通過 REST API 來查詢 HBase 中的數據，第一步，執行如下語句拿到 scannerId;首先需要將要查詢的 rowkey 進行 base64 編碼才能使用，后面需要將結果進行 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"
第二步，執行如下語句根據上條語句返回的 scannerID 查詢數據，可以看到返回的結果：
- curl -vi -X GET \
- -H "Accept: application/json" \
- "http://hostname:8000/dwa_hbase_click_report/scanner/16063768141736ac0a8b5"
第三步，查詢完畢之后，執行如下語句刪除該 scannerId:
- curl -vi -X DELETE \
- -H "Accept: text/xml" \
- "http://hostname:8000/dwa_hbase_click_report/scanner/16063768141736ac0a8b5"
五. 總結
在本篇文章中，我們介紹了 HBase 和 Flink SQL 的結合使用比較廣泛兩種的場景：作為維度表用以及存儲計算結果;同時使用 REST API 對 HBase 中的數據進行查詢，對于查詢用戶來說，避免直接暴露 HBase 的 zk，同時將 rest server 和 HBase 集群解耦。
作者簡介
余敖，360 數據開發高級工程師，目前專注于基于 Flink 的實時數倉建設與平臺化工作。對 Flink、Kafka、Hive、Spark 等進行數據 ETL 和數倉開發有豐富的經驗。
