SparkRDD簡介/常用運算元/依賴/緩存 RDD簡介 RDD（Resilient Distributed Dataset）叫做分散式數據集，是Spark中最基本的數據抽象，它代表一個不可變、可分區、裡面的元素可並行計算的集合。RDD是一個類 RDD的屬性 1.一個列表，存儲存取每個Partitio ...
SparkRDD簡介/常用運算元/依賴/緩存
RDD簡介
RDD（Resilient Distributed Dataset）叫做分散式數據集，是Spark中最基本的數據抽象，它代表一個不可變、可分區、裡面的元素可並行計算的集合。RDD是一個類
RDD的屬性
1.一個列表，存儲存取每個Partition的優先位置（preferred location）。對於一個HDFS文件來說，這個列表保存的就是每個Partition所在的塊的位置。按照“移動數據不如移動計算”的理念，Spark在進行任務調度的時候，會儘可能地將計算任務分配到其所要處理數據塊的存儲位置。
2.保存了計算每個分區的函數，這個計算方法會應用到每一個數據塊上,Spark中RDD的計算是以分片為單位的，每個RDD都會實現compute函數以達到這個目的。compute函數會對迭代器進行複合，不需要保存每次計算的結果。
3.RDD之間的依賴關係。RDD的每次轉換都會生成一個新的RDD，所以RDD之間就會形成類似於流水線一樣的前後依賴關係。在部分分區數據丟失時，Spark可以通過這個依賴關係重新計算丟失的分區數據，而不是對RDD的所有分區進行重新計算。
4.RDD的分片函數（Partitioner），一個是基於哈希的HashPartitioner，另外一個是基於範圍的RangePartitioner。只有對於於key-value的RDD，才會有Partitioner，非key-value的RDD的Parititioner的值是None。Partitioner函數不但決定了RDD本身的分片數量，也決定了parent RDD Shuffle輸出時的分片數量。
5.一組分片（Partition），即數據集的基本組成單位。對於RDD來說，每個分片都會被一個計算任務處理，並決定並行計算的粒度。用戶可以在創建RDD時指定RDD的分片個數，如果沒有指定，那麼就會採用預設值。預設值就是程式所分配到的CPU Core的數目。
如何創建RDD
1.通過序列化集合的方式創建RDD（parallelize，makeRDD）
2.通過讀取外部的數據源(testFile)
3.通過其他的rdd做transformation操作轉換成行的RDD
RDD的兩種運算元：
1.Transformation
- map(func) :返回一個新的分散式數據集，由每個原元素經過func函數轉換後組成
- filter(func) : 返回一個新的數據集，由經過func函數後返回值為true的原元素組成
- flatMap(func) : 類似於map，但是每一個輸入元素，會被映射為0到多個輸出元素（因此，func函數的返回值是一個Seq，而不是單一元素）
- flatMap(func) : 類似於map，但是每一個輸入元素，會被映射為0到多個輸出元素（因此，func函數的返回值是一個Seq，而不是單一元素）
- sample(withReplacement, frac, seed) :
根據fraction指定的比例對數據進行採樣，可以選擇是否使用隨機數進行替換，seed用於指定隨機數生成器種子
- union(otherDataset) : 返回一個新的數據集，由原數據集和參數聯合而成
- reduceByKey(func, [numTasks]) : 在一個（K，V)對的數據集上使用，返回一個（K，V）對的數據集，key相同的值，都被使用指定的reduce函數聚合到一起。和groupbykey類似，任務的個數是可以通過第二個可選參數來配置的。
- join(otherDataset, [numTasks]) :
在類型為（K,V)和（K,W)類型的數據集上調用，返回一個（K,(V,W))對，每個key中的所有元素都在一起的數據集
- groupWith(otherDataset, [numTasks]) : 在類型為（K,V)和(K,W)類型的數據集上調用，返回一個數據集，組成元素為（K, Seq[V], Seq[W]) Tuples。這個操作在其它框架，稱為CoGroup
- cartesian(otherDataset) : 笛卡爾積。但在數據集T和U上調用時，返回一個(T，U）對的數據集，所有元素交互進行笛卡爾積。
- intersection(otherDataset):對源RDD和參數RDD求交集後返回一個新的RDD
- distinct([numTasks])) 對源RDD進行去重後返回一個新的RDD
- groupByKey([numTasks]) 在一個(K,V)的RDD上調用，返回一個(K, Iterator[V])的RDD
reduceByKey(func, [numTasks]) 在一個(K,V)的RDD上調用，返回一個(K,V)的RDD，使用指定的reduce函數，將相同key的值聚合到一起，與groupByKey類似，reduce任務的個數可以通過第二個可選的參數來設置 - aggregateByKey(zeroValue)(seqOp, combOp, [numTasks])
- sortByKey([ascending], [numTasks]) 在一個(K,V)的RDD上調用，K必須實現Ordered介面，返回一個按照key進行排序的(K,V)的RDD
- sortBy(func,[ascending], [numTasks]) 與sortByKey類似，但是更靈活
- join(otherDataset, [numTasks]) 在類型為(K,V)和(K,W)的RDD上調用，返回一個相同key對應的所有元素對在一起的(K,(V,W))的RDD
- cogroup(otherDataset, [numTasks]) 在類型為(K,V)和(K,W)的RDD上調用，返回一個(K,(Iterable
2.Action
- reduce(func) 通過func函數聚集RDD中的所有元素，這個功能必須是課交換且可並聯的
- collect() 在驅動程式中，以數組的形式返回數據集的所有元素
- count() 返回RDD的元素個數
- first() 返回RDD的第一個元素（類似於take(1)）
- take(n) 返回一個由數據集的前n個元素組成的數組
- takeSample(withReplacement,num, [seed]) 返回一個數組，該數組由從數據集中隨機採樣的num個元素組成，可以選擇是否用隨機數替換不足的部分，seed用於指定隨機數生成器種子
- takeOrdered(n, [ordering])
- saveAsTextFile(path) 將數據集的元素以textfile的形式保存到HDFS文件系統或者其他支持的文件系統，對於每個元素，Spark將會調用toString方法，將它裝換為文件中的文本
- saveAsSequenceFile(path) 將數據集中的元素以Hadoop sequencefile的格式保存到指定的目錄下，可以使HDFS或者其他Hadoop支持的文件系統。
- saveAsObjectFile(path)
- countByKey() 針對(K,V)類型的RDD，返回一個(K,Int)的map，表示每一個key對應的元素個數。
- foreach(func) 在數據集的每一個元素上，運行函數func進行更新。
RDD的依賴關係
1.窄依賴
窄依賴指的是每一個父RDD的Partition最多被子RDD的一個Partition使用
總結：窄依賴我們形象的比喻為獨生子女
2.寬依賴
寬依賴指的是多個子RDD的Partition會依賴同一個父RDD的Partition
總結：窄依賴我們形象的比喻為超生
3.Lineage(血統)
RDD只支持粗粒度轉換，即在大量記錄上執行的單個操作。將創建RDD的一系列Lineage（即血統）記錄下來，以便恢復丟失的分區。RDD的Lineage會記錄RDD的元數據信息和轉換行為，當該RDD的部分分區數據丟失時，它可以根據這些信息來重新運算和恢復丟失的數據分區。
DAG的生成
DAG(Directed Acyclic Graph)叫做有向無環圖，原始的RDD通過一系列的轉換就就形成了DAG，根據RDD之間的依賴關係的不同將DAG劃分成不同的Stage，對於窄依賴，partition的轉換處理在Stage中完成計算。對於寬依賴，由於有Shuffle的存在，只能在parent RDD處理完成後，才能開始接下來的計算，因此寬依賴是劃分Stage的依據。
RDD的緩存
Spark速度非常快的原因之一，就是在不同操作中可以在記憶體中持久化或緩存個數據集。當持久化某個RDD後，每一個節點都將把計算的分片結果保存在記憶體中，併在對此RDD或衍生出的RDD進行的其他動作中重用。這使得後續的動作變得更加迅速。RDD相關的持久化和緩存，是Spark最重要的特征之一。可以說，緩存是Spark構建迭代式演算法和快速互動式查詢的關鍵。
找依賴關係劃分stage的目的之一就是劃分緩存， 如何通過stage的劃分設置緩存？
（1）在窄依賴想設置緩存時用cache
（2）在寬依賴想設置緩存時用checkpoint
如何設置cache和checkpoint？
cache：someRDD.cache()就添加成功緩存，放入到記憶體中
someRDD.persist（StorageLevel.MEMORY_AND_DISK）：根據自己的需要設置緩存的位置（記憶體和硬碟）
checkpoint：可以把RDD計算後的數據存儲在本地磁碟上，也可以是hdfs
sc.setCheckpointDIr("hdfs://hadoop1:9000/checkpoint")設置checkpoint的路徑 在寬依賴前設置
someRDD.checkpoint()設置checkpoint
cache 和checkpoint的區別
cache只是緩存數據，不改變RDD的依賴關係，checkpoint生成了一個新的RDD，後面的RDD將依賴新的RDD依賴關係已經改變 。數據恢復的順序：checkpoint ---》cache--》重算
