如何優化火花SQL運行它並行

Question

我是火花新手，使用有一個簡單的火花申請星火SQL/hiveContext到：

1. 從蜂巢表選擇數據（1個十億行）
2. 做一些過濾，聚合包括ROW_NUMBER在窗函數通過選擇第一行，組數（）和max（）等
3. 結果寫入到HBase的（幾百萬行）

我提交工作來運行它在紗線集羣（100個執行器）上，速度很慢，當我查看Spark UI中的DAG可視化時，似乎只有蜂巢表掃描任務並行運行，其餘步驟＃2和＃3僅運行在一個哪些實例可能應該能夠優化並行化？

該應用程序看起來像：

步驟1：

val input = hiveContext 
    .sql(
    SELECT 
      user_id 
      , address 
      , age 
      , phone_number 
      , first_name 
      , last_name 
      , server_ts 
     FROM 
     (  
      SELECT 
       user_id 
       , address 
       , age 
       , phone_number 
       , first_name 
       , last_name 
       , server_ts 
       , row_number() over 
       (partition by user_id, address, phone_number, first_name, last_name order by user_id, address, phone_number, first_name, last_name, server_ts desc, age) AS rn 
      FROM 
      ( 
       SELECT 
        user_id 
        , address 
        , age 
        , phone_number 
        , first_name 
        , last_name 
        , server_ts 
       FROM 
        table 
       WHERE 
        phone_number <> '911' AND 
        server_date >= '2015-12-01' and server_date < '2016-01-01' AND 
        user_id IS NOT NULL AND 
        first_name IS NOT NULL AND 
        last_name IS NOT NULL AND 
        address IS NOT NULL AND 
        phone_number IS NOT NULL AND 
      ) all_rows 
     ) all_rows_with_row_number 
     WHERE rn = 1) 

val input_tbl = input.registerTempTable(input_tbl) 

步驟2：

val result = hiveContext.sql(
    SELECT state, 
     phone_number, 
     address, 
     COUNT(*) as hash_count, 
     MAX(server_ts) as latest_ts 
    FROM 
    (SELECT 
     udf_getState(address) as state 
     , user_id 
     , address 
     , age 
     , phone_number 
     , first_name 
     , last_name 
     , server_ts 
    FROM 
     input_tbl) input 
    WHERE state IS NOT NULL AND state != '' 
    GROUP BY state, phone_number, address) 

步驟3：

result.cache() 
result.map(x => ...).saveAsNewAPIHadoopDataset(conf) 

的DAG可視化看起來像：

正如您所看到的，階段0中的「過濾器」，「項目」和「交換」只在一個實例中運行，stage1和stage2也是如此，所以如果問題是幾個問題和道歉啞：

1. 在每個執行器進行數據混洗之後，Driver中是否會出現「Filter」，「Project」和「Exchange」？
2. 什麼代碼映射到「過濾器」，「項目」和「交換」？
3. 我怎麼可以同時運行「過濾器」，「項目」和「交換」來優化性能？
4. 可以同時運行stage1和stage2嗎？

Answer 1

你不能正確讀取DAG圖 - 每個步驟是使用單箱並不意味着它不使用多個任務（因此核心）來計算可視化的事實步。

通過鑽入舞臺視圖中可以看到每個步驟使用了多少個任務，該視圖顯示該階段的所有任務。

例如，這裏有一個樣本DAG可視化與你相似：

你可以看到每個階段由步驟「單」欄所示。

但是，如果我們看看下面的表格中，我們可以看到任務的每級數量：

其中之一是隻使用2個任務，但其他使用220，這意味着數據被分割成220個分區，並且分區被並行處理，給定足夠的可用資源。

如果深入瞭解該階段，則可以再次看到它已使用了220個任務和所有任務的詳細信息。從磁盤

只有任務讀取數據顯示在圖中具有這些「多點」，以幫助您瞭解多少文件是如何被讀取。

SO--正如Rashid的回答所暗示的，檢查每個階段的任務數量。

Answer 2

這不是很明顯，所以我會做以下事情以解決問題。

1. 計算每個步驟的執行時間。
2. 如果您的表格是文本格式，則第一步可能會很慢，如果數據以實木複合地格式存儲在Hive中，那麼spark通常會更好。
3. 查看您的表是否由where子句中使用的列分區。
4. 如果將數據保存到Hbase的速度很慢，那麼您可能需要預先拆分hbase表，因爲默認情況下數據存儲在單個區域中。
5. 看看階段選項卡中的火花UI，看看有多少任務都開始爲每一個階段，也找數據地方一級描述here

希望，你將能夠零的問題。