Flink SQL链式Join引发状态膨胀的解决方法咨询

问题描述 #

使用Flink 1.19.1编写如下SQL：

select `from all tables`
FROM table1 i
LEFT JOIN table2 ip
ON i.tenantId = ip.tenantId AND i.id = ip.id
    LEFT JOIN table2 t
    ON i.tenantId = t.tenantId AND i.id = t.id
    LEFT JOIN table3 iap
    ON i.tenantId = iap.tenantId AND i.id = iap.id
    LEFT JOIN table4 ia
    ON i.tenantId = ia.tenantId AND i.id = ia.id
    LEFT JOIN table5 iae
    ON i.tenantId = iae.tenantId AND i.id = iae.id
    LEFT JOIN table6 iss
    ON i.tenantId = iss.tenantId AND i.id = iss.id
    LEFT JOIN table7 io
    ON i.tenantId = io.tenantId AND i.id = io.id
    LEFT JOIN table8 ipa
    ON i.tenantId = ipa.tenantId AND i.id = ipa.id
    LEFT JOIN table9 iara
    ON i.tenantId = iara.tenantId AND i.id = iara.id

生成的Join执行计划为链式结构，每个Join会存储前序Join结果加新关联数据，导致状态冗余膨胀，需要优化方案。

优化方案 #

1. 采用Broadcast Join（广播连接） #

如果table2到table9属于数据量较小的维度表，可通过广播提示让Flink将这些表的数据广播到每个TaskManager，使每个维度表直接与主表table1关联，避免链式Join的状态累积。修改SQL示例：

select `from all tables`
FROM table1 i
LEFT JOIN /*+ BROADCAST(ip) */ table2 ip
ON i.tenantId = ip.tenantId AND i.id = ip.id
LEFT JOIN /*+ BROADCAST(t) */ table2 t
ON i.tenantId = t.tenantId AND i.id = t.id
LEFT JOIN /*+ BROADCAST(iap) */ table3 iap
ON i.tenantId = iap.tenantId AND i.id = iap.id
-- 剩余JOIN语句同理添加对应BROADCAST提示

2. 启用Join重写优化 #

开启Flink的Join重写优化，让优化器自动将链式Join转换为星型Join（所有维度表直接关联主表）。可通过配置项或SQL提示开启：

• 配置文件中设置：table.optimizer.join-reorder-enabled: true
• 或在SQL中添加全局优化提示：

/*+ OPTIMIZER_REORDER_JOINS(true) */
select `from all tables`
FROM table1 i
-- 后续JOIN语句不变

3. 使用Lookup Join（针对外部存储维度表） #

若table2到table9是存储在MySQL、HBase等外部系统的维度表，改用Lookup Join替代普通Left Join。Lookup Join不会存储全量维度数据，仅缓存热点数据，需关联时实时查询外部系统，大幅降低状态占用。示例：

-- 先定义带Lookup配置的维度表
CREATE TABLE table2 (
    tenantId STRING,
    id STRING,
    -- 其他字段
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://your-mysql-host:3306/db',
    'table-name' = 'table2',
    'lookup.cache.max-rows' = '10000', -- 热点数据缓存行数
    'lookup.cache.ttl' = '1h' -- 缓存过期时间
);

-- 使用Lookup Join关联
select `from all tables`
FROM table1 i
LEFT JOIN table2 FOR SYSTEM_TIME AS OF i.proctime ip
ON i.tenantId = ip.tenantId AND i.id = ip.id
-- 剩余维度表关联同理修改为Lookup Join

4. 配置状态TTL自动清理过期数据 #

若业务允许清理历史关联数据，可为Join算子配置状态TTL，自动删除过期状态，减少冗余。可通过全局配置或SQL提示设置：

• 全局配置：table.exec.state.ttl: 1d（设置1天过期）
• SQL提示方式：

select /*+ STATE_TTL('1d') */ `from all tables`
FROM table1 i
-- 后续JOIN语句不变

内容的提问来源于stack exchange，提问作者hitesh