flink sql left join資料傾斜問題解決

1. 問題描述

原sql：

select a.user_id, a.其他字段, b.其他字段
from temp.user_log as a
left join user_info as b
on a.user_id = b.user_id
           

flink sql on hive，使用了left join去關聯字典表，由于字典表中某一個值對應的左邊記錄表中的資料條數非常非常多，是其他字典值的數十倍之多，是以在join時發生了資料傾斜，這一個字典值對應的所有記錄都跑到了一個并行度中，如下圖所示：

其他并行度接收的資料量遠遠小于第一個并行度。

2. 解決方案

2.1 字典表資料擴充

将字典表資料進行擴充。

比如字典表為使用者資訊，和記錄表的關聯字段為使用者id，則可以在使用者id後面添加從0~9的數字，将其擴充為原來的十倍，示例sql如下所示：

insert into temp.user_info_10
select concat(user_id, '0') as user_id, 其他字段值保持不變
from temp.user_info
;
insert into temp.user_info_10
select concat(user_id, '1') as user_id, 其他字段值保持不變
from temp.user_info
;
insert into temp.user_info_10
select concat(user_id, '2') as user_id, 其他字段值保持不變
from temp.user_info
;
...
           

上面一個10個sql語句，在用于關聯的字段“使用者id”後面拼接0~9的數字，将原字典表資料擴充為原來的10倍。

2.2 left join

當使用者日志表和使用者字典表關聯時，需要将使用者日志表的“使用者id”字段後面拼接上0~9的數字，兩個目的：

1. 和字典表進行關聯，能夠找到對應的值。由于兩邊都在字段值後面添加了指定範圍内的數字，是以肯定可以關聯上，而且結果和原sql一緻。
2. 将使用者日志記錄資料中的“使用者id”打散到不同的并行度，解決資料傾斜問題。

示例SQL：

select a.user_id, a.其他字段, b.其他字段
from temp.user_log as a
left join user_info_10 as b
on concat(a.user_id, cast(cast(rand(10) * 10 as int) as int)) = b.user_id
           

運作過程圖：

可以看到，和原sql相比，資料傾斜已經不是特别嚴重了。

3. 其他傾斜情況

如果記錄表中的某個使用者，在字典表中并不存在，但是這個使用者的的日志非常多，造成了嚴重的資料傾斜，此時就不用擴充字典表了，直接将對該使用者的id進行随機數拼接，然後打散該使用者的記錄，關聯時使用該使用者拼接之後的id進行關聯即可。

由于該使用者原id就無法關聯到字典表資料，是以打散之後也關聯不到字典表資料，對結果無影響。

示例sql如下：

select a.user_id, a.其他字段, b.其他字段
from
    (
    select 
        user_id,
        case
            when user_id = '記錄最多，但是字典表中沒有的使用者id' then concat(user_id, cast(cast(rand(10) * 10 as int) as string))
            else user_id end
        as user_id_join,
        其他字段
    from temp.user_log 
    ) as a
left join user_info as b
on a.user_id_join = b.user_id
;