您好,登錄后才能下訂單哦!
大數(shù)據(jù)開發(fā)中Spark-拷問靈魂的問題有哪些,很多新手對此不是很清楚,為了幫助大家解決這個難題,下面小編將為大家詳細講解,有這方面需求的人可以來學習下,希望你能有所收獲。
①、把磁盤上的500G數(shù)據(jù)分割為100塊(chunks),每份5GB。(注意,要留一些系統(tǒng)空間?。?/pre>②、順序?qū)⒚糠?GB數(shù)據(jù)讀入內(nèi)存,使用quick sort算法排序。
③、把排序好的數(shù)據(jù)(也是5GB)存放回磁盤。
④、循環(huán)100次,現(xiàn)在,所有的100個塊都已經(jīng)各自排序了。(剩下的工作就是如何把它們合并排序?。?nbsp;
⑤、從100個塊中分別讀取5G/100=0.05 G入內(nèi)存(100input buffers)。
⑥、執(zhí)行100路合并,并將合并結(jié)果臨時存儲于5g基于內(nèi)存的輸出緩沖區(qū)中。當緩沖區(qū)寫滿5GB時,寫入硬盤上最終文件,并清空輸出緩沖區(qū);當100個輸入緩沖區(qū)中任何一個處理完畢時,寫入該緩沖區(qū)所對應的塊中的下一個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
上
其實現(xiàn)過程調(diào)用了 countByKey
countByKey(PairRDDFunctions.scala)
作用在 PairRDD 上
對 key 進行計數(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() //可以
其中join操作是考驗所有數(shù)據(jù)庫性能的一項重要指標,對于Spark來說,考驗join的性能就是Shuffle,Shuffle 需要經(jīng)過磁盤和網(wǎng)絡傳輸,Shuffle數(shù)據(jù)越少性能越好,有時候可以盡量避免程序進行Shuffle ,那么什么情況下有Shuffle ,什么情況下沒有Shuffle 呢
broadcast join 比較好理解,除了自己實現(xiàn)外,Spark SQL
已經(jīng)幫我們默認來實現(xiàn)了,其實就是小表分發(fā)到所有Executors
,控制參數(shù)是:spark.sql.autoBroadcastJoinThreshold
默認大小是10m, 即小于這個閾值即自動使用broadcast join
.
其實rdd方式和table類似,不同的是后者要寫入Bucket表,這里主要講rdd的方式,原理就是,當兩個rdd根據(jù)相同分區(qū)方式,預先做好分區(qū),分區(qū)結(jié)果是一致的,這樣就可以進行Bucket join, 另外這種join沒有預先的算子,需要在寫程序時候自己來開發(fā),對于表的這種join可以看一下 字節(jié)跳動在Spark SQL上的核心優(yōu)化實踐 ??梢钥聪孪旅娴睦?/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
, 關于spark的join原理可以查看:大數(shù)據(jù)開發(fā)-Spark Join原理詳解
有個算子例外,那就是sortByKey,其底層有個抽樣算法,水塘抽樣,最后需要根據(jù)抽樣的結(jié)果,進行RangePartition的,所以從job角度來說會看到兩個job,除了觸發(fā)action的本身算子之外,記住下面的
sortByKey → 水塘抽樣→ collect
我們都知道,廣播變量是把數(shù)據(jù)放到每個excutor上,也都知道廣播變量的數(shù)據(jù)一定是從driver開始出去的,什么意思呢,如果廣播表放在hive表中,那么它的存儲就是在各個block塊上,也對應多個excutor (不一樣的叫法),首先將數(shù)據(jù)拉到driver上,然后再進行廣播,廣播時候不是全部廣播,是根據(jù)excutor預先用到數(shù)據(jù)的,首先拿數(shù)據(jù),然后通過bt協(xié)議進行傳輸,什么是bt協(xié)議呢,就是數(shù)據(jù)在分布式點對點網(wǎng)絡上,根據(jù)網(wǎng)絡距離來去拉對應的數(shù)據(jù),下載者也是上傳者,這樣就不同每個task (excutor)都從driver上來拉數(shù)據(jù),這樣就減少了壓力,另外在spark1.幾的時候還是task級別,現(xiàn)在是共同的一個鎖,整個excutor上的task共享這份數(shù)據(jù)。
看完上述內(nèi)容是否對您有幫助呢?如果還想對相關知識有進一步的了解或閱讀更多相關文章,請關注億速云行業(yè)資訊頻道,感謝您對億速云的支持。
免責聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進行舉報,并提供相關證據(jù),一經(jīng)查實,將立刻刪除涉嫌侵權(quán)內(nèi)容。