Spark RDD算子介紹

Spark學習筆記總結

01. Spark基礎

1. 介紹

Spark可以用於批處理、交互式查詢（Spark SQL）、實時流處理（Spark Streaming）、機器學習（Spark MLlib）和圖計算（GraphX）。
Spark是MapReduce的替代方案，而且兼容HDFS、Hive，可融入Hadoop的生態系統，以彌補MapReduce的不足。

2. Spark-Shell

1. spark-shell是Spark自帶的交互式Shell程序，用戶可以在該命令行下用scala編寫spark程序。
2. 直接啟動spark-shell，實質是spark的local模式，在master:8080中並未顯示客戶端連接。
3. 集群模式：
   /usr/local/spark/bin/spark-shell
   --master spark://172.23.27.19:7077
   --executor-memory 2g
   --total-executor-cores 2
4. spark-shell中編寫wordcount
   sc.textFile("hdfs://172.23.27.19:9000/wrd/wc/srcdata/").flatMap(.split(" ")).map((,1)).reduceByKey(+).sortBy(_._2,false).collect

3. RDD介紹與屬性

1. 介紹

RDD（Resilient Distributed Dataset）叫做分布式數據集，是Spark中最基本的數據抽象，它代表一個不可變（創建了內容不可變）、可分區、里面的元素可並行計算的集合。

2. 屬性：

1. 由多個分區組成。對於RDD來說，每個分片都會被一個計算任務處理，並決定並行計算的粒度。
2. 一個計算函數用於每個分區。Spark中RDD的計算是以分片為單位的，每個RDD都會實現compute函數以達到這個目的。
3. RDD之間的依賴關系。RDD的每次轉換都會生成一個新的RDD，所以RDD之間就會形成類似於流水線一樣的前后依賴關系。數據丟失時，根據依賴重新計算丟失的分區而不是整個分區。
4. 一個Partitioner，即RDD的分片函數。默認是HashPartition
5. 分區數據的最佳位置去計算。就是將計算任務分配到其所要處理數據塊的存儲位置。數據本地化。

3. 創建方式：

1. 可通過並行化scala集合創建RDD
   val rdd1 = sc.parallelize(Array(1,2,3,4,5,6,7,8))
2. 通過HDFS支持的文件系統創建，RDD里沒有真的數據，只是記錄了元數據
   val rdd2 = sc.textFile("hdfs://172.23.27.19:9000/wrd/wc/srcdata/")

查看該rdd的分區數量
rdd1.partitions.length

3. 基礎的transformation和action

RDD中兩種算子：
transformation轉換，是延遲加載的

常用的transformation：
（1）map、flatMap、filter
（2）intersection求交集、union求並集：注意類型要一致
distinct:去重
（3）join：類型為(K,V)和(K,W)的RDD上調用，返回一個相同key對應的所有元素對在一起的(K,(V,W))的RDD
（4）groupByKey:在一個(K,V)的RDD上調用，返回一個(K, Iterator[V])的RDD
但是效率reduceByKey較高，因為有一個本地combiner的過程。
（5）cartesian笛卡爾積

常用的action
（1）collect()、count()
（2）reduce：通過func函數聚集RDD中的所有元素
（3）take(n):取前n個；top(2)：排序取前兩個
（4）takeOrdered(n)，排完序后取前n個

4. 較難的transformation和action

參考《http://homepage.cs.latrobe.edu.au/zhe/ZhenHeSparkRDDAPIExamples.html》

（1）mapPartitions(func)和
mapPartitions(func)：
獨立地在RDD的每一個分片上運行，但是返回值；foreachPartition(func)也常用，不需要返回值

mapPartitionsWithIndex(func)：
可以看到分區的編號，以及該分區數據。
類似於mapPartitions，但func帶有一個整數參數表示分片的索引值，func的函數類型必須是
(Int, Interator[T]) => Iterator[U]

val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 2)
val func = (index: Int, iter: Iterator[(Int)]) => {iter.toList.map(x => "[partID:" +  index + ", val: " + x + "]").iterator}
rdd1.mapPartitionsWithIndex(func).collect

（2）aggregate
action操作,
第一個參數是初始值,
第二個參數:是2個函數[每個函數都是2個參數(第一個參數:先對個個分區進行的操作, 第二個:對個個分區合並后的結果再進行合並), 輸出一個參數]

例子：

rdd1.aggregate(0)(_+_, _+_)
//前一個是對每一個分區進行的操作，第二個是對各分區結果進行的結果

rdd1.aggregate(5)(math.max(_, _), _ + _)
//結果：5 + (5+9) = 19

val rdd3 = sc.parallelize(List("12","23","345","4567"),2)
rdd3.aggregate("")((x,y) => math.max(x.length, y.length).toString, (x,y) => x + y)
//結果：24或者42

val rdd4 = sc.parallelize(List("12","23","345",""),2)
rdd4.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y)
//結果01或者10

（3）aggregateByKey
將key值相同的，先局部操作，再整體操作。。和reduceByKey內部實現差不多

val pairRDD = sc.parallelize(List( ("cat",2), ("cat", 5), ("mouse", 4),("cat", 12), ("dog", 12), ("mouse", 2)), 2)
pairRDD.aggregateByKey(0)(math.max(_, _), _ + _).collect
//結果：Array((dog,12), (cat,17), (mouse,6))

PS：
和reduceByKey(+)調用的都是同一個方法，只是aggregateByKey要底層一些，可以先局部再整體操作。

（4）combineByKey
和reduceByKey是相同的效果，是reduceByKey的底層。
第一個參數x:原封不動取出來, 第二個參數:是函數, 局部運算, 第三個:是函數, 對局部運算后的結果再做運算
每個分區中每個key中value中的第一個值,

val rdd1 = sc.textFile("hdfs://master:9000/wordcount/input/").flatMap(_.split(" ")).map((_, 1))
val rdd2 = rdd1.combineByKey(x => x, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n)
rdd2.collect

第一個參數的含義：
每個分區中相同的key中value中的第一個值
如：
(hello,1)(hello,1)(good,1)-->(hello(1,1),good(1))-->x就相當於hello的第一個1, good中的1

val rdd3 = rdd1.combineByKey(x => x + 10, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n)
rdd3.collect
//每個會多加3個10

val rdd4 = sc.parallelize(List("dog","cat","gnu","salmon","rabbit","turkey","wolf","bear","bee"), 3)
val rdd5 = sc.parallelize(List(1,1,2,2,2,1,2,2,2), 3)
val rdd6 = rdd5.zip(rdd4)
val rdd7 = rdd6.combineByKey(List(_), (x: List[String], y: String) => x :+ y, (m: List[String], n: List[String]) => m ++ n)
//將key相同的數據，放入一個集合中

（5）collectAsMap
Action
Map(b -> 2, a -> 1)//將Array的元祖轉換成Map，以后可以通過key取值

val rdd = sc.parallelize(List(("a", 1), ("b", 2)))
rdd.collectAsMap
//可以下一步使用

（6）countByKey
根據key計算key的數量
Action

val rdd1 = sc.parallelize(List(("a", 1), ("b", 2), ("b", 2), ("c", 2), ("c", 1)))
rdd1.countByKey
rdd1.countByValue//將("a", 1)當做一個元素，統計其出現的次數

（7）flatMapValues
對每一個value進行操作后壓平