本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

優化洗牌

某些作業 (例如join()和groupByKey()) 需要 Spark 才能執行隨機播放。隨機播放是 Spark 重新分配資料的機制，以便在RDD分割區之間以不同的方式分組資料。洗牌可協助修正效能瓶頸。但是，由於洗牌通常涉及在 Spark 執行程序之間複製數據，因此隨機播放是一個複雜且昂貴的操作。例如，隨機播放會產生下列成本：

隨機播放是 Spark 應用程式效能下降的最重要因素之一。在存儲中間數據時，它可能會耗盡執行程序的本地磁盤上的空間，這會導致 Spark 作業失敗。

您可以在 CloudWatch 指標和 Spark UI 中評估隨機播放效能。

CloudWatch 度量

如果隨機播放字節寫入值與隨機字節讀取相比較高，您的 Spark 作業可能會使用隨機播放操作，例如或。join() groupByKey()

Spark UI

在 Spark UI 的「舞台」索引標籤上，您可以檢查「隨機播放讀取大小/記錄」值。您也可以在「執行者」選項卡上看到它。

在下面的屏幕截圖中，每個執行者與隨機播放過程交換大約 18.6GB/4020000 條記錄，總隨機播放讀取大小約為 75 GB）。

隨機溢出 (磁碟) 欄會顯示大量的資料溢滿記憶體至磁碟，這可能會造成磁碟已滿或效能問題。

如果您觀察到這些症狀，而且階段與您的績效目標相比需要太長的時間，或者失敗Out Of Memory或No space left on device錯誤，請考慮下列解決方案。

優化加入

連接資料表的join()作業是最常用的隨機播放作業，但通常是效能瓶頸。由於 join 是一項昂貴的操作，因此我們建議您不要使用它，除非它對您的業務需求至關重要。請提出以下問題，仔細檢查您是否有效地使用資料管線：

確認加入作業對您的業務需求至關重要之後，請參閱下列選項，以符合您需求的方式最佳化您的聯結。

加入前使用下推

在執行聯結 DataFrame 之前，請先篩選出中不必要的列和欄。這具有以下優點：

# Default
df_joined  = df1.join(df2, ["product_id"])

# Use Pushdown
df1_select = df1.select("product_id","product_title","star_rating").filter(col("star_rating")>=4.0)
df2_select = df2.select("product_id","category_id")
df_joined  = df1_select.join(df2_select, ["product_id"])

使用 DataFrame 加入

嘗試使用 Spark 高級別，API例如 Spark SQL DataFrame，和數據集，而不是RDDAPI或 DynamicFrame 加入。您可以 DataFrame 使 DynamicFrame 用方法調用轉換為dyf.toDF()。如 Apache Spark 中「關鍵主題」一節所述，這些聯結作業會在內部利用 Catalyst 最佳化工具的查詢最佳化功能。

隨機播放和廣播雜湊連接和提示

星火支持兩種類型的連接：隨機連接和廣播哈希加入。廣播哈希連接不需要混洗，並且與隨機連接相比，它可能需要更少的處理。但是，僅在將小桌子連接到大桌子時才適用。當加入可容納單個 Spark 執行程序內存的表時，請考慮使用廣播哈希聯接。

下圖顯示了廣播哈希聯接和隨機連接的高級結構和步驟。

每個聯結的詳細資訊如下：

在 AWS Glue 3.0 及更新版本中，您可以啟用「調適性查詢執行」和其他參數，以自動利用廣播雜湊聯結。當任一聯結端的執行階段統計資料小於調適性廣播雜湊聯結臨界值時，調適性查詢執行會將排序合併聯結轉換為廣播雜湊聯結。

在 AWS Glue 3.0 中，您可以通過設置啟用自適應查詢執行spark.sql.adaptive.enabled=true。在 AWS Glue 4.0 中預設會啟用調適性查詢執行。

您可以設置與洗牌和廣播哈希連接相關的其他參數：

如需相關參數的詳細資訊，請參閱將排序合併聯結轉換為廣播聯結。

在 AWS Glue 3.0 和更高版本中，您可以使用其他連接提示進行隨機播放來調整您的行為。

-- Join Hints for shuffle sort merge join
SELECT /*+ SHUFFLE_MERGE(t1) /  FROM t1 INNER JOIN t2 ON t1.key = t2.key;
SELECT /*+ MERGEJOIN(t2) /  FROM t1 INNER JOIN t2 ON t1.key = t2.key;
SELECT /*+ MERGE(t1) /  FROM t1 INNER JOIN t2 ON t1.key = t2.key;

-- Join Hints for shuffle hash join
SELECT /*+ SHUFFLE_HASH(t1) /  FROM t1 INNER JOIN t2 ON t1.key = t2.key;

-- Join Hints for shuffle-and-replicate nested loop join
SELECT /*+ SHUFFLE_REPLICATE_NL(t1) /  FROM t1 INNER JOIN t2 ON t1.key = t2.key;

使用分段設定

排序合併連接需要兩個階段，隨機排序和排序，然後合併。這兩個階段可能會超載 Spark 執行程序，並導致OOM和性能問題，當一些執行者正在合併和其他執行程序同時排序。在這種情況下，可能可以通過使用分段來有效地加入。Bucketing 將在聯接鍵上預先洗牌和預先排序您的輸入，然後將排序的數據寫入中間表。透過預先定義已排序的中繼資料表，在連接大型資料表時，可以減少隨機排序和排序步驟的成本。

分組表格對下列項目非常有用：

您可以通過使用下面的代碼創建一個分組的表。

df.write.bucketBy(50, "account_id").sortBy("age").saveAsTable("bucketed_table")

在聯接之前 DataFrames 對聯接鍵進行重新分區

若要 DataFrames 在連接之前重新分割兩個連接鍵，請使用下列陳述式。

df1_repartitioned = df1.repartition(N,"join_key")
df2_repartitioned = df2.repartition(N,"join_key")
df_joined  = df1_repartitioned.join(df2_repartitioned,"product_id")

這將RDDs在啟動聯接之前對連接鍵進行兩個分區（仍然是分開的）。如果兩者在具有相同分區代碼的同一個密鑰上進行分區，則您計劃聯接在一起的RDD記錄將RDDs很有可能在混洗聯接之前共同位於同一個 Worker 上。這可能會透過減少聯結期間的網路活動和資料偏差來改善效能。

克服資料偏斜

資料偏斜是 Spark 工作瓶頸的最常見原因之一。當數據不是跨RDD分區均勻分佈時，就會發生這種情況。這會導致該分區的任務比其他分區花費更長的時間，從而延遲應用程序的整體處理時間。

若要識別資料偏差，請在 Spark UI 中評估下列量度：

如果您看到資料偏斜，請嘗試下列動作：

使用快取

使用重複性時 DataFrames，請使用或df.persist()將計算結果緩存在每個 Spark 執行程序的內存和磁盤上，避免額外的隨機播放df.cache()或計算。Spark 還支持RDDs在磁盤上保存或跨多個節點（存儲級別）進行複製。

例如，您可以 DataFrames 通過添加來保留df.persist(). 當不再需要緩存時，您可以使unpersist用丟棄緩存的數據。

df = spark.read.parquet("s3://<Bucket>/parquet/product_category=Books/")
df_high_rate = df.filter(col("star_rating")>=4.0)
df_high_rate.persist()

df_joined1 = df_high_rate.join(<Table1>, ["key"])
df_joined2 = df_high_rate.join(<Table2>, ["key"])
df_joined3 = df_high_rate.join(<Table3>, ["key"])
... 
df_high_rate.unpersist()

移除不需要的火花動作

避免執行不必要的動作count，例如show、或collect。正如在 Apache 星火一節中的關鍵主題討論的那樣，星火是懶惰的。每次對其執行動作時，都RDD可能會重新計算每個轉換。當您使用許多 Spark 動作時，會呼叫每個動作的多個來源存取、工作計算和隨機播放執行。

如果您不需要collect()或在商業環境中執行其他動作，請考慮將其移除。