序列化

閉包檢查

序列化方法和屬性

依賴關(guān)系?

RDD 血緣關(guān)系

RDD 窄依賴

RDD 寬依賴

RDD 任務(wù)劃分

RDD 持久化

RDD Cache 緩存

RDD CheckPoint 檢查點

緩存和檢查點區(qū)別

---

序列化

閉包檢查

????????從計算的角度, 算子以外的代碼都是在 Driver 端執(zhí)行, 算子里面的代碼都是在 Executor 端執(zhí)行。那么在 scala 的函數(shù)式編程中，就會導(dǎo)致算子內(nèi)經(jīng)常會用到算子外的數(shù)據(jù)，這樣就 形成了閉包的效果，如果使用的算子外的數(shù)據(jù)無法序列化，就意味著無法傳值給 Executor 端執(zhí)行，就會發(fā)生錯誤，所以需要在執(zhí)行任務(wù)計算前，檢測閉包內(nèi)的對象是否可以進行序列 化，這個操作我們稱之為閉包檢測。Scala2.12 版本后閉包編譯方式發(fā)生了改變

序列化方法和屬性

????????從計算的角度, 算子以外的代碼都是在 Driver 端執(zhí)行, 算子里面的代碼都是在 Executor 端執(zhí)行。

object spark_02 {def main(args: Array[String]): Unit = {//準(zhǔn)備環(huán)境//"*"代表線程的核數(shù)   應(yīng)用程序名稱"RDD"val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")val sc = new SparkContext(sparkConf)val rdd: RDD[String] = sc.makeRDD(Array("hello world", "hello spark", "hive", "atguigu"))//創(chuàng)建查詢對象val search = new Search("h")//函數(shù)傳遞，打印：ERROR Task not serializablesearch.getMatch1(rdd).collect().foreach(println)println("========================================")//屬性傳遞，打印：ERROR Task not serializablesearch.getMatch2(rdd).collect().foreach(println)//關(guān)閉環(huán)境sc.stop()}
}//查詢對象
//類的構(gòu)造參數(shù)是類的屬性，構(gòu)造參數(shù)需要進行閉包檢查（對類進行閉包檢查）
class Search(query:String) extends Serializable {def isMatch(s: String): Boolean = {s.contains(query)}// 函數(shù)序列化案例def getMatch1 (rdd: RDD[String]): RDD[String] = {rdd.filter(isMatch)}// 屬性序列化案例def getMatch2(rdd: RDD[String]): RDD[String] = {rdd.filter(x => x.contains(query))}
}

依賴關(guān)系?

????????相鄰的兩個RDD關(guān)系稱之為依賴關(guān)系

RDD 血緣關(guān)系

????????多個連續(xù)的RDD的依賴關(guān)系稱之為血緣關(guān)系

????????RDD 只支持粗粒度轉(zhuǎn)換，即在大量記錄上執(zhí)行的單個操作。將創(chuàng)建 RDD 的一系列 Lineage （血統(tǒng)）記錄下來，以便恢復(fù)丟失的分區(qū)。RDD 的 Lineage 會記錄 RDD 的元數(shù)據(jù)信息和轉(zhuǎn) 換行為，當(dāng)該 RDD 的部分分區(qū)數(shù)據(jù)丟失時，它可以根據(jù)這些信息來重新運算和恢復(fù)丟失的 數(shù)據(jù)分區(qū)。

val fileRDD: RDD[String] = sc.textFile("input/1.txt")
println(fileRDD.toDebugString) //打印輸出血緣關(guān)系
println("----------------------")
val wordRDD: RDD[String] = fileRDD.flatMap(_.split(" "))
println(wordRDD.toDebugString)
println("----------------------")
val mapRDD: RDD[(String, Int)] = wordRDD.map((_,1))
println(mapRDD.toDebugString)
println("----------------------")
val resultRDD: RDD[(String, Int)] = mapRDD.reduceByKey(_+_)
println(resultRDD.toDebugString)
resultRDD.collect()

RDD 窄依賴

????????窄依賴表示每一個父(上游)RDD 的 Partition 最多被子（下游）RDD 的一個 Partition 使用， 窄依賴我們形象的比喻為獨生子女。

RDD 寬依賴

????????寬依賴表示同一個父（上游）RDD 的 Partition 被多個子（下游）RDD 的 Partition 依賴，會 引起 Shuffle，總結(jié)：寬依賴我們形象的比喻為多生。

RDD 任務(wù)劃分

????????RDD 任務(wù)切分中間分為：Application、Job、Stage 和 Task：

• Application：初始化一個 SparkContext 即生成一個 Application；
• Job：一個 Action 算子就會生成一個 Job；
• Stage：Stage 等于寬依賴(ShuffleDependency)的個數(shù)加 1；
• Task：一個 Stage 階段中，最后一個 RDD 的分區(qū)個數(shù)就是 Task 的個數(shù)。

注意：Application->Job->Stage->Task 每一層都是 1 對 n 的關(guān)系。

RDD 持久化

RDD Cache 緩存

????????RDD 通過 Cache 或者 Persist 方法將前面的計算結(jié)果緩存，默認(rèn)情況下會把數(shù)據(jù)以緩存 在 JVM 的堆內(nèi)存中。但是并不是這兩個方法被調(diào)用時立即緩存，而是觸發(fā)后面的 action 算 子時，該 RDD 將會被緩存在計算節(jié)點的內(nèi)存中，并供后面重用。

// cache 操作會增加血緣關(guān)系，不改變原有的血緣關(guān)系
println(wordToOneRdd.toDebugString)
// 數(shù)據(jù)緩存。
wordToOneRdd.cache()
// 可以更改存儲級別
//mapRdd.persist(StorageLevel.MEMORY_AND_DISK_2)

????????緩存有可能丟失，或者存儲于內(nèi)存的數(shù)據(jù)由于內(nèi)存不足而被刪除，RDD 的緩存容錯機 制保證了即使緩存丟失也能保證計算的正確執(zhí)行。通過基于 RDD 的一系列轉(zhuǎn)換，丟失的數(shù) 據(jù)會被重算，由于 RDD 的各個 Partition 是相對獨立的，因此只需要計算丟失的部分即可， 并不需要重算全部 Partition。 Spark 會自動對一些 Shuffle 操作的中間數(shù)據(jù)做持久化操作(比如：reduceByKey)。這樣 做的目的是為了當(dāng)一個節(jié)點 Shuffle 失敗了避免重新計算整個輸入。但是，在實際使用的時 候，如果想重用數(shù)據(jù)，仍然建議調(diào)用 persist 或 cache。

RDD CheckPoint 檢查點

????????所謂的檢查點其實就是通過將 RDD 中間結(jié)果寫入磁盤 由于血緣依賴過長會造成容錯成本過高，這樣就不如在中間階段做檢查點容錯，如果檢查點 之后有節(jié)點出現(xiàn)問題，可以從檢查點開始重做血緣，減少了開銷。 對 RDD 進行 checkpoint 操作并不會馬上被執(zhí)行，必須執(zhí)行 Action 操作才能觸發(fā)。

// 設(shè)置檢查點路徑
sc.setCheckpointDir("./checkpoint1")
// 創(chuàng)建一個 RDD，讀取指定位置文件:hello atguigu atguigu
val lineRdd: RDD[String] = sc.textFile("input/1.txt")
// 業(yè)務(wù)邏輯
val wordRdd: RDD[String] = lineRdd.flatMap(line => line.split(" "))
val wordToOneRdd: RDD[(String, Long)] = wordRdd.map {word => {(word, System.currentTimeMillis())}
}
// 增加緩存,避免再重新跑一個 job 做 checkpoint
wordToOneRdd.cache()
// 數(shù)據(jù)檢查點：針對 wordToOneRdd 做檢查點計算
wordToOneRdd.checkpoint()
// 觸發(fā)執(zhí)行邏輯
wordToOneRdd.collect().foreach(println)

緩存和檢查點區(qū)別

1. Cache 緩存只是將數(shù)據(jù)保存起來，不切斷血緣依賴。Checkpoint 檢查點切斷血緣依賴。
2. Cache 緩存的數(shù)據(jù)通常存儲在磁盤、內(nèi)存等地方，可靠性低。Checkpoint 的數(shù)據(jù)通常存 儲在 HDFS 等容錯、高可用的文件系統(tǒng)，可靠性高。
3. 建議對 checkpoint()的 RDD 使用 Cache 緩存，這樣 checkpoint 的 job 只需從 Cache 緩存 中讀取數(shù)據(jù)即可，否則需要再從頭計算一次 RDD。