1.spark中窄依賴的時候不需要shuffle，只有寬依賴的時候需要shuffle，mapreduce中map到reduce必須經過shuffle
2.spark中的shuffle fetch的時候進行merge操作利用aggregator來進行，實際上是個hashmap，放在內存中
1 // Map: "cat" -> c, cat 2 val rdd1 = rdd.Map(x => (x.charAt(0), x)) 3 // groupby same key and count 4 val rdd2 = rdd1.groupBy(x => x._1). 5 Map(x => (x._1, x._2.toList.length))
第一個 Map 操作將 RDD 裏的各個元素進行映射, RDD 的各個數據元素之間不存在依賴,可以在集羣的各個內存中獨立計算,也就是並行化,第二個 groupby 之後的 Map 操作,爲了計算相同 key 下的元素個數,需要把相同 key 的元素聚集到同一個 partition 下,所以造成了數據在內存中的重新分佈,即 shuffle 操作.shuffle 操作是 spark 中最耗時的操作,應儘量避免不必要的 shuffle.
寬依賴主要有兩個過程: shuffle write 和 shuffle fetch. 類似 Hadoop 的 Map 和 Reduce 階段.shuffle write 將 ShuffleMapTask 任務產生的中間結果緩存到內存中, shuffle fetch 獲得 ShuffleMapTask 緩存的中間結果進行 ShuffleReduceTask 計算,這個過程容易造成OutOfMemory.
shuffle 過程內存分配使用 ShuffleMemoryManager 類管理,會針對每個 Task 分配內存,Task 任務完成後通過 Executor 釋放空間.這裏可以把 Task 理解成不同 key 的數據對應一個 Task. 早期的內存分配機制使用公平分配,即不同 Task 分配的內存是一樣的,但是這樣容易造成內存需求過多的 Task 的 OutOfMemory, 從而造成多餘的 磁盤 IO 過程,影響整體的效率.(例:某一個 key 下的數據明顯偏多,但因爲大家內存都一樣,這一個 key 的數據就容易 OutOfMemory).1.5版以後 Task 共用一個內存池,內存池的大小默認爲 JVM 最大運行時內存容量的16%,分配機制如下:假如有 N 個 Task,ShuffleMemoryManager 保證每個 Task 溢出之前至少可以申請到1/2N 內存,且至多申請到1/N,N 爲當前活動的 shuffle Task 數,因爲N 是一直變化的,所以 manager 會一直追蹤 Task 數的變化,重新計算隊列中的1/N 和1/2N.但是這樣仍然容易造成內存需要多的 Task 任務溢出,所以最近有很多相關的研究是針對 shuffle 過程內存優化的.
早期的shuffle write有兩個比較大的問題：
- Map的輸出必須先全部存儲到內存中，然後寫入磁盤。這對內存是一個非常大的開銷，當內存不足以存儲所有的Map output時就會出現OOM。
- 每一個Mapper都會產生Reducer number個shuffle文件，如果Mapper個數是1k，Reducer個數也是1k，那麼就會產生1M個shuffle文件，這對於文件系統是一個非常大的負擔。同時在shuffle數據量不大而shuffle文件又非常多的情況下，隨機寫也會嚴重降低IO的性能。
Spark 0.8顯著減少了shuffle的內存壓力，現在Map output不需要先全部存儲在內存中，再flush到硬盤，而是record-by-record寫入到磁盤中。
爲了解決shuffle文件過多的情況，Spark 0.8.1引入了新的shuffle consolidation，以期顯著減少shuffle文件的數量。
假定該job有4個Mapper和4個Reducer，有2個core，也就是能並行運行兩個task。我們可以算出Spark的shuffle write共需要16個bucket，也就有了16個write handler。在之前的Spark版本中，每一個bucket對應的是一個文件，因此在這裏會產生16個shuffle文件。
而在shuffle consolidation中每一個bucket並非對應一個文件，而是對應文件中的一個segment，同時shuffle consolidation所產生的shuffle文件數量與Spark core的個數也有關係。在上面的圖例中，job的4個Mapper分爲兩批運行，在第一批2個Mapper運行時會申請8個bucket，產生8個shuffle文件；而在第二批Mapper運行時，申請的8個bucket並不會再產生8個新的文件，而是追加寫到之前的8個文件後面，這樣一共就只有8個shuffle文件，而在文件內部這有16個不同的segment。因此從理論上講shuffle consolidation所產生的shuffle文件數量爲C×R，其中C是Spark集羣的core number，R是Reducer的個數。
需要注意的是當 M=C時shuffle consolidation所產生的文件數和之前的實現是一樣的。
Shuffle consolidation顯著減少了shuffle文件的數量，解決了之前版本一個比較嚴重的問題，但是writer handler的buffer開銷過大依然沒有減少，若要減少writer handler的buffer開銷，我們只能減少Reducer的數量，但是這又會引入新的問題，下文將會有詳細介紹。
Shuffle Fetch and Aggregator
Shuffle write寫出去的數據要被Reducer使用，就需要shuffle fetcher將所需的數據fetch過來，這裏的fetch包括本地和遠端，因爲shuffle數據有可能一部分是存儲在本地的。Spark對shuffle fetcher實現了兩套不同的框架：NIO通過socket連接去fetch數據；OIO通過netty server去fetch數據。分別對應的類是BasicBlockFetcherIterator和NettyBlockFetcherIterator。
在Spark 0.7和更早的版本中，只支持BasicBlockFetcherIterator，而BasicBlockFetcherIterator在shuffle數據量比較大的情況下performance始終不是很好，無法充分利用網絡帶寬，爲了解決這個問題，添加了新的shuffle fetcher來試圖取得更好的性能。對於早期shuffle性能的評測可以參看Spark usergroup。當然現在BasicBlockFetcherIterator的性能也已經好了很多，使用的時候可以對這兩種實現都進行測試比較。
接下來說一下aggregator。我們都知道在Hadoop MapReduce的shuffle過程中，shuffle fetch過來的數據會進行merge sort，使得相同key下的不同value按序歸併到一起供Reducer使用，這個過程可以參看下圖：
所有的merge sort都是在磁盤上進行的，有效地控制了內存的使用，但是代價是更多的磁盤IO。
那麼Spark是否也有merge sort呢，還是以別的方式實現，下面我們就細細說明。
首先雖然Spark屬於MapReduce體系，但是對傳統的MapReduce算法進行了一定的改變。Spark假定在大多數用戶的case中，shuffle數據的sort不是必須的，比如word count，強制地進行排序只會使性能變差，因此Spark並不在Reducer端做merge sort。既然沒有merge sort那Spark是如何進行reduce的呢？這就要說到aggregator了。
aggregator本質上是一個hashmap，它是以map output的key爲key，以任意所要combine的類型爲value的hashmap。當我們在做word count reduce計算count值的時候，它會將shuffle fetch到的每一個key-value pair更新或是插入到hashmap中(若在hashmap中沒有查找到，則插入其中；若查找到則更新value值)。這樣就不需要預先把所有的key-value進行merge sort，而是來一個處理一個，省下了外部排序這一步驟。但同時需要注意的是reducer的內存必須足以存放這個partition的所有key和count值，因此對內存有一定的要求。
在上面word count的例子中，因爲value會不斷地更新，而不需要將其全部記錄在內存中，因此內存的使用還是比較少的。考慮一下如果是group by key這樣的操作，Reducer需要得到key對應的所有value。在Hadoop MapReduce中，由於有了merge sort，因此給予Reducer的數據已經是group by key了，而Spark沒有這一步，因此需要將key和對應的value全部存放在hashmap中，並將value合併成一個array。可以想象爲了能夠存放所有數據，用戶必須確保每一個partition足夠小到內存能夠容納，這對於內存是一個非常嚴峻的考驗。因此Spark文檔中建議用戶涉及到這類操作的時候儘量增加partition，也就是增加Mapper和Reducer的數量。
增加Mapper和Reducer的數量固然可以減小partition的大小，使得內存可以容納這個partition。但是我們在shuffle write中提到，bucket和對應於bucket的write handler是由Mapper和Reducer的數量決定的，task越多，bucket就會增加的更多，由此帶來write handler所需的buffer也會更多。在一方面我們爲了減少內存的使用採取了增加task數量的策略，另一方面task數量增多又會帶來buffer開銷更大的問題，因此陷入了內存使用的兩難境地。
爲了減少內存的使用，只能將aggregator的操作從內存移到磁盤上進行，Spark社區也意識到了Spark在處理數據規模遠遠大於內存大小時所帶來的問題。因此PR303提供了外部排序的實現方案，相信在Spark 0.9 release的時候，這個patch應該能merge進去，到時候內存的使用量可以顯著地減少。