如何进行MaxCompute full outer join改写left anti join的实践分析

46次阅读

共计 3503 个字符，预计需要花费 9 分钟才能阅读完成。

如何进行 MaxCompute full outer join 改写 left anti join 的实践分析，针对这个问题，这篇文章详细介绍了相对应的分析和解答，希望可以帮助更多想解决这个问题的小伙伴找到更简单易行的方法。

简介：

ods 层数据同步时经常会遇到增全量合并的模型，即 T - 1 天增量表 + T- 2 全量表 = T- 1 全量表。可以通过 full outer join 脚本来完成合并，但是数据量很大时非常消耗资源。下面将为您介绍在做增量数据的增加、更新时如何通过 full outer join 改写 left anti join 来实现的最佳实践。

背景

ods 层数据同步时经常会遇到增全量合并的模型，即 T - 1 天增量表 + T- 2 全量表 = T- 1 全量表。可以通过 full outer join 脚本来完成合并，但是数据量很大时非常消耗资源。

insert overwrite table tb_test partition(ds= ${bizdate} )select case when a.id is not null then a.id esle b.id end as id 
 ,if(a.name is not null, a.name, b.name) as name ,coalesce(a.age, b.age) as age 
 -- 这 3 种写法一样，都是优先取 delta 表的字段 from( select * from tb_test_delta where ds= ${bizdate} ) afull outer join( select * from tb_test where ds= ${bizdate-1} ) bon a.id =b.id;

这种写法可实现新增和更新操作：

新增是指增量表中新出现的数据，而全量表中没有；

更新是指增量表和全量表中都有的数据，但优先取增量表的数据，覆盖历史表的数据。
如下图所示，R2_1 是增量表当天去重后增量数据，M3 是全量表前一天的数据，而 J4_2_3 则是 full outer join 的执行图。

将 J4_2_3 展开会发现里面将增量和全量进行了 merge join，当数据量很大（1288 亿条）时会产生很大的 shuffle 开销。此时优化方案就是将 full outer join 改成 union all，从而避免 join shuffle。

优化模型

结论：full outer join 改成 hash cluster + left join +union all 可以有效地降低计算成本，且有两种应用场景。先将模型进行抽象，假设有 a 和 b 两个表，a 是增量表，b 是全量表：

with 
 a as ( select * from values (1, 111)
 ,(2, two)
 ,(7, 777) as (id,name) ) -- 增量,b as ( select * from values (1,)
 ,(2, 222)
 ,(3, 333)
 ,(4, 444) as (id,name) ) -- 全量

场景 1: 只合并新增数据到全量表

left anti join 相当于 not in，增量 not in 全量, 过滤后只剩下完全新增的 id，对全量中已有的 id 不修改：

-- 查询完全新增的 idselect * from a left anti join b on a.id=b.id ;-- 结果如下 +------------+------+| id | name |
+------------+------+| 7 | 777 |
+------------+------+

-- 完全新增的合并全量表 select * from a -- 增量表 left anti join b on a.id=b.id 
union all select * from b -- 全量表 -- 结果如下 +------------+------+| id | name |
+------------+------+| 1 | |
| 2 | 222 |
| 3 | 333 |
| 4 | 444 |
| 7 | 777 |
+------------+------+

场景 2: 合并新增数据到全量表，且更新历史数据

全量 not in 增量, 过滤后只剩下历史的 id，然后 union all 增量，既新增也修改

-- 查询历史全量数据 select * from b left anti join a on a.id=b.id;-- 结果如下 +------------+------+| id | name |
+------------+------+| 3 | 333 |
| 4 | 444 |
+------------+------+

-- 合并新增数据到全量表，且更新历史数据 select * from b -- 全量表 left anti join a on a.id=b.idunion all select * from a ; -- 增量表 -- 结果如下 +------------+------+| id | name |
+------------+------+| 1 | 111 |
| 2 | two |
| 7 | 777 |
| 3 | 333 |
| 4 | 444 |
+------------+------+

优化实践

步骤 1：表属性修改

表、作业属性修改, 对原来的表、作业进行属性优化，可以提升优化效果。

set odps.sql.reducer.instances=3072; -- 可选。默认最大 1111 个 reducer,1111 哈希桶。alter table table_name clustered by(contact_id) sorted by(contact_id) into 3072 buckets;-- 必选

步骤 2：按照上述模型的场景 1 或者场景 2 进行代码改造。

这里先给出代码改造后的资源消耗对比：

原来的 full outer jionleft anti join 初始化原来的 full outer jionleft anti join 第二天以后
时间消耗 8h40min38s1h5min48s7h42min30s32min30scpu 消耗 29666.02 Core * Min65705.30 Core * Min31126.86 Core * Min30589.29 Core * Minmem 消耗 109640.80 GB * Min133922.25 GB * Min114764.80 GB * Min65509.28 GB * Min

可以发现 hash cluster 分桶操作在初始化有额外的开销，主要是按主键进行散列和排序，但是这是值得的，可一劳永逸，后续的读取速度非常快。以前每天跑需要 8 小时，现在除了分桶初始化需要 1 小时，以后每天实际只需要 30 分钟。

初始化执行图

图 1：

M2 是读全量表。

M4 是读取增量表, 在场景 2 的模型中增量表被读取了两次，其中：

R5_4 是对主键去重（row_number）后用于后面的 union all，里面包含了所有的增量数据；

R1_4 是对主键去重（row_number）后用于 left anti join，里面只包含了主键。

J3_1_2 是 left anti join, 将它展开后看到这里还是有 mergJoin，但是这只是初始化的操作，后面每天就不会有了。展开后如图 2。

R6_3_5 是将增量和全量进行 union all，展开后如图 3。

R7_6 则是将索引信息写入元数据，如图 3 的 MetaCollector1 会在 R7_6 中 sink。
因此：图 1 中除了 R5_4 和 R1_4 是去重必须的，有 shuffle。还有 J3_1_2 和 R6_3_5 这两个地方有 shuffle。

图 2：

第二天以后的执行图

R3_2 和 R1_2 是对增量去重必要对操作，有 shuffle，这里忽略。

初始化执行图的 J3_1_2 和 R6_3_5 已经被合并到了 M4_1_3，将其展开后如图 2。即 left anti join 和 union all 这两步操作在一个阶段完成了，且这个阶段是 Map 任务（M4_1_3），而不是 Join 任务或 Reduce 任务。而且全量表不在单独占用一个 Map 任务，也被合并到了 M4_1_3，因此整个过程下来没有 shuffle 操作，速度提升非常明显。也就是说只需要一个 M4_1_3 就能完成所有到操作，直接 sink 到表。

R5_4 则是将索引信息写入元数据，如图 2 的 MetaCollector1 会在 R5_4 中 sink。

图 2：

关于如何进行 MaxCompute full outer join 改写 left anti join 的实践分析问题的解答就分享到这里了，希望以上内容可以对大家有一定的帮助，如果你还有很多疑惑没有解开，可以关注丸趣 TV 行业资讯频道了解更多相关知识。

正文完