關(guān)于Spark拷問的5個問題分別是什么

1.Spark計算依賴內(nèi)存，如果目前只有10g內(nèi)存，但是需要將500G的文件排序并輸出，需要如何操作？

①、把磁盤上的500G數(shù)據(jù)分割為100塊（chunks），每份5GB。（注意，要留一些系統(tǒng)空間！）

②、順序?qū)⒚糠?GB數(shù)據(jù)讀入內(nèi)存，使用quick sort算法排序。

③、把排序好的數(shù)據(jù)（也是5GB）存放回磁盤。

④、循環(huán)100次，現(xiàn)在，所有的100個塊都已經(jīng)各自排序了。（剩下的工作就是如何把它們合并排序！）

⑤、從100個塊中分別讀取5G/100=0.05 G入內(nèi)存（100input buffers）。

⑥、執(zhí)行100路合并，并將合并結(jié)果臨時存儲于5g基于內(nèi)存的輸出緩沖區(qū)中。當(dāng)緩沖區(qū)寫滿5GB時，寫入硬盤上最終文件，并清空輸出緩沖區(qū)；當(dāng)100個輸入緩沖區(qū)中任何一個處理完畢時，寫入該緩沖區(qū)所對應(yīng)的塊中的下一個0.05 GB，直到全部處理完成。

2.countByValue和countByKey的區(qū)別

首先從源碼角度來看：

// PairRDDFunctions.scala
def countByKey(): Map[K, Long] = self.withScope {
  self.mapValues(_ => 1L).reduceByKey(_ + _).collect().toMap
}

// RDD.scala
def countByValue()(implicit ord: Ordering[T] = null): Map[T, Long] = withScope {
  map(value => (value, null)).countByKey()
} 

countByValue（RDD.scala）

• 作用在普通的RDD上

• 其實(shí)現(xiàn)過程調(diào)用了 countByKey

countByKey（PairRDDFunctions.scala）

• 作用在 PairRDD 上

• 對 key 進(jìn)行計數(shù)

• 數(shù)據(jù)要收到Driver端，結(jié)果集大時，不適用

問題：

• countByKey可以作用在 普通的RDD上嗎

• countByValue可以作用在 PairRDD上嗎

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() //可以

3.兩個rdd join 什么時候有shuffle什么時候沒有shuffle

其中join操作是考驗(yàn)所有數(shù)據(jù)庫性能的一項(xiàng)重要指標(biāo)，對于Spark來說，考驗(yàn)join的性能就是Shuffle,Shuffle 需要經(jīng)過磁盤和網(wǎng)絡(luò)傳輸，Shuffle數(shù)據(jù)越少性能越好，有時候可以盡量避免程序進(jìn)行Shuffle ,那么什么情況下有Shuffle ，什么情況下沒有Shuffle 呢

3.1 Broadcast join

broadcast join 比較好理解，除了自己實(shí)現(xiàn)外，Spark SQL 已經(jīng)幫我們默認(rèn)來實(shí)現(xiàn)了，其實(shí)就是小表分發(fā)到所有Executors，控制參數(shù)是：spark.sql.autoBroadcastJoinThreshold 默認(rèn)大小是10m, 即小于這個閾值即自動使用broadcast join.

3.2 Bucket join

其實(shí)rdd方式和table類似，不同的是后者要寫入Bucket表，這里主要講rdd的方式，原理就是，當(dāng)兩個rdd根據(jù)相同分區(qū)方式，預(yù)先做好分區(qū)，分區(qū)結(jié)果是一致的，這樣就可以進(jìn)行Bucket join, 另外這種join沒有預(yù)先的算子，需要在寫程序時候自己來開發(fā)，對于表的這種join可以看一下 字節(jié)跳動在Spark SQL上的核心優(yōu)化實(shí)踐 ?？梢钥聪孪旅娴睦?/p>

rdd1、rdd2都是Pair RDD

rdd1、rdd2的數(shù)據(jù)完全相同

一定有shuffle

rdd1 => 5個分區(qū)

rdd2 => 6個分區(qū)

rdd1 => 5個分區(qū) => (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個分區(qū) => (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個分區(qū) => （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個分區(qū) => （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的算子，如果數(shù)據(jù)提前做好了分區(qū)（partitionBy），很多情況下沒有Shuffle.

除上面兩種方式外，一般就是有Shuffle的join, 關(guān)于spark的join原理可以查看：大數(shù)據(jù)開發(fā)-Spark Join原理詳解

4..transform 是不是一定不觸發(fā)action

有個算子例外，那就是sortByKey,其底層有個抽樣算法，水塘抽樣，最后需要根據(jù)抽樣的結(jié)果，進(jìn)行RangePartition的,所以從job角度來說會看到兩個job，除了觸發(fā)action的本身算子之外，記住下面的

sortByKey  → 水塘抽樣→ collect

5.廣播變量是怎么設(shè)計的

我們都知道，廣播變量是把數(shù)據(jù)放到每個excutor上，也都知道廣播變量的數(shù)據(jù)一定是從driver開始出去的，什么意思呢，如果廣播表放在hive表中，那么它的存儲就是在各個block塊上，也對應(yīng)多個excutor (不一樣的叫法)，首先將數(shù)據(jù)拉到driver上，然后再進(jìn)行廣播，廣播時候不是全部廣播，是根據(jù)excutor預(yù)先用到數(shù)據(jù)的，首先拿數(shù)據(jù)，然后通過bt協(xié)議進(jìn)行傳輸，什么是bt協(xié)議呢，就是數(shù)據(jù)在分布式點(diǎn)對點(diǎn)網(wǎng)絡(luò)上，根據(jù)網(wǎng)絡(luò)距離來去拉對應(yīng)的數(shù)據(jù)，下載者也是上傳者，這樣就不同每個task （excutor）都從driver上來拉數(shù)據(jù)，這樣就減少了壓力，另外在spark1.幾的時候還是task級別，現(xiàn)在是共同的一個鎖，整個excutor上的task共享這份數(shù)據(jù)。