背景信息

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;