如何進行MaxCompute full outer join改寫left anti join的實踐分析

如何進行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進行代碼改造。

這里先給出代碼改造后的資源消耗對比：

可以發現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的實踐分析問題的解答就分享到這里了，希望以上內容可以對大家有一定的幫助，如果你還有很多疑惑沒有解開，可以關注億速云行業資訊頻道了解更多相關知識。