大數(shù)據(jù)學習路線分享彈性分布式數(shù)據(jù)集RDD，RDD定義，RDD（Resilient Distributed Dataset）叫做分布式數(shù)據(jù)集，是Spark中最基本的數(shù)據(jù)抽象，它代表一個不可變(數(shù)據(jù)和元數(shù)據(jù))、可分區(qū)、里面的元素可并行計算的集合。
RDD的特點:自動容錯,位置感知性調(diào)度和可伸縮性
RDD的屬性
1.一組分片
即數(shù)據(jù)集的基本組成單位。對于RDD來說，每個分片都會被一個計算任務(wù)處理，并決定并行計算的粒度。用戶可以在創(chuàng)建RDD時指定RDD的分片個數(shù)，如果沒有指定，那么就會采用默認值。默認值就是程序所分配到的CPU Core的數(shù)目。
2.一個計算每個分區(qū)的函數(shù)。
Spark中RDD的計算是以分片為單位的，每個RDD都會實現(xiàn)compute函數(shù)以達到這個目的。compute函數(shù)會對迭代器進行復(fù)合，不需要保存每次計算的結(jié)果。
3.RDD之間的依賴關(guān)系。
RDD的每次轉(zhuǎn)換都會生成一個新的RDD，所以RDD之間就會形成類似于流水線一樣的前后依賴關(guān)系。
容錯處理: 在部分分區(qū)數(shù)據(jù)丟失時，Spark可以通過這個依賴關(guān)系重新計算丟失的分區(qū)數(shù)據(jù)，而不是對RDD的所有分區(qū)進行重新計算。
4.一個Partitioner,分區(qū)器
即RDD的分片函數(shù)。當前Spark中實現(xiàn)了兩種類型的分片函數(shù)，一個是基于哈希的HashPartitioner，另外一個是基于范圍的RangePartitioner。只有對于key-value的RDD，才會有Partitioner，非key-value的RDD的Parititioner的值是None。Partitioner函數(shù)不但決定了RDD本身的分片數(shù)量，也決定了parent RDD Shuffle輸出時的分片數(shù)量。
5.一個列表
存儲存取每個Partition的優(yōu)先位置（preferred location）。-> 就近原則
對于一個HDFS文件來說，這個列表保存的就是每個Partition所在的塊的位置。按照“移動數(shù)據(jù)不如移動計算”的理念，Spark在進行任務(wù)調(diào)度的時候，會盡可能地將計算任務(wù)分配到其所要處理數(shù)據(jù)塊的存儲位置。
RDD類型
1.Transformation -> 記錄計算過程(記錄參數(shù),計算方法)
轉(zhuǎn)換
含義
map(func)
返回一個新的RDD，該RDD由每一個輸入元素經(jīng)過func函數(shù)轉(zhuǎn)換后組成
filter(func)
返回一個新的RDD，該RDD由經(jīng)過func函數(shù)計算后返回值為true的輸入元素組成
flatMap(func)
類似于map，但是每一個輸入元素可以被映射為0或多個輸出元素（所以func應(yīng)該返回一個序列，而不是單一元素）
mapPartitions(func)
類似于map，但獨立地在RDD的每一個分片上運行，因此在類型為T的RDD上運行時，func的函數(shù)類型必須是Iterator[T] => Iterator[U]
mapPartitionsWithIndex(func)
類似于mapPartitions，但func帶有一個整數(shù)參數(shù)表示分片的索引值，因此在類型為T的RDD上運行時，func的函數(shù)類型必須是
(Int, Iterator[T]) => Iterator[U]
sample(withReplacement, fraction, seed)
根據(jù)fraction指定的比例對數(shù)據(jù)進行采樣，可以選擇是否使用隨機數(shù)進行替換，seed用于指定隨機數(shù)生成器種子
union(otherDataset)
對源RDD和參數(shù)RDD求并集后返回一個新的RDD
intersection(otherDataset)
diff -> 差集
對源RDD和參數(shù)RDD求交集后返回一個新的RDD
distinct([numTasks]))
[改變分區(qū)數(shù)]
對源RDD進行去重后返回一個新的RDD
groupByKey([numTasks])
在一個(K,V)的RDD上調(diào)用，返回一個(K, Iterator[V])的RDD
reduceByKey(func, [numTasks])
在一個(K,V)的RDD上調(diào)用，返回一個(K,V)的RDD，使用指定的reduce函數(shù)，將相同key的值聚合到一起，與groupByKey類似，reduce任務(wù)的個數(shù)可以通過第二個可選的參數(shù)來設(shè)置
aggregateByKey(zeroValue)(seqOp, combOp, [numTasks])
sortByKey([ascending], [numTasks])
在一個(K,V)的RDD上調(diào)用，K必須實現(xiàn)Ordered接口，返回一個按照key進行排序的(K,V)的RDD
sortBy(func,[ascending], [numTasks])
與sortByKey類似，但是更靈活
join(otherDataset, [numTasks])
在類型為(K,V)和(K,W)的RDD上調(diào)用，返回一個相同key對應(yīng)的所有元素對在一起的(K,(V,W))的RDD
cogroup(otherDataset, [numTasks])
在類型為(K,V)和(K,W)的RDD上調(diào)用，返回一個(K,(Iterable<V>,Iterable<W>))類型的RDD
cartesian(otherDataset)
笛卡爾積
pipe(command, [envVars])
coalesce(numPartitions)
repartition(numPartitions)
重新分區(qū)
repartitionAndSortWithinPartitions(partitioner)
2.Action -> 觸發(fā)生成job(一個job對應(yīng)一個action算子)
動作
含義
reduce(func)
通過func函數(shù)聚集RDD中的所有元素，這個功能必須是可交換且可并聯(lián)的
collect()
在驅(qū)動程序中，以數(shù)組的形式返回數(shù)據(jù)集的所有元素
count()
返回RDD的元素個數(shù)
first()
返回RDD的第一個元素（類似于take(1)）
take(n)
取數(shù)據(jù)集的前n個元素組成的數(shù)組
takeSample(withReplacement,num, [seed])
返回一個數(shù)組，該數(shù)組由從數(shù)據(jù)集中隨機采樣的num個元素組成，可以選擇是否用隨機數(shù)替換不足的部分，seed用于指定隨機數(shù)生成器種子
takeOrdered(n, [ordering])
takeOrdered和top類似，只不過以和top相反的順序返回元素
saveAsTextFile(path)
將數(shù)據(jù)集的元素以textfile的形式保存到HDFS文件系統(tǒng)或者其他支持的文件系統(tǒng)，對于每個元素，Spark將會調(diào)用toString方法，將它裝換為文件中的文本
saveAsSequenceFile(path)
將數(shù)據(jù)集中的元素以Hadoop sequencefile的格式保存到指定的目錄下，可以使HDFS或者其他Hadoop支持的文件系統(tǒng)。
saveAsObjectFile(path)
countByKey()
針對(K,V)類型的RDD，返回一個(K,Int)的map，表示每一個key對應(yīng)的元素個數(shù)。
foreach(func)
在數(shù)據(jù)集的每一個元素上，運行函數(shù)func進行更新。
創(chuàng)建RDD
Linux進入sparkShell:
/usr/local/spark.../bin/spark-shell \
--master spark://hadoop01:7077 \
--executor-memory 512m \
--total-executor-cores 2
或在Maven下:
object lx03 {
def main(args: Array[String]): Unit = {
val conf : SparkConf = new SparkConf()
.setAppName("SparkAPI")
.setMaster("local[*]")
val sc: SparkContext = new SparkContext(conf)
//通過并行化生成rdd
val rdd1: RDD[Int] = sc.parallelize(List(24,56,3,2,1))
//對add1的每個元素乘以2然后排序
val rdd2: RDD[Int] = rdd1.map(_ * 2).sortBy(x => x,true)
println(rdd2.collect().toBuffer)
//過濾出大于等于10的元素
// val rdd3: RDD[Int] = rdd2.filter(_ >= 10)
// println(rdd3.collect().toBuffer)
}
練習2
val rdd1 = sc.parallelize(Array("a b c", "d e f", "h i j"))
//將rdd1里面的每一個元素先切分在壓平
val rdd2 = rdd1.flatMap(_.split(' '))
rdd2.collect
//復(fù)雜的：
val rdd1 = sc.parallelize(List(List("a b c", "a b b"), List("e f g", "a f g"), List("h i j", "a a b")))
//將rdd1里面的每一個元素先切分在壓平
val rdd2 = rdd1.flatMap(_.flatMap(_.split(" ")))
練習3
val rdd1 = sc.parallelize(List(5, 6, 4, 3))
val rdd2 = sc.parallelize(List(1, 2, 3, 4))
//求并集
val rdd3 = rdd1.union(rdd2)
//求交集
val rdd4 = rdd1.intersection(rdd2)
//去重
rdd3.distinct.collect
rdd4.collect
練習4
val rdd1 = sc.parallelize(List(("tom", 1), ("jerry", 3), ("kitty", 2)))
val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 1), ("shuke", 2)))
//求join
val rdd3 = rdd1.join(rdd2) -> 相同的key組成新的key,value
//結(jié)果: Array[(String,(Int,Int))] = Array((tom,(1,1)),(jerry,(3,2)))
rdd3.collect
//求左連接和右連接
val rdd3 = rdd1.leftOuterJoin(rdd2)
rdd3.collect
val rdd3 = rdd1.rightOuterJoin(rdd2)
rdd3.collect
//求并集
val rdd4 = rdd1 union rdd2
//按key進行分組
rdd4.groupByKey
rdd4.collect
//分別用groupByKey和reduceByKey實現(xiàn)單詞計數(shù)
val rdd3 = rdd1 union rdd2
rdd3.groupByKey().mapValues(_.sum).collect
rdd3.reduceByKey(_+_).collect
groupByKey和reduceByKey的區(qū)別
reduceByKey算子比較特殊,它首先會進行局部聚合,再全局聚合,我們只需要傳一個局部聚合的函數(shù)就可以了
練習5
val rdd1 = sc.parallelize(List(("tom", 1), ("tom", 2), ("jerry", 3), ("kitty", 2)))
val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 1), ("shuke", 2)))
//cogroup
val rdd3 = rdd1.cogroup(rdd2)
//注意cogroup與groupByKey的區(qū)別
rdd3.collect
val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5))
//reduce聚合
val rdd2 = rdd1.reduce(_ + _)
//按value的降序排序
val rdd5 = rdd4.map(t => (t._2, t._1)).sortByKey(false).map(t => (t._2, t._1))
rdd5.collect
//笛卡爾積
val rdd3 = rdd1.cartesian(rdd2)
計算元素個數(shù)
scala> val rdd1 = sc.parallelize(List(2,3,1,5,7,3,4))
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:27
scala> rdd1.count
res0: Long = 7
top先升序排序在取值
scala> rdd1.top(3)
res1: Array[Int] = Array(7, 5, 4)
scala> rdd1.top(0)
res2: Array[Int] = Array()
scala> rdd1.top(100)
res3: Array[Int] = Array(7, 5, 4, 3, 3, 2, 1)
take原集合前N個,有幾個取幾個
scala> rdd1.take(3)
res4: Array[Int] = Array(2, 3, 1)
scala> rdd1.take(100)
res5: Array[Int] = Array(2, 3, 1, 5, 7, 3, 4)
scala> rdd1.first
res6: Int = 2
takeordered倒序排序再取值
scala> rdd1.takeOrdered(3)
res7: Array[Int] = Array(1, 2, 3)
scala> rdd1.takeOrdered(30)
res8: Array[Int] = Array(1, 2, 3, 3, 4, 5, 7)
生成RDD的兩種方式
1.并行化方式生成 (默認分區(qū)兩個)
手動指定分區(qū)
scala> val rdd1 = sc.parallelize(List(1,2,3,5))
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[5] at parallelize at <console>:27
scala> rdd1.partitions.length //獲取分區(qū)數(shù)
res9: Int = 2
scala> val rdd1 = sc.parallelize(List(1,2,3,5),3)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[6] at parallelize at <console>:27
scala> rdd1.partitions.length
res10: Int = 3
2.使用textFile讀取文件存儲系統(tǒng)里的數(shù)據(jù)
scala> val rdd2 = sc.textFile("hdfs://hadoop01:9000/wordcount/input/a.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)
rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[11] at reduceByKey at <console>:27
scala> rdd2.collect //調(diào)用算子得到RDD顯示結(jié)果
res11: Array[(String, Int)] = Array((hello,6), (beijing,1), (java,1), (gp1808,1), (world,1), (good,1), (qianfeng,1))
scala> val rdd2 = sc.textFile("hdfs://hadoop01:9000/wordcount/input/a.txt",4).flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)
rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[26] at reduceByKey at <console>:27
scala> rdd2.partitions.length //也可以自己指定分區(qū)數(shù)
res15: Int = 4
