當前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

【 RDD持久化】

發布時間：2024/1/18 编程问答 35 豆豆

生活随笔收集整理的這篇文章主要介紹了【 RDD持久化】小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

文章目錄

RDD持久化
- 操作：
- 總結：
- cache vs persist 區別：
- 存儲級別選擇：
- 移除rdd持久化：
- 2.血緣關系
- 依賴關系分類：
- 4.shuffle 算子：
- 案例分析：

spark_05_RDD持久化

RDD持久化

rdda => rddb =>rddc action rdda => rddb =>rdde action

rddb 持久化操作 =》調優的

操作：

1.persist() or cache() methods 2.觸發action之后會對rdd數據進行持久化的

總結：

1.cache() 不是action算子是lazy 是懶加載的

rdda => action job
rdda => cache => action job => rdd持久化生效
rdda => action job rdda的數據從 rdd持久化的地方加載數據

rdda => rddb => rddc
rdda => rddb => rdde
rdda => rddb => rddf

rddb.cache 之后 rddb之后數據就不用從頭開機計算提升計算效率

補充：
對rdd做持久化就是對rdd里面的分區做持久化

好處：
1.much faster 【計算效率】
2. reuse 復用

cache vs persist 區別：

1.cache底層就是調用 persist算子 2.spark-core 持久化默認存儲級別：StorageLevel.MEMORY_ONLY

StorageLevel：
private var _useDisk: Boolean,
private var _useMemory: Boolean,
private var _useOffHeap: Boolean,
private var _deserialized: Boolean,
private var _replication: Int = 1

val NONE = new StorageLevel(false, false, false, false)
val DISK_ONLY = new StorageLevel(true, false, false, false)
val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)
val DISK_ONLY_3 = new StorageLevel(true, false, false, false, 3)
val MEMORY_ONLY = new StorageLevel(false, true, false, true)
val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)
val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)
val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)
val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)
val OFF_HEAP = new StorageLevel(true, true, true, false, 1)

存儲級別選擇：

1.MEMORY_ONLY 首選 2.MEMORY_ONLY_SER 次選 1.Java serialization: By default,2.Kryo： 1.注冊 class

移除rdd持久化：

1.lru 2.手動： RDD.unpersist(true) 立即執行的 eager

2.血緣關系

lineage:rdda => rddb => rddc

一個rdd是如何從父rdd計算得來的

textFile(path) => map => fiter => … => collect
每一個轉換都會形成一個rdd

好處：
容錯(性能 + 容錯)
容錯：
1.假如說RDDB 分區 6 8 元素在計算的時候掛了

一個鏈路 200個轉換算到第199個轉換數據壞了，如果從頭計算也是挺麻煩的一件事情：
1.spark-core 提供cache
而且持久化的數據集也支持容錯

3.依賴關系：
rdda => rddb
不同的依賴會導致生成rdd分區數發生變化的

依賴關系分類：

1.寬依賴：1.一個父rdd的parition會被子rdd的parition使用多次 2.會產生shuffle 會有新的stage產生 2.窄依賴：1.一個父rdd的parition至多被子rdd的partition使用一次2.不會產生shuffle，都是在一個stage里面完成的shuffle:數據重新洗牌

和算子：轉換算子
1.窄依賴算子
2.寬依賴算子

補充：
通過寬窄依賴可以知道 spark-core里面轉換算子哪些算子可能引起shuffle

寬依賴：
xxxbykey shuffle
其他： join reduce
窄依賴：
map filter xxx
spark： stage是如何劃分？ ****
spark-core 產生寬依賴就會劃分stage
算子：引起shuffle 就會劃分stage
一個shuffle算子會劃分2個stage

兩個shuffle算子會產生幾個stage？？？
3

4.shuffle 算子：

“生產上能使用窄依賴算子就不使用寬依賴算子”：
1.不準確
1.生產上大部分需求必須使用寬依賴的

2.容錯的角度： 1.如果經過寬依賴之后的rdd的某一個分區數據掛掉需要去父RDD分區重新計算會把父rdd里面的所有分區都會算一下才行

?
引起shuffle的算子：
? 1.xxxbykey =》
? 2. repartition and coalesce【不準確】
? 3. join：
? map join 不會引發shuffle
? reduce join /common join =》引起shuffle

生產上調整計算的并行度？
coalesce：一般用于減少rdd的分區數 =》窄依賴 =》不會引起shuffle
repartition：增加rdd的分區數
coalesce(num,shuffle=true)

思考：
可不可以使用coalesce 增加rdd分區數？可以的
repartition 減少rdd的分區數？不能
思考：
coalesce 增加rdd分區數？走不走shuffle？必然走shuffle

注意：
生產上用于調整計算的并行度

思考：
rdd =》 hdfs 200個小文件
rdd.coalesce(10) 10個文件

application :
driver:
executor:
yarn container

rdd: => executor
code => executor 判斷code 操作對象是不是rdd里面的元素
=> driver

案例分析：

網站訪問量排名：
domain uid flow
www.baidu.com,uid01,1
www.baidu.com,uid01,10
www.baidu.com,uid02,3
www.baidu.com,uid02,5
www.github.com,uid01,11
www.github.com,uid01,10
www.github.com,uid02,30
www.github.com,uid02,50
www.bibili.com,uid01,110
www.bibili.com,uid01,10
www.bibili.com,uid02,2
www.bibili.com,uid02,3

需求：
每個域名每個用戶的訪問量的top3

sql=> 分組+聚合+ 開窗
code =》分組+聚合 + topn

((www.bibili.com,uid02),5)
((www.github.com,uid02),80)
((www.baidu.com,uid02),8)
((www.github.com,uid01),21)
((www.baidu.com,uid01),11)
((www.bibili.com,uid01),120)

top2:
uid02 www.github.com 80
uid02 www.bibili.com 5

存在安全隱患：
x.toList.

sparkcore 進行數據分析：
rdd進行操作不要使用scala里面的結合進行存儲

思想：
分而治之類似 mr 分組

sparkcore：沒講的
1.廣播變量 =》 sss
2.累加器 =》項目

需求：
spark-core =》 wc
input：hdfs
todo：wc
output：hdfs

idea：
spark code =》 jar =》服務器上

部署spark作業：
1.jar
2.spark-submit 提交作業

// val in = “hdfs://bigdata32:9000/input/”
// val out = “hdfs://bigdata32:9000/output/”

提交spark作業：
spark-submit
–class com.dl2262.sparkcore.day02.WCApp
–master local[2]
–name wordcount
/home/hadoop/project/spark/spark-2262-1.0.jar
hdfs://bigdata32:9000/input/1 hdfs://bigdata32:9000/output/

作業：
1.yarn 提交？

2.–conf 任意Spark配置屬性

–properties-file 要從中加載額外文件的文件路徑

spark-submit \ --class com.dl2262.sparkcore.day02.WCApp \ --master yarn \ --deploy-mode client \ --name wordcount \ --driver-memory 1G /home/hadoop/project/spark/spark-2262-1.0.jar \ hdfs://bigdata32:9000/input/1 hdfs://bigdata32:9000/output/

spark-submit
–class com.dl2262.sparkcore.day02.WCApp
–master local[2]
–name wordcount
–conf spark.input.path=hdfs://bigdata32:9000/input/1
–conf spark.output.path=hdfs://bigdata32:9000/output1/
/home/hadoop/project/spark/spark-2262-1.0.jar

總結

以上是生活随笔為你收集整理的【 RDD持久化】的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

持久
rdd