我想重點介紹資料處理不同層上的Spark分區的主題。
在實體級别上，，分為三個階段。 它們是input，shuffle和output。
例如，在input和output處，您可以控制分區的大小，但是在output處，我也可以通過合并或重新分區來控制檔案數或任務數。 使用shuffle，您可以控制将在網絡中移動的資料數量。
在輸入階段進行分區
首先，根據輸入資料集的大小确定分區數。
Spark可以很好地處理輸入。在如何将資料放置在HDFS或Cassandra等資料存儲中以及Spark在讀取時如何拆分資料之間存在明确的比對。
讓我們想象一下，輸入資料集的大小約為HDFS上未壓縮文本檔案的30 GB（約30000 MB），該檔案正在10個節點上分發。
當Spark從HDFS讀取檔案時，它将為單個輸入拆分建立一個分區。輸入拆分由用于讀取此檔案的Hadoop設定。如果您在HDFS上存儲了30GB的未壓縮文本檔案，則使用預設的HDFS塊大小設定（128MB）和預設（128MB），它将存儲在240個塊中，這意味着您從該檔案讀取的資料幀将具有240個分區
.InputFormatspark.files.maxPartitionBytes
,這等于Spark預設的分區數
spark.default.parallelism
如果您的資料無法拆分，Spark将使用預設數量的分區。作業啟動時，分區數等于所有執行程式節點上的核心總數。
shuffle分區
任何Spark管道中最痛苦的地方是wide transformation，需要來自其他分區的資訊并觸發shuffle。不幸的是，您無論如何都無法擺脫這種轉變，但是您可以減少shuffle對性能的影響。
shuffle分區是在資料混洗中用于wide transformation的分區。但是，對于wide transformation，shuffle分區的數量設定為200。無論您的資料是大還是小，或者如果叢集配置中有20個執行程式，它仍然是200，都是如此。是的，是的，這就是我們看到的在分區部分，那是神秘的數字。
是以，控制由shuffle導緻的并行性的參數是一個稱為參數的參數。預設值為200的原因是來自真實經驗，這是一個非常好的預設值。但實際上，該值通常總是不好的。
spark.sql.shuffle.partitions
當處理少量資料時，通常應減少shuffle分區的數量，否則最終将導緻許多分區，每個分區中的條目數量很少，這會導緻所有執行程式的使用率不足，并增加了資料處理時間。通過網絡從執行者轉移到執行者。
另一方面，當您有太多的資料和分區太少時，這會導緻執行器中要處理的任務減少，但會增加每個執行器的負擔，并經常導緻記憶體錯誤。 另外，如果将分區的大小增加到大于執行程式中可用記憶體的大小，則會導緻磁盤溢出。 溢出是您可能能夠做的最慢的事情。 從本質上講，在磁盤溢出期間，如果Spark操作無法将其部分RAM放入磁盤中，則Spark作業可以在任何大小的資料上正常運作。 盡管它不會破壞您的管道，但由于磁盤I / O的額外開銷和垃圾收集的增加，它也使效率極低。
是以，當使用
Spark.spark.sql.shuffle.partitions
時，它是配置最頻繁的參數之一
output分區
儲存在适當的選擇條件的資料可以顯著加快在未來處理流水線所需的資料的讀出和檢索。
首先，在某些情況下，可以在發現DataSource的分區之後使用分區修剪，這會限制Spark在查詢時讀取的檔案和分區的數量。在某些情況下（例如AWS s3），它甚至避免了不必要的分區發現。 Spark 3.0中的動态分區修剪的概念也很有價值。但是有時候，這些優化可能會使情況變得更糟，例如對檔案系統進行遞歸掃描以擷取中繼資料以了解初始查詢的分區可能需要很長時間（如果分區數量很多）。同樣，所有表中繼資料必須具體化到驅動程式程序的記憶體中，并且可能顯着增加其記憶體負擔。
其次，将
DataFrame
儲存到磁盤時，請特别注意分區大小。在寫入期間，Spark将為每個任務生成一個檔案（即每個分區一個檔案），并且在讀取時将讀取任務中的至少一個檔案。這裡的問題是，如果儲存
DataFrame
的叢集設定具有大的的記憶體，是以可以毫無問題地處理大型分區，而較小的叢集可能會在讀取儲存的
DataFrame
時遇到問題。
例如，您有一個大型叢集和一個較小的(更具成本效益的服務叢集)。在這種情況下，解決方案将是在寫入之前将
DataFrame
重新分區為更多分區，以使下一個叢集不會阻塞。
結論
如果你想增加分區數,用
repartition()
.
如果你想減少分區數,用
coalesce()
,可以盡量避免shuffle
引用
https://luminousmen.com/post/spark-partitions