sparksql中的broadcast join和prestodb中的dynamic filter比較

今天在prestodb的qq群裡看到有人提到說一個子查詢在presto中非常慢：

SELECT *
FROM his_data_opt
WHERE act_no IN (
        SELECT act_no
        FROM id_act_map
        WHERE id_number = '726067685144725'
        );
           

可以看出，這是一個普通的非相關子查詢，如果内部子查詢經過過濾條件隻剩幾條，那麼整個查詢應該非常完美的在幾秒中出結果，結果卻卡着不動。原來是presto join處理的問題，對于普通的where條件，外部查詢會把這個where條件下推，進行表過濾，但現在外表這個過濾條件是一個動态生成的條件，presto在進行上層的邏輯計劃優化時，不知道這個動态生成的條件到底會産生多少條結果，于是presto把外部表進行了全表掃描，這在presto中成為dynamic filter，目前有人提了PR，還沒有合并到主版本中。來看presto中的執行計劃，查詢語句如下：

explain

SELECT *
FROM nation
WHERE regionkey IN (
        SELECT regionkey
        FROM region
        WHERE regionkey = 
        );
           

執行計劃：

從圖中可以看出，presto對region表進行了謂詞下推，而對nation表則沒有生成Filter這樣的實體操作符。同時可以看出，即使對于region和nation這樣隻有幾十行的表，presto依然會對這兩個表進行repartition，即shuffle操作，實體操作符是RemoteExchange，這代價也忒大了點吧，難怪會慢的不行。另外可以看出，對于這樣的子查詢，presto把它解析成了一個semi join來處理，即子查詢中如果有重複的regionkey被查出來，隻傳回一個即可，這滿足semi join的語義。

那麼來看看sparksql中怎麼處理這個問題的吧，同樣的表，同樣的sql語句，來看sparksql中的執行計劃：

從圖中可以看出，通過Broadcast實作了連接配接查詢，是不是表本身太小了導緻的呢？換大表試試：

explain

SELECT *
FROM orders
WHERE o_orderkey IN (
        SELECT l_orderkey
        FROM lineitem
        WHERE l_orderkey = 
        );
           

orders表150萬行，lineitem表600萬行，執行計劃如下：

發現還是broadcast實作的join，于是直接執行了一下，然後去yarn上看看DAG圖，如下：

可以看出，對lineitem首先進行過濾，過濾完後隻剩5條，然後把這5條資料進行broadcast，和presto進行比較的話，相當于在這裡把一個dynamic filter廣播到了叢集的所有節點上。然後用這個條件去查詢外表，就會快很多了。可見，presto和sparksql在join的實作上，還有有一定差距的，presto實作了類似hadoop裡的map-side join，才有可能使用dynamic filter。沒有對比就沒有傷害啊！

還沒完，prestodb的社群裡，來看看上邊提到的這個dynamic filter相關的PR：

https://github.com/prestodb/presto/issues/7428

SELECT 
    count(*)
FROM 
    fact_table a 
JOIN d_date b 
ON a.date_key = b.date_key 
WHERE 
    b.year = ;
           

pr中有這個查詢語句，雖然不像上邊那樣的子查詢，但是卻可以像上邊子查詢那樣來分析，比如，b.year = 2017這個查詢條件下推到b表後，b表過濾後的資料條數非常少，比如隻有一條，那麼如果用這一條再和a表join，那麼查詢将會非常快，可惜，presto中還是因為沒有實作broadcast，導緻其不能完成這樣的優化。