本文為您介紹如何使用云數據庫HBase連接器。
背景信息
云數據庫HBase是低成本、高擴展、云智能的大數據NoSQL,兼容標準HBase訪問協議,提供低成本存儲、高擴展吞吐、智能數據處理等核心優勢,是為淘寶推薦、花唄風控、廣告投放、監控大屏、菜鳥物流軌跡、支付寶賬單、手淘消息等眾多阿里巴巴核心服務提供支撐的數據庫,具備PB規模、高并發、秒級伸縮、毫秒響應、跨機房高可用、全托管、全球分布等企業能力。
HBase連接器支持的信息如下:
類別 | 詳情 |
支持類型 | 維表和結果表 |
運行模式 | 流模式 |
數據格式 | 暫不支持 |
特有監控指標 |
|
API種類 | SQL |
是否支持更新或刪除結果表數據 | 是 |
前提條件
注意事項
使用前,請確認已創建數據庫實例類型,并選擇正確的連接器,使用錯誤的連接器可能導致不可預期問題:
云數據庫HBase實例,使用本文的HBase連接器。
Lindorm實例兼容HBase模式,使用Lindorm連接器,詳情請參見云原生多模數據庫Lindorm。
如果連接開源HBase,則無法保證數據的正確性。
語法結構
CREATE TABLE hbase_table(
rowkey INT,
family1 ROW<q1 INT>,
family2 ROW<q2 STRING, q3 BIGINT>,
family3 ROW<q4 DOUBLE, q5 BOOLEAN, q6 STRING>
) WITH (
'connector'='cloudhbase',
'table-name'='<yourTableName>',
'zookeeper.quorum'='<yourZookeeperQuorum>'
);
HBase的列族(Column Family)必須聲明為ROW類型,列族名即該ROW的字段名。例如,DDL定義中聲明了family1、family2和family3三個列族。
HBase列族中的列(Column)與對應ROW中嵌套的每個字段對應,列名即字段名。例如,DDL定義中列族family2聲明了q2和q3兩列。
除了類型為ROW的字段外,只能有一個原始類型(Atomic Type)的字段(例如STRING或BIGINT),該字段將被視作HBase的行鍵(Row Key),例如DDL定義中的Rowkey。
必須將HBase的行鍵定義為結果表的主鍵(Primary Key),如果沒有顯示定義主鍵,默認使用行鍵作為主鍵。
結果表中不需要將HBase表的所有列族和列都進行聲明,只聲明需要的即可。
WITH參數
通用
參數
說明
數據類型
是否必填
默認值
備注
connector
表類型。
String
是
無
固定值為
cloudhbase
。table-name
HBase表名。
String
是
無
無。
zookeeper.znode.quorum
HBase的zookeeper住址。
String
是
無
無。
zookeeper.znode.parent
HBase在zookeeper中的根目錄。
String
否
/hbase
僅在HBase標準版中生效。
userName
用戶名。
String
否
無
僅在HBase增強版中生效。
password
密碼。
String
否
無
僅在HBase增強版中生效。
haclient.cluster.id
HBase高可用實例ID。
String
否
無
只有訪問同城主備實例時才需要配置僅在HBase增強版中生效。
retires.number
HBase客戶端的重試次數。
Integer
否
31
無。
null-string-literal
HBase字段類型為字符串時,如果Flink字段數據為null,則將該字段賦值為
null-string-literal
,并寫入HBase。String
否
null
無。
結果表獨有
參數
說明
數據類型
是否必填
默認值
備注
sink.buffer-flush.max-size
寫入HBase前,內存中緩存的數據量(字節)大小。調大該值有利于提高HBase寫入性能,但會增加寫入延遲和內存使用。
String
否
2MB
支持字節單位B、KB、MB和GB,不區分大小寫。設置為0表示不進行緩存。
sink.buffer-flush.max-rows
寫入HBase前,內存中緩存的數據條數。調大該值有利于提高HBase寫入性能,但會增加寫入延遲和內存使用。
Integer
否
1000
設置為0表示不進行緩存。
sink.buffer-flush.interval
將緩存數據周期性寫入到HBase的間隔,可以控制寫入HBase的延遲。
Duration
否
1s
支持時間單位ms、s、min、h和d。設置為0表示關閉定期寫入。
dynamic.table
是否使用支持動態列的HBase表。
Boolean
否
false
參數取值如下:
true:使用支持動態列的HBase表。
false:不使用支持動態列的HBase表。
sink.ignore-delete
是否忽略撤回消息。
Boolean
否
false
參數取值如下:
true:忽略撤回消息。
false:不忽略撤回消息。
說明僅實時計算引擎VVR 4.0.10及以上版本支持該參數。
sink.sync-write
是否同步寫入HBase。
Boolean
否
true
參數取值如下:
true:同步寫,保證順序,會犧牲一定性能。
false:異步寫,不保證順序,性能更好。
說明僅實時計算引擎VVR 4.0.13及以上版本支持該參數。
sink.buffer-flush.batch-rows
同步寫入HBase時內存中緩存的數據條數,調大該值有利于提高HBase寫入性能,但會增加寫入延遲和內存使用。
Integer
否
100
僅當sink.sync-write為true時生效。
說明僅實時計算引擎VVR 4.0.13及以上版本支持該參數。
sink.ignore-null
是否忽略寫入null值。
Boolean
否
false
說明設置成true時,參數
null-string-literal
將不再生效。僅實時計算引擎VVR 8.0.9及以上版本支持該參數。
維表獨有(比如Cache參數)
參數
說明
數據類型
是否必填
默認值
備注
cache
緩存策略。
String
否
ALL
目前云數據庫HBase版維表支持以下三種緩存策略:
None:無緩存。
LRU:緩存維表里的部分數據。源表的每條數據都會觸發系統先在Cache中查找數據,如果沒有找到,則去物理維表中查找。
說明需要配置相關參數:緩存大?。╟acheSize)和緩存更新時間間隔(cacheTTLMs)。
ALL(默認值):緩存維表里的所有數據。在Job運行前,系統會將維表中所有數據加載到Cache中,之后所有的維表查找數據都會通過Cache進行。如果在Cache中無法找到數據,則KEY不存在,并在Cache過期后重新加載一遍全量Cache。
說明適用于遠程表數據量小且MISS KEY(源表數據和維表JOIN時,ON條件無法關聯)特別多的場景。需要配置相關參數:緩存更新時間間隔cacheTTLMs,更新時間黑名單cacheReloadTimeBlackList。
因為系統會異步加載維表數據,所以在使用CACHE ALL時,需要增加維表JOIN節點的內存,增加的內存大小為遠程表數據量的兩倍。
cacheSize
緩存大小。
Long
否
10000
當緩存策略選擇LRU時,可以設置緩存大小。
cacheTTLMs
緩存失效時間,單位為毫秒。
Long
否
無
cacheTTLMs配置和cache有關:
如果cache配置為None,則cacheTTLMs可以不配置,表示緩存不超時。
如果cache配置為LRU,則cacheTTLMs為緩存超時時間。默認不過期。
如果cache配置為ALL,則cacheTTLMs為緩存加載時間。默認不重新加載。
cacheEmpty
是否緩存空結果。
Boolean
否
true
無。
cacheReloadTimeBlackList
更新時間黑名單。在緩存策略選擇為ALL時,啟用更新時間黑名單,防止在此時間內做Cache更新(例如雙11場景)。
String
否
無
格式為2017-10-24 14:00 -> 2017-10-24 15:00,2017-11-10 23:30 -> 2017-11-11 08:00。分隔符的使用情況如下所示:
用英文逗號(,)來分隔多個黑名單。
用箭頭(->)來分割黑名單的起始結束時間。
cacheScanLimit
讀取全量HBase數據,RPC(Remote Procedure Call Protocol)服務端一次返回給客戶端的行數。
Integer
否
100
緩存策略選擇ALL時啟用。
類型映射
Flink中的數據類型在HBase中通過org.apache.hadoop.hbase.util.Bytes
轉換成字節數組,解碼過程有以下兩種情況:
對于Flink的非字符串類型,如果HBase中的值為空字節數組,則解碼為null。
對于Flink的字符串類型,如果HBase中的值為
null-string-literal
字節數組,則解碼為null。
Flink SQL類型 | 寫入Bytes時CloudHBase轉換函數 | 從CloudHBase讀取Bytes的轉換函數 |
CHAR | byte[] toBytes(String s) | String toString(byte[] b) |
VARCHAR | ||
STRING | ||
BOOLEAN | byte[] toBytes(boolean b) | boolean toBoolean(byte[] b) |
BINARY | byte[] | byte[] |
VARBINARY | ||
DECIMAL | byte[] toBytes(BigDecimal v) | BigDecimal toBigDecimal(byte[] b) |
TINYINT | new byte[] { val } | bytes[0] |
SMALLINT | byte[] toBytes(short val) | short toShort(byte[] bytes) |
INT | byte[] toBytes(int val) | int toInt(byte[] bytes) |
BIGINT | byte[] toBytes(long val) | long toLong(byte[] bytes) |
FLOAT | byte[] toBytes(float val) | float toFloat(byte[] bytes) |
DOUBLE | byte[] toBytes(double val) | double toDouble(byte[] bytes) |
DATE | 將日期轉換成自1970.01.01以來的天數,用int表示,并通過 | HBase字節數組通過 |
TIME | 將時間轉換成自00:00:00以來的毫秒數,用int表示,并通過 | HBase字節數組通過 |
TIMESTAMP | 將時間戳轉換成自1970-01-01 00:00:00以來的毫秒數,用long表示,并通過 | HBase字節數組通過 |
代碼示例
維表示例。
CREATE TEMPORARY TABLE datagen_source ( a INT, b BIGINT, c STRING, `proc_time` AS PROCTIME() ) WITH ( 'connector'='datagen' ); CREATE TEMPORARY TABLE hbase_dim ( rowkey INT, family1 ROW<col1 INT>, family2 ROW<col1 STRING, col2 BIGINT>, family3 ROW<col1 DOUBLE, col2 BOOLEAN, col3 STRING> ) WITH ( 'connector' = 'cloudhbase', 'table-name' = '<yourTableName>', 'zookeeper.quorum' = '<yourZookeeperQuorum>' ); CREATE TEMPORARY TABLE blackhole_sink( a INT, f1c1 INT, f3c3 STRING ) WITH ( 'connector' = 'blackhole' ); INSERT INTO blackhole_sink SELECT a, family1.col1 as f1c1, family3.col3 as f3c3 FROM datagen_source JOIN hbase_dim FOR SYSTEM_TIME AS OF datagen_source.`proc_time` as h ON datagen_source.a = h.rowkey;
結果表示例。
CREATE TEMPORARY TABLE datagen_source ( rowkey INT, f1q1 INT, f2q1 STRING, f2q2 BIGINT, f3q1 DOUBLE, f3q2 BOOLEAN, f3q3 STRING ) WITH ( 'connector'='datagen' ); CREATE TEMPORARY TABLE hbase_sink ( rowkey INT, family1 ROW<q1 INT>, family2 ROW<q1 STRING, q2 BIGINT>, family3 ROW<q1 DOUBLE, q2 BOOLEAN, q3 STRING>, PRIMARY KEY (rowkey) NOT ENFORCED ) WITH ( 'connector'='cloudhbase', 'table-name'='<yourTableName>', 'zookeeper.quorum'='<yourZookeeperQuorum>' ); INSERT INTO hbase_sink SELECT rowkey, ROW(f1q1), ROW(f2q1, f2q2), ROW(f3q1, f3q2, f3q3) FROM datagen_source;
結果動態表示例。
CREATE TEMPORARY TABLE datagen_source ( id INT, f1hour STRING, f1deal BIGINT, f2day STRING, f2deal BIGINT ) WITH ( 'connector'='datagen' ); CREATE TEMPORARY TABLE hbase_sink ( rowkey INT, f1 ROW<`hour` STRING, deal BIGINT>, f2 ROW<`day` STRING, deal BIGINT> ) WITH ( 'connector'='cloudhbase', 'table-name'='<yourTableName>', 'zookeeper.quorum'='<yourZookeeperQuorum>', 'dynamic.table'='true' ); INSERT INTO hbase_sink SELECT id, ROW(f1hour, f1deal), ROW(f2day, f2deal) FROM datagen_source;
當dynamic.table參數值為true時,表示使用支持動態列的HBase表。
每個列族對應的ROW中必須聲明兩個字段:第1個字段的值表示動態列,第2個字段的值表示動態列的值。
如果datagen_source表存在一條數據,代表ID為1的商品,在10:00-11:00點之間的成交額是100,在2020年7月26日當天的成交額是10000,則HBase中將插入行鍵為1的行,其中f1:10為100,f2:2020-7-26為10000。