Spark 工作調整提示

下列各節提供一些提示，協助您微調 Managed Service for Apache Spark 應用程式。

使用臨時叢集

使用 Managed Service for Apache Spark「臨時」叢集模式時，您會為每項工作建立專屬叢集，並在工作完成後刪除叢集。使用臨時模型時，您可以分開處理儲存空間和運算，將工作輸入和輸出資料儲存在 Cloud Storage 或 BigQuery 中，並僅將叢集用於運算和暫存資料。

永久叢集的缺點

使用臨時單一工作叢集，可避免下列與使用共用和長時間執行的「永久」叢集相關的陷阱和潛在問題：

  • 單點故障：共用叢集錯誤狀態可能會導致所有工作失敗，阻礙整個資料管道。調查及復原錯誤可能需要數小時。由於臨時叢集只會保留叢集內的暫時狀態，因此發生錯誤時，可以快速刪除並重新建立。
  • 難以維護及遷移 HDFS、MySQL 或本機檔案系統中的叢集狀態
  • 工作之間的資源爭用，對服務等級目標造成負面影響
  • 記憶體壓力導致服務 Daemon 無回應
  • 記錄和暫存檔累積過多，可能超出磁碟容量
  • 叢集可用區缺貨，因此無法向上擴充
  • 不支援過時的叢集映像檔版本。

臨時叢集的優點

臨時叢集有以下優點：

  • 使用不同的 Managed Service for Apache Spark VM 服務帳戶，為不同工作設定不同的 IAM 權限。
  • 針對每項工作最佳化叢集的硬體和軟體設定，並視需要變更叢集設定。
  • 升級新叢集的映像檔版本，取得最新的安全性修補程式、錯誤修正和最佳化功能。
  • 在獨立的單一工作叢集上，更快速地排解問題。
  • 只需支付臨時叢集執行時間的費用，不必支付共用叢集上工作之間的閒置時間費用，因此可節省成本。

使用 Spark SQL

Spark SQL DataFrame API 是 RDD API 的重大最佳化。如果您與使用 RDD 的程式碼互動，請考慮先將資料讀取為 DataFrame，再將 RDD 傳入程式碼。在 Java 或 Scala 程式碼中，建議使用 Spark SQL Dataset API 做為 RDD 和 DataFrame 的超集。

使用 Apache Spark 3

Managed Service for Apache Spark 2.0 會安裝 Spark 3，其中包含下列功能和效能改善項目：

  • GPU 支援
  • 可讀取二進位檔案
  • 提升效能
  • 動態分區剪枝
  • 自動調整查詢執行：即時最佳化 Spark 工作

使用動態分配

Apache Spark 包含動態分配功能，可調整叢集內 worker 的 Spark 執行器數量。這項功能可讓工作使用完整的 Managed Service for Apache Spark 叢集，即使叢集擴充也一樣。Managed Service for Apache Spark 預設會啟用這項功能 (spark.dynamicAllocation.enabled 設為 true)。詳情請參閱「Spark 動態分配」。

使用 Managed Service for Apache Spark 自動調度資源

Managed Service for Apache Spark 自動調度資源功能會動態新增及移除叢集中的 Managed Service for Apache Spark worker，確保 Spark 工作有足夠資源可快速完成。

最佳做法是將自動調度資源政策設為僅調度次要 worker。

使用 Managed Service for Apache Spark 彈性強化模式

如果叢集含有先占 VM 或自動調度資源政策，當 worker 在完成向縮減器提供重組資料前遭到先占或移除，叢集可能會收到 FetchFailed 例外狀況。這項例外狀況可能會導致工作重試，並延長工作完成時間。

建議：使用 Managed Service for Apache Spark 強化彈性模式，不會將中繼重組資料儲存在次要 worker 上，因此可以安全地搶占或縮減次要 worker。

設定分區和重組

Spark 會將資料儲存在叢集的暫時分區中。如果應用程式將 DataFrame 分組或彙整，則會根據分組和低階設定，將資料重組至新的分區。

資料分區會大幅影響應用程式效能：分區太少會限制工作平行處理和叢集資源使用率；分區太多則會因額外的分區處理和重組而降低工作速度。

設定分區

下列屬性會控管分區的數量和大小：

  • spark.sql.files.maxPartitionBytes：從 Cloud Storage 讀取資料時，分區的大小上限。預設值為 128 MB，對於處理資料量少於 100 TB 的大多數應用程式來說，這個值已足夠。

  • spark.sql.shuffle.partitions：執行重組後的分割區數量。如果是 2.2 以上版本的映像檔叢集，預設值為 1000。建議：將這個值設為叢集 vCPU 數量的 3 倍。

  • spark.default.parallelism：執行需要重組的 RDD 轉換 (例如 join、reduceByKey 和 parallelize) 後，傳回的分區數量。預設值為叢集中的 vCPU 總數。在 Spark 工作中使用 RDD 時，您可以將這個數字設為 vCPU 的 3 倍

限制檔案數量

Spark 讀取大量小型檔案時，效能會降低。儲存較大的檔案，例如 256 MB 至 512 MB 範圍內的檔案。同樣地，請限制輸出檔案數量 (如要強制重組，請參閱「避免不必要的重組」)。

設定自動調整查詢執行 (Spark 3)

自動調整查詢執行 (在 Managed Service for Apache Spark 映像檔 2.0 版中預設啟用) 可提升 Spark 工作效能，包括：

雖然預設設定適用於大多數用途，但將 spark.sql.adaptive.advisoryPartitionSizeInBytes 設為 spark.sql.files.maxPartitionBytes (預設為 128 MB) 可能會有所幫助。

避免不必要的重組

Spark 允許使用者透過 repartition 函式手動觸發重組，重新平衡資料。重組作業相當耗費資源，因此請謹慎使用重新隨機排序資料。只要適當設定分區設定，Spark 就能自動分區資料。

例外狀況：將資料欄分區資料寫入 Cloud Storage 時，請對特定資料欄重新分區，避免寫入許多小檔案，以加快寫入速度。

df.repartition("col_name").write().partitionBy("col_name").save("gs://...")

以 Parquet 或 Avro 格式儲存資料

Spark SQL 預設會讀取及寫入 Snappy 壓縮的 Parquet 檔案中的資料。Parquet 採用高效率的欄式檔案格式，可讓 Spark 只讀取執行應用程式所需的資料。處理大型資料集時，這是一項重要優勢。其他資料欄格式 (例如 Apache ORC) 也有不錯的效能。

對於非欄式資料，Apache Avro 提供有效率的二進位資料列檔案格式。雖然 Avro 的速度通常比 Parquet 慢，但效能優於 CSV 或 JSON 等文字格式。

最佳化磁碟大小

永久磁碟的處理量會隨著磁碟大小而改變，這可能會影響 Spark 工作的效能，因為工作會將中繼資料和重組資料寫入磁碟。使用標準永久磁碟時，每個 worker 的磁碟大小應至少為 1 TB (請參閱「依永久磁碟大小區分的效能」)。

如要在Cloud de Confiance 控制台監控 worker 磁碟處理量，請按照下列步驟操作：

  1. 在「Clusters」(叢集) 頁面中，按一下叢集名稱。
  2. 按一下「VM 執行個體」分頁標籤。
  3. 按一下任一工作站名稱。
  4. 按一下「監控」分頁標籤，然後捲動至「磁碟處理量」，查看 worker 處理量。

磁碟考量事項

臨時 Managed Service for Apache Spark 叢集無法使用永久儲存空間，但可使用本機 SSD。本機 SSD 會實際連接至叢集，且處理量高於永久磁碟 (請參閱效能表)。本機 SSD 的固定大小為 375 GB，但您可以新增多個 SSD 來提升效能。

叢集關閉後，本機 SSD 不會保留資料。如需永久儲存空間，可以使用 SSD 永久磁碟，其處理量比標準永久磁碟更高。如果分區大小小於 8 KB，SSD 永久磁碟也是不錯的選擇 (不過，請避免使用小型分區)。

將 GPU 連接至叢集

Spark 3 支援 GPU。搭配使用 GPU 和 CUDA-X 初始化動作，透過 NVIDIA cuDF for Apache Spark 程式庫加速 Spark 工作。GPU 驅動程式初始化動作，用於設定具備 GPU 的叢集。

常見的工作失敗原因和修正方式

以下各節說明工作失敗的常見原因和解決方案。

記憶體不足

範例：

  • 「Lost executor」
  • 「java.lang.OutOfMemoryError: GC overhead limit exceeded」
  • 「Container killed by YARN for exceeding memory limits」

可能的修正方式：

重組擷取失敗

範例：

  • 「FetchFailedException」(Spark 錯誤)
  • 「無法連線至…」(Spark 錯誤)
  • 「Failed to fetch」(MapReduce 錯誤)

通常是因為過早移除仍有重組資料要提供的 worker。

可能原因和修正方式：

  • 先占 worker VM 已回收，或非先占 worker VM 已由自動配置器移除。解決方案：使用強化版彈性模式，讓次要 worker 安全地先占或擴充。
  • 執行器或對應器因 OutOfMemory 錯誤而異常終止。解決方法：增加執行程式或對應程式的記憶體。
  • Spark shuffle 服務可能超載。解決方法：減少工作分區數量。

YARN 節點健康狀態不良

範例 (來自 YARN 記錄)： 

...reported UNHEALTHY with details: 1/1 local-dirs usable space is below
configured utilization percentage/no more usable space
[ /hadoop/yarn/nm-local-dir : used space above threshold of 90.0% ]

通常與重組資料的磁碟空間不足有關。查看記錄檔來診斷問題：

  • 在 Cloud de Confiance 控制台中開啟專案的「Clusters」(叢集) 頁面，然後按一下叢集名稱。
  • 按一下「查看記錄」。
  • 依hadoop-yarn-nodemanager篩選記錄。
  • 搜尋「UNHEALTHY」。

可能的修正方式：

  • 使用者快取會儲存在 yarn-site.xml file 中 yarn.nodemanager.local-dirs 屬性指定的目錄。這個檔案位於 /etc/hadoop/conf/yarn-site.xml。您可以查看 /hadoop/yarn/nm-local-dir 路徑中的可用空間，並刪除 /hadoop/yarn/nm-local-dir/usercache 使用者快取資料夾來釋出空間。
  • 如果記錄顯示「UNHEALTHY」狀態，請重新建立叢集並增加磁碟空間，這樣就能提高處理量上限。

使用者快取占用磁碟空間

在 Managed Service for Apache Spark 中，使用者快取會儲存在 yarn-site.xml (/etc/hadoop/conf/yarn-site.xml) 中 yarn.nodemanager.local-dirs 屬性指定的目錄中。在大多數情況下，位置是 /hadoop/yarn/nm-local-dir。

可能的修正方式：

  • 手動清理：檢查 /hadoop/yarn/nm-local-dir 路徑中的可用空間，然後刪除 /hadoop/yarn/nm-local-dir/usercache 使用者快取資料夾，釋出空間。

  • 長期解決方案：設定下列叢集屬性，管理清理程序：

    • yarn.nodemanager.localizer.cache.cleanup.interval-ms
    • yarn.nodemanager.localizer.cache.target-size-mb

    如果不想重新建立叢集，可以在所有節點的 yarn-site.xml 中更新這些屬性，然後在 worker 節點上執行下列指令，重新啟動 NodeManager：

    sudo systemctl restart hadoop-yarn-nodemanager.service

  • 調整叢集大小：如果記錄檔回報「UNHEALTHY」狀態，請重新建立叢集並增加磁碟空間，這樣就能提高處理量上限。

驅動程式記憶體不足，導致工作失敗

在叢集模式下執行工作時，如果工作記憶體大小超過 worker 節點記憶體大小，工作就會失敗。

驅動程式記錄範例： 

'Exception in thread "main" java.lang.IllegalArgumentException:
Required AM memory (32768+3276 MB) is above the max threshold (12288 MB) of this cluster!
Please check the values of 'yarn.scheduler.maximum -allocation-mb' and/or 'yarn.nodemanager.resource.memory-mb'.'

可能的修正方式：

  • spark:spark.driver.memory 必須小於 yarn:yarn.scheduler.maximum-allocation-mb。
  • 主要節點和 worker 節點使用相同的機型。

後續步驟