Realtime Compute for Apache Flink：雲原生多模資料庫Lindorm

背景資訊

Lindorm是面向物聯網、互連網、車連網等設計和最佳化的雲原生多模超融合資料庫，是日誌、監控、賬單、廣告、社交、出行、風控等情境首選資料庫，也是為阿里巴巴核心業務提供支撐的資料庫之一。詳情請參見什麼是雲原生多模資料庫Lindorm。

前提條件

• 已經建立了Lindorm寬表引擎以及資料表，詳情請參見建立執行個體。

• Lindorm叢集與Flink全託管叢集處於網路連通的環境下，例如在同一個VPC下。

使用限制

僅Flink計算引擎VVR 4.0.8及以上版本支援Lindorm。

文法結構

CREATE TABLE white_list (
id varchar,
name varchar,
age int,
PRIMARY KEY (id) NOT ENFORCED
) WITH (
'connector' = 'lindorm',
'seedserver' = '<yourSeedServer>',
'namespace' = '<yourNamespace>',
'username' = '<yourUsername>',
'password' = '<yourPassword>',
'tableName' = '<yourTableName>',
'columnFamily' = '<yourColumnFamily>'
);

WITH參數

動態表模式

動態表模式適用於在表定義中並未指定列名的情況，根據作業運行情況動態建立資料列並插入的情境。例如統計每天每小時的交易量，以天作為主鍵，小時作為列，每個小時的資料都是動態產生的，樣本如下。

動態表需要遵循特殊的DDL定義。其主鍵需要定義為前若干列，最後兩列中前一列的值作為列名變數，最後一列的值作為該列對應的值，且要求最後兩列的類型均為varchar。程式碼範例如下。

CREATE TABLE lindorm_dynamic_output(
pk1varchar,
pk2varchar,
pk3varchar,
c1varchar,
c2varchar,
PRIMARYKEY(pk1,pk2,pk3)notenforced
) WITH (
'connector' = 'lindorm',
'seedserver' = '<yourSeedServer>',
'namespace' = '<yourNamespace>',
'username' = '<yourUsername>',
'password' = '<yourPassword>',
'tableName' = '<yourTableName>',
'columnFamily' = '<yourColumnFamily>'
);

上述定義中，pk1、pk2、pk3為主鍵，c1、c2為動態表模式所必須的兩列且一定為最後兩列，不可存在其他的非主鍵的列。每次寫入資料時，會在主鍵<pk1, pk2, pk3>對應的條目中添加或更改一列，列名為c1的值，該列的值為c2的值。每次一條資料來臨時，只會添加或更改一列對應的值，其他列的值不會改變。

類型映射

程式碼範例

CREATE TEMPORARY TABLE example_source(
 id INT,
 proc_time AS PROCTIME()
) WITH (
 'connector' = 'datagen',
 'number-of-rows' = '10',
 'fields.id.kind' = 'sequence',
 'fields.id.start' = '0',
 'fields.id.end' = '9'
);

CREATE TEMPORARY TABLE lindorm_hbase_dim(
 `id` INT,
 `name` VARCHAR,
 `birth` VARCHAR,
 PRIMARY KEY (id) NOT ENFORCED
) WITH (
 'connector'='lindorm',
 'tablename'='${lindorm_dim_table}',
 'seedserver'='${lindorm_seed_server}',
 'namespace'='default',
 'username'='${lindorm_username}',
 'password'='${lindorm_username}'
);

CREATE TEMPORARY TABLE lindorm_hbase_sink(
 `id` INT,
 `name` VARCHAR,
 `birth` VARCHAR,
 PRIMARY KEY (id) NOT ENFORCED
) WITH (
 'connector'='lindorm',
 'tablename'='${lindorm_sink_table}',
 'seedserver'='${lindorm_seed_server}',
 'namespace'='default',
 'username'='${lindorm_username}',
 'password'='${lindorm_username}'
);

INSERT INTO lindorm_hbase_sink
SELECT example_source.id as id, lindorm_hbase_dim.name as name, lindorm_hbase_dim.birth as birth
FROM example_source JOIN lindorm_hbase_dim FOR SYSTEM_TIME AS OF PROCTIME() ON example_source.id = lindorm_hbase_dim.id;