共計(jì) 3503 個(gè)字符，預(yù)計(jì)需要花費(fèi) 9 分鐘才能閱讀完成。
如何進(jìn)行 MaxCompute full outer join 改寫 left anti join 的實(shí)踐分析，針對(duì)這個(gè)問題，這篇文章詳細(xì)介紹了相對(duì)應(yīng)的分析和解答，希望可以幫助更多想解決這個(gè)問題的小伙伴找到更簡單易行的方法。
簡介：
ods 層數(shù)據(jù)同步時(shí)經(jīng)常會(huì)遇到增全量合并的模型，即 T - 1 天增量表 + T- 2 全量表 = T- 1 全量表。可以通過 full outer join 腳本來完成合并，但是數(shù)據(jù)量很大時(shí)非常消耗資源。下面將為您介紹在做增量數(shù)據(jù)的增加、更新時(shí)如何通過 full outer join 改寫 left anti join 來實(shí)現(xiàn)的最佳實(shí)踐。
背景
ods 層數(shù)據(jù)同步時(shí)經(jīng)常會(huì)遇到增全量合并的模型，即 T - 1 天增量表 + T- 2 全量表 = T- 1 全量表。可以通過 full outer join 腳本來完成合并，但是數(shù)據(jù)量很大時(shí)非常消耗資源。
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 種寫法一樣，都是優(yōu)先取 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;這種寫法可實(shí)現(xiàn)新增和更新操作：
新增是指增量表中新出現(xiàn)的數(shù)據(jù)，而全量表中沒有；
更新是指增量表和全量表中都有的數(shù)據(jù)，但優(yōu)先取增量表的數(shù)據(jù)，覆蓋歷史表的數(shù)據(jù)。
如下圖所示，R2_1 是增量表當(dāng)天去重后增量數(shù)據(jù)，M3 是全量表前一天的數(shù)據(jù)，而 J4_2_3 則是 full outer join 的執(zhí)行圖。
將 J4_2_3 展開會(huì)發(fā)現(xiàn)里面將增量和全量進(jìn)行了 merge join，當(dāng)數(shù)據(jù)量很大（1288 億條）時(shí)會(huì)產(chǎn)生很大的 shuffle 開銷。此時(shí)優(yōu)化方案就是將 full outer join 改成 union all，從而避免 join shuffle。
優(yōu)化模型
結(jié)論：full outer join 改成 hash cluster + left join +union all 可以有效地降低計(jì)算成本，且有兩種應(yīng)用場(chǎng)景。先將模型進(jìn)行抽象，假設(shè)有 a 和 b 兩個(gè)表，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) ) -- 全量場(chǎng)景 1: 只合并新增數(shù)據(jù)到全量表
left anti join 相當(dāng)于 not in，增量 not in 全量, 過濾后只剩下完全新增的 id，對(duì)全量中已有的 id 不修改：
-- 查詢完全新增的 idselect * from a left anti join b on a.id=b.id ;-- 結(jié)果如下 +------------+------+| id | name |
+------------+------+| 7 | 777 |
+------------+------+-- 完全新增的合并全量表 select * from a -- 增量表 left anti join b on a.id=b.id
union all select * from b -- 全量表 -- 結(jié)果如下 +------------+------+| id | name |
+------------+------+| 1 | |
| 2 | 222 |
| 3 | 333 |
| 4 | 444 |
| 7 | 777 |
+------------+------+場(chǎng)景 2: 合并新增數(shù)據(jù)到全量表，且更新歷史數(shù)據(jù)
全量 not in 增量, 過濾后只剩下歷史的 id，然后 union all 增量，既新增也修改
-- 查詢歷史全量數(shù)據(jù) select * from b left anti join a on a.id=b.id;-- 結(jié)果如下 +------------+------+| id | name |
+------------+------+| 3 | 333 |
| 4 | 444 |
+------------+------+-- 合并新增數(shù)據(jù)到全量表，且更新歷史數(shù)據(jù) select * from b -- 全量表 left anti join a on a.id=b.idunion all select * from a ; -- 增量表 -- 結(jié)果如下 +------------+------+| id | name |
+------------+------+| 1 | 111 |
| 2 | two |
| 7 | 777 |
| 3 | 333 |
| 4 | 444 |
+------------+------+優(yōu)化實(shí)踐
步驟 1：表屬性修改
表、作業(yè)屬性修改, 對(duì)原來的表、作業(yè)進(jìn)行屬性優(yōu)化，可以提升優(yōu)化效果。
set odps.sql.reducer.instances=3072; -- 可選。默認(rèn)最大 1111 個(gè) reducer,1111 哈希桶。alter table table_name clustered by(contact_id) sorted by(contact_id) into 3072 buckets;-- 必選步驟 2：按照上述模型的場(chǎng)景 1 或者 場(chǎng)景 2 進(jìn)行代碼改造。
這里先給出代碼改造后的資源消耗對(duì)比：
原來的 full outer jionleft anti join 初始化原來的 full outer jionleft anti join 第二天以后
時(shí)間消耗 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
可以發(fā)現(xiàn) hash cluster 分桶操作在初始化有額外的開銷，主要是按主鍵進(jìn)行散列和排序，但是這是值得的，可一勞永逸，后續(xù)的讀取速度非常快。以前每天跑需要 8 小時(shí)，現(xiàn)在除了分桶初始化需要 1 小時(shí)，以后每天實(shí)際只需要 30 分鐘。
初始化執(zhí)行圖
圖 1：
M2 是讀全量表。
M4 是讀取增量表, 在場(chǎng)景 2 的模型中增量表被讀取了兩次，其中：
R5_4 是對(duì)主鍵去重（row_number）后用于后面的 union all，里面包含了所有的增量數(shù)據(jù)；
R1_4 是對(duì)主鍵去重（row_number）后用于 left anti join，里面只包含了主鍵。
J3_1_2 是 left anti join, 將它展開后看到這里還是有 mergJoin，但是這只是初始化的操作，后面每天就不會(huì)有了。展開后如圖 2。
R6_3_5 是將增量和全量進(jìn)行 union all，展開后如圖 3。
R7_6 則是將索引信息寫入元數(shù)據(jù)，如圖 3 的 MetaCollector1 會(huì)在 R7_6 中 sink。
因此：圖 1 中除了 R5_4 和 R1_4 是去重必須的，有 shuffle。還有 J3_1_2 和 R6_3_5 這兩個(gè)地方有 shuffle。
圖 2：
第二天以后的執(zhí)行圖
R3_2 和 R1_2 是對(duì)增量去重必要對(duì)操作，有 shuffle，這里忽略。
初始化執(zhí)行圖的 J3_1_2 和 R6_3_5 已經(jīng)被合并到了 M4_1_3，將其展開后如圖 2。即 left anti join 和 union all 這兩步操作在一個(gè)階段完成了，且這個(gè)階段是 Map 任務(wù)（M4_1_3），而不是 Join 任務(wù)或 Reduce 任務(wù)。而且全量表不在單獨(dú)占用一個(gè) Map 任務(wù)，也被合并到了 M4_1_3，因此整個(gè)過程下來沒有 shuffle 操作，速度提升非常明顯。也就是說只需要一個(gè) M4_1_3 就能完成所有到操作，直接 sink 到表。
R5_4 則是將索引信息寫入元數(shù)據(jù)，如圖 2 的 MetaCollector1 會(huì)在 R5_4 中 sink。
圖 2：
關(guān)于如何進(jìn)行 MaxCompute full outer join 改寫 left anti join 的實(shí)踐分析問題的解答就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，如果你還有很多疑惑沒有解開，可以關(guān)注丸趣 TV 行業(yè)資訊頻道了解更多相關(guān)知識(shí)。
