大資料開發 Spark 拷問靈魂的5個問題

①、把磁碟上的500g資料分割為100塊（chunks），每份5gb。（注意，要留一些系統空間！）

②、順序將每份5gb資料讀入記憶體，使用quick sort演算法排序。

③、把排序好的資料（也是5gb）存放回磁碟。

④、迴圈100次，現在，所有的100個塊都已經各自排序了。（剩下的工作就是如何把它們合併排序！）

⑤、從100個塊中分別讀取5g/100=0.05 g入記憶體（100input buffers）。

⑥、執行100路合併，並將合併結果臨時儲存於5g基於記憶體的輸出緩衝區中。當緩衝區寫滿5gb時，寫入硬碟上最終檔案，並清空輸出緩衝區；當100個輸入緩衝區中任何乙個處理完畢時，寫入該緩衝區所對應的塊中的下乙個0.05 gb，直到全部處理完成。

首先從原始碼角度來看：

// pairrddfunctions.scala
def countbykey(): map[k, long] = self.withscope 
// rdd.scala
def countbyvalue()(implicit ord: ordering[t] = null): map[t, long] = withscope

countbyvalue（rdd.scala）

countbykey（pairrddfunctions.scala）

問題：

val rdd1: rdd[int] = sc.makerdd(1 to 10)
val rdd2: rdd[(int, int)] = sc.makerdd((1 to 10).tolist.zipwithindex)
val result1 = rdd1.countbyvalue() //可以
val result2 = rdd1.countbykey() //語法錯誤
val result3 = rdd2.countbyvalue() //可以
val result4 = rdd2.countbykey() //可以

其中join操作是考驗所有資料庫效能的一項重要指標，對於spark來說，考驗join的效能就是shuffle,shuffle 需要經過磁碟和網路傳輸，shuffle資料越少效能越好，有時候可以盡量避免程式進行shuffle ,那麼什麼情況下有shuffle ，什麼情況下沒有shuffle 呢

broadcast join 比較好理解，除了自己實現外，spark sql已經幫我們預設來實現了，其實就是小表分發到所有executors，控制引數是：spark.sql.autobroadcastjointhreshold預設大小是10m, 即小於這個閾值即自動使用broadcast join.

其實rdd方式和table類似，不同的是後者要寫入bucket表，這裡主要講rdd的方式，原理就是，當兩個rdd根據相同分割槽方式，預先做好分割槽，分割槽結果是一致的，這樣就可以進行bucket join, 另外這種join沒有預先的運算元，需要在寫程式時候自己來開發，對於表的這種join可以看一下位元組跳動在spark sql上的核心優化實踐。可以看下下面的例子

rdd1、rdd2都是pair rdd

rdd1、rdd2的資料完全相同

一定有shuffle

rdd1 => 5個分割槽

rdd2 => 6個分割槽

rdd1 => 5個分割槽 => (1, 0), (2,0), || (1, 0), (2,0), || (1, 0), (2,0), || (1, 0), (2,0),(1, 0), || (2,0),(1, 0), (2,0)

rdd2 => 5個分割槽 => (1, 0), (2,0), || (1, 0), (2,0), || (1, 0), (2,0), || (1, 0), (2,0),(1, 0), || (2,0),(1, 0), (2,0)

一定沒有shuffle

rdd1 => 5個分割槽 => （1,0), （1,0), （1,0), （1,0), （1,0), || (2,0), (2,0), (2,0), (2,0), (2,0), (2,0), (2,0) || 空 || 空 || 空

rdd2 => 5個分割槽 => （1,0), （1,0), （1,0), （1,0), （1,0), || (2,0), (2,0), (2,0), (2,0), (2,0), (2,0), (2,0) || 空 || 空 || 空

這樣所有shuffle的運算元，如果資料提前做好了分割槽（partitionby），很多情況下沒有shuffle.

除上面兩種方式外，一般就是有shuffle的join, 關於spark的join原理可以檢視：大資料開發-spark join原理詳解

有個運算元例外，那就是sortbykey,其底層有個抽樣演算法，水塘抽樣，最後需要根據抽樣的結果，進行rangepartition的,所以從job角度來說會看到兩個job，除了觸發action的本身運算元之外，記住下面的

sortbykey → 水塘抽樣→ collect

大資料開發 Spark 拷問靈魂的5個問題

靈魂拷問如何提高系統的效能？

靈魂拷問說說你對前端的理解

大資料開發 Spark 閉包的理解

大資料開發 Spark 拷問靈魂的5個問題

靈魂拷問 如何提高系統的效能？

靈魂拷問 說說你對前端的理解

大資料開發 Spark 閉包的理解

相關推薦

靈魂拷問如何提高系統的效能？

靈魂拷問說說你對前端的理解