Realtime Compute for Apache 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的設定方式，詳情請參見資料分桶。

• 自動分區與分區管理原則，詳情請參見資料分區。