实时计算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(
pk1 varchar,
pk2 varchar,
pk3 varchar,
c1 varchar,
c2 varchar,
PRIMARY KEY (pk1,pk2,pk3) NOT ENFORCED
) 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;

常见问题

Lindorm连接出现报错与解决方法