RDD簡(jiǎn)介

RDD(Resilient Distributed Dataset)叫做分布式數(shù)據(jù)集，是Spark中最基本的數(shù)據(jù)抽象，它代表一個(gè)不可變、可分區(qū)、里面的元素可并行計(jì)算的集合。RDD是一個(gè)類

RDD的屬性

1.一個(gè)列表，存儲(chǔ)存取每個(gè)Partition的優(yōu)先位置(preferred location)。對(duì)于一個(gè)HDFS文件來(lái)說(shuō)，這個(gè)列表保存的就是每個(gè)Partition所在的塊的位置。按照“移動(dòng)數(shù)據(jù)不如移動(dòng)計(jì)算”的理念，Spark在進(jìn)行任務(wù)調(diào)度的時(shí)候，會(huì)盡可能地將計(jì)算任務(wù)分配到其所要處理數(shù)據(jù)塊的存儲(chǔ)位置。

2.保存了計(jì)算每個(gè)分區(qū)的函數(shù)，這個(gè)計(jì)算方法會(huì)應(yīng)用到每一個(gè)數(shù)據(jù)塊上,Spark中RDD的計(jì)算是以分片為單位的，每個(gè)RDD都會(huì)實(shí)現(xiàn)compute函數(shù)以達(dá)到這個(gè)目的。compute函數(shù)會(huì)對(duì)迭代器進(jìn)行復(fù)合，不需要保存每次計(jì)算的結(jié)果。

3.RDD之間的依賴關(guān)系。RDD的每次轉(zhuǎn)換都會(huì)生成一個(gè)新的RDD，所以RDD之間就會(huì)形成類似于流水線一樣的前后依賴關(guān)系。在部分分區(qū)數(shù)據(jù)丟失時(shí)，Spark可以通過(guò)這個(gè)依賴關(guān)系重新計(jì)算丟失的分區(qū)數(shù)據(jù)，而不是對(duì)RDD的所有分區(qū)進(jìn)行重新計(jì)算。

4.RDD的分片函數(shù)(Partitioner)，一個(gè)是基于哈希的HashPartitioner，另外一個(gè)是基于范圍的RangePartitioner。只有對(duì)于于key-value的RDD，才會(huì)有Partitioner，非key-value的RDD的Parititioner的值是None。Partitioner函數(shù)不但決定了RDD本身的分片數(shù)量，也決定了parent RDD Shuffle輸出時(shí)的分片數(shù)量。

5.一組分片(Partition)，即數(shù)據(jù)集的基本組成單位。對(duì)于RDD來(lái)說(shuō)，每個(gè)分片都會(huì)被一個(gè)計(jì)算任務(wù)處理，并決定并行計(jì)算的粒度。用戶可以在創(chuàng)建RDD時(shí)指定RDD的分片個(gè)數(shù)，如果沒(méi)有指定，那么就會(huì)采用默認(rèn)值。默認(rèn)值就是程序所分配到的CPU Core的數(shù)目。

如何創(chuàng)建RDD

1.通過(guò)序列化集合的方式創(chuàng)建RDD(parallelize，makeRDD)

2.通過(guò)讀取外部的數(shù)據(jù)源(testFile)

3.通過(guò)其他的rdd做transformation操作轉(zhuǎn)換成行的RDD

RDD的兩種算子：

1.Transformation

• map(func) :返回一個(gè)新的分布式數(shù)據(jù)集，由每個(gè)原元素經(jīng)過(guò)func函數(shù)轉(zhuǎn)換后組成
• filter(func) : 返回一個(gè)新的數(shù)據(jù)集，由經(jīng)過(guò)func函數(shù)后返回值為true的原元素組成
• flatMap(func) : 類似于map，但是每一個(gè)輸入元素，會(huì)被映射為0到多個(gè)輸出元素(因此，func函數(shù)的返回值是一個(gè)Seq，而不是單一元素)
• flatMap(func) : 類似于map，但是每一個(gè)輸入元素，會(huì)被映射為0到多個(gè)輸出元素(因此，func函數(shù)的返回值是一個(gè)Seq，而不是單一元素)
• sample(withReplacement, frac, seed) :
• 根據(jù)fraction指定的比例對(duì)數(shù)據(jù)進(jìn)行采樣，可以選擇是否使用隨機(jī)數(shù)進(jìn)行替換，seed用于指定隨機(jī)數(shù)生成器種子
• union(otherDataset) : 返回一個(gè)新的數(shù)據(jù)集，由原數(shù)據(jù)集和參數(shù)聯(lián)合而成
• reduceByKey(func, [numTasks]) : 在一個(gè)(K，V)對(duì)的數(shù)據(jù)集上使用，返回一個(gè)(K，V)對(duì)的數(shù)據(jù)集，key相同的值，都被使用指定的reduce函數(shù)聚合到一起。和groupbykey類似，任務(wù)的個(gè)數(shù)是可以通過(guò)第二個(gè)可選參數(shù)來(lái)配置的。
• join(otherDataset, [numTasks]) :
• 在類型為(K,V)和(K,W)類型的數(shù)據(jù)集上調(diào)用，返回一個(gè)(K,(V,W))對(duì)，每個(gè)key中的所有元素都在一起的數(shù)據(jù)集
• groupWith(otherDataset, [numTasks]) : 在類型為(K,V)和(K,W)類型的數(shù)據(jù)集上調(diào)用，返回一個(gè)數(shù)據(jù)集，組成元素為(K, Seq[V], Seq[W]) Tuples。這個(gè)操作在其它框架，稱為CoGroup
• cartesian(otherDataset) : 笛卡爾積。但在數(shù)據(jù)集T和U上調(diào)用時(shí)，返回一個(gè)(T，U)對(duì)的數(shù)據(jù)集，所有元素交互進(jìn)行笛卡爾積。
• intersection(otherDataset):對(duì)源RDD和參數(shù)RDD求交集后返回一個(gè)新的RDD
• distinct([numTasks])) 對(duì)源RDD進(jìn)行去重后返回一個(gè)新的RDD
• groupByKey([numTasks]) 在一個(gè)(K,V)的RDD上調(diào)用，返回一個(gè)(K, Iterator[V])的RDD
• reduceByKey(func, [numTasks]) 在一個(gè)(K,V)的RDD上調(diào)用，返回一個(gè)(K,V)的RDD，使用指定的reduce函數(shù)，將相同key的值聚合到一起，與groupByKey類似，reduce任務(wù)的個(gè)數(shù)可以通過(guò)第二個(gè)可選的參數(shù)來(lái)設(shè)置
• aggregateByKey(zeroValue)(seqOp, combOp, [numTasks])
• sortByKey([ascending], [numTasks]) 在一個(gè)(K,V)的RDD上調(diào)用，K必須實(shí)現(xiàn)Ordered接口，返回一個(gè)按照key進(jìn)行排序的(K,V)的RDD
• sortBy(func,[ascending], [numTasks]) 與sortByKey類似，但是更靈活
• join(otherDataset, [numTasks]) 在類型為(K,V)和(K,W)的RDD上調(diào)用，返回一個(gè)相同key對(duì)應(yīng)的所有元素對(duì)在一起的(K,(V,W))的RDD
• cogroup(otherDataset, [numTasks]) 在類型為(K,V)和(K,W)的RDD上調(diào)用，返回一個(gè)(K,(Iterable

2.Action

• reduce(func) 通過(guò)func函數(shù)聚集RDD中的所有元素，這個(gè)功能必須是課交換且可并聯(lián)的
• collect() 在驅(qū)動(dòng)程序中，以數(shù)組的形式返回?cái)?shù)據(jù)集的所有元素
• count() 返回RDD的元素個(gè)數(shù)
• first() 返回RDD的***個(gè)元素(類似于take(1))
• take(n) 返回一個(gè)由數(shù)據(jù)集的前n個(gè)元素組成的數(shù)組
• takeSample(withReplacement,num, [seed]) 返回一個(gè)數(shù)組，該數(shù)組由從數(shù)據(jù)集中隨機(jī)采樣的num個(gè)元素組成，可以選擇是否用隨機(jī)數(shù)替換不足的部分，seed用于指定隨機(jī)數(shù)生成器種子
• takeOrdered(n, [ordering])
• saveAsTextFile(path) 將數(shù)據(jù)集的元素以textfile的形式保存到HDFS文件系統(tǒng)或者其他支持的文件系統(tǒng)，對(duì)于每個(gè)元素，Spark將會(huì)調(diào)用toString方法，將它裝換為文件中的文本
• saveAsSequenceFile(path) 將數(shù)據(jù)集中的元素以Hadoop sequencefile的格式保存到指定的目錄下，可以使HDFS或者其他Hadoop支持的文件系統(tǒng)。
• saveAsObjectFile(path)
• countByKey() 針對(duì)(K,V)類型的RDD，返回一個(gè)(K,Int)的map，表示每一個(gè)key對(duì)應(yīng)的元素個(gè)數(shù)。
• foreach(func) 在數(shù)據(jù)集的每一個(gè)元素上，運(yùn)行函數(shù)func進(jìn)行更新。

RDD的依賴關(guān)系

1.窄依賴

窄依賴指的是每一個(gè)父RDD的Partition最多被子RDD的一個(gè)Partition使用

總結(jié)：窄依賴我們形象的比喻為獨(dú)生子女

2.寬依賴

寬依賴指的是多個(gè)子RDD的Partition會(huì)依賴同一個(gè)父RDD的Partition

總結(jié)：窄依賴我們形象的比喻為超生

3.Lineage(血統(tǒng))

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

DAG的生成

DAG(Directed Acyclic Graph)叫做有向無(wú)環(huán)圖，原始的RDD通過(guò)一系列的轉(zhuǎn)換就就形成了DAG，根據(jù)RDD之間的依賴關(guān)系的不同將DAG劃分成不同的Stage，對(duì)于窄依賴，partition的轉(zhuǎn)換處理在Stage中完成計(jì)算。對(duì)于寬依賴，由于有Shuffle的存在，只能在parent RDD處理完成后，才能開(kāi)始接下來(lái)的計(jì)算，因此寬依賴是劃分Stage的依據(jù)。

RDD的緩存

Spark速度非?？斓脑蛑?，就是在不同操作中可以在內(nèi)存中持久化或緩存?zhèn)€數(shù)據(jù)集。當(dāng)持久化某個(gè)RDD后，每一個(gè)節(jié)點(diǎn)都將把計(jì)算的分片結(jié)果保存在內(nèi)存中，并在對(duì)此RDD或衍生出的RDD進(jìn)行的其他動(dòng)作中重用。這使得后續(xù)的動(dòng)作變得更加迅速。RDD相關(guān)的持久化和緩存，是Spark最重要的特征之一。可以說(shuō)，緩存是Spark構(gòu)建迭代式算法和快速交互式查詢的關(guān)鍵。

找依賴關(guān)系劃分stage的目的之一就是劃分緩存， 如何通過(guò)stage的劃分設(shè)置緩存?

(1)在窄依賴想設(shè)置緩存時(shí)用cache

(2)在寬依賴想設(shè)置緩存時(shí)用checkpoint

如何設(shè)置cache和checkpoint?

cache：someRDD.cache()就添加成功緩存，放入到內(nèi)存中

someRDD.persist(StorageLevel.MEMORY_AND_DISK)：根據(jù)自己的需要設(shè)置緩存的位置(內(nèi)存和硬盤)

checkpoint：可以把RDD計(jì)算后的數(shù)據(jù)存儲(chǔ)在本地磁盤上，也可以是hdfs

sc.setCheckpointDIr("hdfs://hadoop1:9000/checkpoint")設(shè)置checkpoint的路徑 在寬依賴前設(shè)置

someRDD.checkpoint()設(shè)置checkpoint

cache 和checkpoint的區(qū)別

cache只是緩存數(shù)據(jù)，不改變RDD的依賴關(guān)系，checkpoint生成了一個(gè)新的RDD，后面的RDD將依賴新的RDD依賴關(guān)系已經(jīng)改變 。數(shù)據(jù)恢復(fù)的順序：checkpoint ---》cache--》重算