Realtime Compute for Apache Flink：維表JOIN語句

調優參數分類與傳遞方式

維表 JOIN 調優涉及三類參數，每類參數的傳遞方式不同，用錯方式 hint 會被靜默忽略。

通過 OPTIONS hint 覆蓋連接器選項

OPTIONS() hint 用於在 SQL 中臨時覆蓋某張表的 WITH 選項，無需修改建表語句。常用於調整 Cache 行為。

SELECT t.id, t.name, w.phoneNumber
FROM kafka_input AS t
LEFT JOIN phoneNumber /*+ OPTIONS(
  'lookup.cache' = 'PARTIAL',
  'lookup.partial-cache.max-rows' = '1000'
) */ FOR SYSTEM_TIME AS OF PROCTIME() AS w
ON t.name = w.name;

OPTIONS 中傳入的 key 必須為連接器本身支援的 WITH 選項。

通過 LOOKUP hint 配置 Lookup 行為

LOOKUP hint 功能與社區保持一致，用於在單個 Join 操作上配置維表的同步、非同步、重試及 shuffle 策略。詳情參見 Apache Flink Lookup Hint。

• 僅 VVR 8.0 及以上版本支援 LOOKUP hint。

• 僅 VVR 8.0.8 及以上版本支援通過 'shuffle' = 'true' 配置 shuffle 策略。

• VVR 8.0 以上支援使用別名；如果維表定義了別名，hint 中必須使用別名。

-- 只對維表 dim1 配置維表聯結 shuffle 策略
SELECT /*+ LOOKUP('table'='dim1', 'shuffle' = 'true') */ ...
FROM src AS T
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() ON T.a = dim1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() ON T.b = dim2.b;

-- 同時對維表 dim1、dim2 配置維表聯結 shuffle 策略
SELECT /*+ LOOKUP('table'='dim1', 'shuffle' = 'true'), LOOKUP('table'='dim2', 'shuffle' = 'true') */ ...
FROM src AS T
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() ON T.a = dim1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() ON T.b = dim2.b;

-- 維表 dim1 使用別名 D1 時，hint 中必須使用別名
SELECT /*+ LOOKUP('table'='D1', 'shuffle' = 'true') */ ...
FROM src AS T
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() AS D1 ON T.a = D1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() AS D2 ON T.b = D2.b;

-- 同時對維表 dim1、dim2 通過別名配置維表聯結 shuffle 策略
SELECT /*+ LOOKUP('table'='D1', 'shuffle' = 'true'), LOOKUP('table'='D2', 'shuffle' = 'true') */ ...
FROM src AS T
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() AS D1 ON T.a = D1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() AS D2 ON T.b = D2.b;

通過 SET 配置 Job 級 TableConfig

table.exec.* 等 TableConfig 屬於 Job 級全域配置，必須通過 SET 命令傳遞。LOOKUP hint 和 OPTIONS hint 均不接受這類參數。

常用維表 JOIN 相關 TableConfig：

-- 在作業開頭設定 Job 級預設值
SET 'table.exec.async-lookup.buffer-capacity' = '3072';
SET 'table.exec.async-lookup.timeout' = '180s';

INSERT INTO sink_table
SELECT ...
FROM src_table
LEFT JOIN dim_table FOR SYSTEM_TIME AS OF PROCTIME() AS d
ON src_table.key = d.key;

通過 Join 策略 Hints 控制 Shuffle

維表 Join 策略 Hints 用於控制 Shuffle 行為，包括 SHUFFLE_HASH、REPLICATED_SHUFFLE_HASH 和 SKEW。維表 Cache 策略和聯結策略的適用情境如下。

-- 只對維表 dim1 開啟 SHUFFLE_HASH 聯結
SELECT /*+ SHUFFLE_HASH(dim1) */ ...

-- 同時對維表 dim1、dim2 均開啟 SHUFFLE_HASH 聯結
SELECT /*+ SHUFFLE_HASH(dim1, dim2) */ ...

-- 維表 dim1 使用別名 D1 時，hint 中必須使用別名
SELECT /*+ SHUFFLE_HASH(D1) */ ...

SELECT /*+ REPLICATED_SHUFFLE_HASH(dim1) */ ...

SELECT /*+ SKEW(src) */  ...

樣本一：基礎維表 JOIN

最基礎的寫法，使用 MySQL 維表補全 Kafka 流的欄位。無任何調優 hint。

CREATE TEMPORARY TABLE kafka_input (
  id   BIGINT,
  name VARCHAR,
  age  BIGINT
) WITH (
  'connector' = 'kafka',
  'topic' = '<yourTopic>',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

CREATE TEMPORARY TABLE phoneNumber (
  name        VARCHAR,
  phoneNumber BIGINT,
  PRIMARY KEY (name) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'hostname' = '<yourHostname>',
  'port' = '3306',
  'username' = '<yourUsername>',
  'password' = '<yourPassword>',
  'database-name' = '<yourDatabaseName>',
  'table-name' = '<yourTableName>'
);

CREATE TEMPORARY TABLE result_infor (
  id          BIGINT,
  phoneNumber BIGINT,
  name        VARCHAR
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO result_infor
SELECT
  t.id,
  w.phoneNumber,
  t.name
FROM kafka_input AS t
JOIN phoneNumber FOR SYSTEM_TIME AS OF PROCTIME() AS w
ON t.name = w.name;

樣本二：通過 OPTIONS hint 啟用 Cache 並通過 LOOKUP hint 配置非同步查詢

調優情境：維表 QPS 不足，需要在維表上開啟分區 Cache，同時啟用非同步 lookup 並增大緩衝隊列。

INSERT INTO user_behavior_wide
SELECT /*+ LOOKUP('table' = 't2', 'async' = 'true', 'capacity' = '3072') */
  t1.member_id AS member_id,
  t2.tag AS tag
FROM user_behavior_datagen AS t1
LEFT JOIN fluss.fluss.user_active_info /*+ OPTIONS(
    'lookup.cache' = 'PARTIAL',
    'lookup.partial-cache.max-rows' = '1000'
  ) */
FOR SYSTEM_TIME AS OF PROCTIME() AS t2
ON t1.member_id = t2.member_id;

• 連接器選項（lookup.cache、lookup.partial-cache.max-rows）通過 OPTIONS() hint 寫在維表後，覆蓋建表時的 WITH 選項。

• Lookup 行為（async、capacity）通過 LOOKUP() hint 寫在 SELECT 之後。OPTIONS 與 LOOKUP 是兩類不同的 hint，必須分別書寫，混寫會導致 hint 靜默失效。

樣本三：通過 SET 調整 Job 級非同步 lookup 緩衝

調優情境：作業內多個維表 JOIN 都需要更大的非同步 lookup 緩衝，希望全域生效，並配合 LOOKUP hint 啟用 async。

SET 'table.exec.async-lookup.buffer-capacity' = '3072';
SET 'table.exec.async-lookup.timeout' = '180s';

INSERT INTO user_behavior_wide
SELECT /*+ LOOKUP('table' = 't2', 'async' = 'true') */
  t1.member_id,
  t2.tag
FROM user_behavior_datagen AS t1
LEFT JOIN fluss.fluss.user_active_info
FOR SYSTEM_TIME AS OF PROCTIME() AS t2
ON t1.member_id = t2.member_id;

• table.exec.async-lookup.buffer-capacity 等參數為 Job 級 TableConfig，僅可通過 SET 設定，寫入 OPTIONS 或 LOOKUP hint 均不生效。

• 若 LOOKUP hint 中未指定 capacity，則使用 SET 設定的全域值；若同時指定，則單個 JOIN 上的 hint 優先生效。