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}

通过 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`;

Lookup Joins

Lookup Joins 是流式查询中的一种连接。它用于从 Arctic 查询的数据来打宽一个表。该连接要求事实表有一个 process time 属性。

create catalog arctic with (
    'type'='arctic',
    'metastore.url'='thrift://***:***/trino_online_env_hive'-- 具体地址可以查询 AMS
);

-- enrich each order with customer information
SELECT o.order_id, o.total, c.country, c.zip
FROM Orders AS o
  JOIN arctic.db_test.customer FOR SYSTEM_TIME AS OF o.proc_time AS c
    ON o.customer_id = c.id;

加速初始化维表

Lookup Join 算子将在本地维护一个 RocksDB 缓存，并根据 lookup.reloading.interval 拉取该表的最新数据更新到 rocksdb。Lookup join 算子将只提取必要的数据，所以你的分区的过滤条件对性能非常重要；Projection filter 下推也对加载性能有很大提升，如果不需要使用的维表字段可以不 Select。

Shuffle Hash Lookup join

开启以下配置可以加速维表初始化，提高维表 Cache 命中率，提升 Lookup Join 性能

-- 加速初始化，提高维表cache命中率
set 'table.lookup.hash-shuffle.enabled'='true';

限制

左表如果是有更新数据（update/delete）数据，需要左表的主键字段包含 lookup keys，否则左表数据经过 Hash shuffle 后数据有可能会乱序。

Arctic 维表参数说明