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(rand() * 10 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
;