维表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 优先生效。