实时计算Flink版：维表关联

维表关联（Lookup Join）

• Lookup Join是维表 Join的一种实现方式，详情请参见维表JOIN语句。

• 仅Fluss的主键表支持作为维表进行join，且join条件需包含维表的全部主键。如果建表时将Bucket Key设置为join key，join条件也可以是主键的前缀子集，详情请参见基于Bucket Key的维表查询。

• 维表为分区主键表时，join条件还需包含分区键。

• Fluss的维表关联默认使用异步模式，吞吐更高，可通过 SQL Hint设置'lookup.async' = 'false'，将维表关联改成同步模式。

关联两个维表的SQL示例

创建两个维表

USE CATALOG `fluss_catalog`;
USE `my_db`;

CREATE TABLE `orders` (
  `o_orderkey` INT NOT NULL,
  `o_custkey` INT NOT NULL,
  `o_orderstatus` CHAR(1) NOT NULL,
  `o_totalprice` DECIMAL(15, 2) NOT NULL,
  `o_orderdate` DATE NOT NULL,
  `o_orderpriority` CHAR(15) NOT NULL,
  `o_clerk` CHAR(15) NOT NULL,
  `o_shippriority` INT NOT NULL,
  `o_comment` STRING NOT NULL,
  `o_dt` STRING NOT NULL,
  PRIMARY KEY (o_orderkey) NOT ENFORCED
);

CREATE TABLE `customer` (
  `c_custkey` INT NOT NULL,
  `c_name` STRING NOT NULL,
  `c_address` STRING NOT NULL,
  `c_nationkey` INT NOT NULL,
  `c_phone` CHAR(15) NOT NULL,
  `c_acctbal` DECIMAL(15, 2) NOT NULL,
  `c_mktsegment` CHAR(10) NOT NULL,
  `c_comment` STRING NOT NULL,
  PRIMARY KEY (c_custkey) NOT ENFORCED
);

两个维表关联

CREATE TEMPORARY TABLE `lookup_join_sink` (
  `order_key`        INT NOT NULL,
  `order_totalprice` DECIMAL(15, 2) NOT NULL,
  `customer_name`    STRING NOT NULL,
  `customer_address` STRING NOT NULL
) WITH ('connector' = 'blackhole');

-- 异步模式（默认）
INSERT INTO `lookup_join_sink`
SELECT `o`.`o_orderkey`, `o`.`o_totalprice`, `c`.`c_name`, `c`.`c_address`
FROM (SELECT `orders`.*, PROCTIME() AS `ptime` FROM `orders`) AS `o`
LEFT JOIN `customer` FOR SYSTEM_TIME AS OF `o`.`ptime` AS `c`
ON `o`.`o_custkey` = `c`.`c_custkey`;

同步模式关联

-- 同步模式
INSERT INTO `lookup_join_sink`
SELECT `o`.`o_orderkey`, `o`.`o_totalprice`, `c`.`c_name`, `c`.`c_address`
FROM (SELECT `orders`.*, PROCTIME() AS `ptime` FROM `orders`) AS `o`
LEFT JOIN `customer` /*+ OPTIONS('lookup.async' = 'false') */ FOR SYSTEM_TIME AS OF `o`.`ptime` AS `c`
ON `o`.`o_custkey` = `c`.`c_custkey`;

分区表维表关联

维表为分区主键表时，join条件需包含全部主键，即主键与分区键。

-- 主键为 (c_custkey, dt)，分区键为 dt
CREATE TABLE `customer_partitioned` (
  `c_custkey` INT NOT NULL,
  `c_name` STRING NOT NULL,
  `c_address` STRING NOT NULL,
  `c_nationkey` INT NOT NULL,
  `c_phone` CHAR(15) NOT NULL,
  `c_acctbal` DECIMAL(15, 2) NOT NULL,
  `c_mktsegment` CHAR(10) NOT NULL,
  `c_comment` STRING NOT NULL,
  `dt` STRING NOT NULL,
  PRIMARY KEY (`c_custkey`, `dt`) NOT ENFORCED
)
PARTITIONED BY (`dt`)
WITH (
  'table.auto-partition.enabled' = 'true',
  'table.auto-partition.time-unit' = 'year'
);

join条件中除主键外，还需带上分区键。

INSERT INTO `lookup_join_sink`
SELECT `o`.`o_orderkey`, `o`.`o_totalprice`, `c`.`c_name`, `c`.`c_address`
FROM (SELECT `orders`.*, PROCTIME() AS `ptime` FROM `orders`) AS `o`
LEFT JOIN `customer_partitioned` FOR SYSTEM_TIME AS OF `o`.`ptime` AS `c`
ON `o`.`o_custkey` = `c`.`c_custkey` AND `o`.`o_dt` = `c`.`dt`;

分区表的详情请参见数据分区。

基于Bucket Key的维表查询（Prefix Lookup）

-- 主键为 (c_custkey, c_nationkey, dt)，Bucket Key 为 c_custkey，分区键为 dt
CREATE TABLE `customer_partitioned_with_bucket_key` (
  `c_custkey` INT NOT NULL,
  `c_name` STRING NOT NULL,
  `c_address` STRING NOT NULL,
  `c_nationkey` INT NOT NULL,
  `c_phone` CHAR(15) NOT NULL,
  `c_acctbal` DECIMAL(15, 2) NOT NULL,
  `c_mktsegment` CHAR(10) NOT NULL,
  `c_comment` STRING NOT NULL,
  `dt` STRING NOT NULL,
  PRIMARY KEY (`c_custkey`, `c_nationkey`, `dt`) NOT ENFORCED
)
PARTITIONED BY (`dt`)
WITH (
  'bucket.key' = 'c_custkey',
  'table.auto-partition.enabled' = 'true',
  'table.auto-partition.time-unit' = 'year'
);

-- join条件为主键前缀（不含分区键）+ 分区键
INSERT INTO `lookup_join_sink`
SELECT `o`.`o_orderkey`, `o`.`o_totalprice`, `c`.`c_name`, `c`.`c_address`
FROM (SELECT `orders`.*, PROCTIME() AS `ptime` FROM `orders`) AS `o`
LEFT JOIN `customer_partitioned_with_bucket_key` FOR SYSTEM_TIME AS OF `o`.`ptime` AS `c`
ON `o`.`o_custkey` = `c`.`c_custkey` AND `o`.`o_dt` = `c`.`dt`;

-- 主键为 (c_custkey, c_nationkey)，Bucket Key 为 c_custkey
CREATE TABLE `customer_with_bucket_key` (
  `c_custkey` INT NOT NULL,
  `c_name` STRING NOT NULL,
  `c_address` STRING NOT NULL,
  `c_nationkey` INT NOT NULL,
  `c_phone` CHAR(15) NOT NULL,
  `c_acctbal` DECIMAL(15, 2) NOT NULL,
  `c_mktsegment` CHAR(10) NOT NULL,
  `c_comment` STRING NOT NULL,
  PRIMARY KEY (`c_custkey`, `c_nationkey`) NOT ENFORCED
) WITH (
  'bucket.key' = 'c_custkey'
);

-- Prefix Lookup，异步模式（默认）
INSERT INTO `lookup_join_sink`
SELECT `o`.`o_orderkey`, `o`.`o_totalprice`, `c`.`c_name`, `c`.`c_address`
FROM (SELECT `orders`.*, PROCTIME() AS `ptime` FROM `orders`) AS `o`
LEFT JOIN `customer_with_bucket_key` FOR SYSTEM_TIME AS OF `o`.`ptime` AS `c`
ON `o`.`o_custkey` = `c`.`c_custkey`;

-- Prefix Lookup，同步模式
INSERT INTO `lookup_join_sink`
SELECT `o`.`o_orderkey`, `o`.`o_totalprice`, `c`.`c_name`, `c`.`c_address`
FROM (SELECT `orders`.*, PROCTIME() AS `ptime` FROM `orders`) AS `o`
LEFT JOIN `customer_with_bucket_key` /*+ OPTIONS('lookup.async' = 'false') */ FOR SYSTEM_TIME AS OF `o`.`ptime` AS `c`
ON `o`.`o_custkey` = `c`.`c_custkey`;

历史分区查询（Historical Partition Lookup）

ALTER TABLE `customer_partitioned_with_bucket_key` SET (
  'table.datalake.historical-partition.enabled' = 'true'
);

Insert If Not Exists

CREATE TABLE uid_mapping (
  uid VARCHAR NOT NULL,
  uid_int32 INT,PRIMARY KEY (uid) NOT ENFORCED
) WITH ('auto-increment.fields' = 'uid_int32','bucket.num' = '1');

-- 流表 ods_events 中的 uid 自动注册到字典表 uid_mapping，-- 并获取对应的整型 ID uid_int32SELECT
  ods.country,
  ods.prov,
  ods.city,
  ods.ymd,
  ods.uid,
  dim.uid_int32
FROM ods_events AS ods
JOIN uid_mapping /*+ OPTIONS('lookup.insert-if-not-exists' = 'true') */FOR SYSTEM_TIME AS OF ods.proctime AS dim
  ON dim.uid = ods.uid;

相关文档

• 维表关联支持的配置项，详情请参见Connector 参数。

• Bucket Key的设置方式，详情请参见数据分桶。

• 自动分区与分区管理策略，详情请参见数据分区。