Arctic 连接器

简述

支持 Arctic 实时数据湖的读写

创建 Catalog

CREATE CATALOG slothTest WITH (
  'type'='arctic',
  'metastore.url'='thrift://127.0.0.1:14230/catalogName'
);

• type：catalog 类型
• metastore.url：元数据中心地址。thrift://{ip}:{port}/{catalogName}
• arctic.version：Flink 1.12 版本的可选配置项，表示是否指定 arctic 版本为 0.2.8 以上的高版本，当不填写该配置项时，默认为 0.2.8 的 arctic 版本，当需要指定为非 0.2.8 版本时，需要指定该配置项为 0.3

通过 SQL 读取

读取文件数据（Filestore）

非主键表

-- 在当前 session 中以流的模式运行 Flink 任务
SET execution.runtime-mode = streaming;

-- 打开动态表参数配置开关，让 Flink SQL 中配置的 hint options 生效
SET table.dynamic-table-options.enabled = true;

-- 读当前快照之后的增量数据
SELECT * FROM unkeyed /*+ OPTIONS('monitor-interval'='1s')*/ ;

Hint Options

主键表

-- 在当前 session 中以流的模式运行 Flink 任务
SET execution.runtime-mode = streaming;

-- 打开动态表参数配置开关，让 Flink SQL 中配置的 hint options 生效
SET table.dynamic-table-options.enabled = true;

-- 读全量数据及 changelog 中的 CDC 数据
SELECT * FROM keyed;

Hint Options

读取实时数据，即log数据

通过下面的 SQL 可以从 Arctic 表中读取到实时的流式数据：

-- 在当前 session 中以流的模式运行 Flink 任务
SET execution.runtime-mode = streaming;

-- 打开动态表参数配置开关，让 Flink SQL 中配置的 hint options 生效
SET table.dynamic-table-options.enabled=true;

SELECT * FROM test_table /*+ OPTIONS('arctic.read.mode'='log') */;

支持以下 Hint Options ：

通过 SQL 写入

Arctic 表支持通过 Flink Sql 往 Log 或 File 写入数据

INSERT OVERWRITE

当前仅支持非主键表的 INSERT OVERWRITE。INSERT OVERWRITE 只允许以 Flink Batch 的模式运行。 替换表中的数据，Overwrite 为原子操作。 分区会由查询语句中动态生成，这些分区的数据会被全量覆盖。

INSERT OVERWRITE unkeyed VALUES (1, 'a', '2022-07-01');

也支持覆盖指定分区的数据

INSERT OVERWRITE `arctic_catalog`.`arctic_db`.`unkeyed` PARTITION(data='2022-07-01') SELECT 5, 'b';

对于无分区的表，INSERT OVERWRITE 将覆盖表里的全量数据

INSERT INTO

对于 Arctic 表，可以指定往 File 或 Log（需在建表时 开启 Log 配置）写入数据。 对于 Arctic 主键表，写 File 会将 CDC 数据写入 ChangeStore 中.

-- 开启通过 SQL Hint 设置表参数
SET table.dynamic-table-options.enabled=true;

-- 同时往 Arctic 表的 Filestore, Logstore 中写入数据
INSERT INTO `arctic_catalog`.`arctic_db`.`test_table` 
    /*+ OPTIONS('arctic.emit.mode'='log,file') */
SELECT id, name from `source`;