Spark（四）：RDD算子分類。

時間 2019-11-11

標籤 spark rdd 算子分類欄目 Spark 简体版

原文原文鏈接

RDD算子分類，大體能夠分爲兩類，即：es6

1. Transformation：轉換算子，這類轉換並不觸發提交做業，完成做業中間過程處理。apache

2. Action：行動算子，這類算子會觸發SparkContext提交Job做業。數組

一：Transformation：轉換算子緩存

1. map：app

將原來RDD的每一個數據項經過map中的用戶自定義函數f映射轉變爲一個新的元素。源碼中map算子至關於初始化一個RDD，新RDD叫作MappedRDD(this,sc.clean(f) )。即：dom

map是對RDD中的每一個元素都執行一個指定的函數來產生一個新的RDD。 任何原RDD中的元素在新RDD中都有且只有一個元素與之對應。分佈式

scala> val a = sc.parallelize(1 to 9, 3)
a: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[6] at parallelize at :27

scala> val b = a.map(x => x*3)
b: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[7] at map at :29

scala> a.collect
res7: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)                             

scala> b.collect
res8: Array[Int] = Array(3, 6, 9, 12, 15, 18, 21, 24, 27)

上述例子中把原RDD中每一個元素都乘以3來產生一個新的RDD。函數

2. mapPartitions：oop

mapPartitions函數獲取到每一個分區的迭代器，在函數中經過這個分區總體的迭代器對整個分區的元素進行操做。內部實現是生成MapPartitionsRDD。學習

scala> val a = sc.parallelize(1 to 9, 3)
a: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[10] at parallelize at :27

scala> a.collect
res11: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)                            

scala> var c = a.mapPartitions( a=>a.filter(_>=7) )
c: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[11] at mapPartitions at :29

scala> c.collect
res12: Array[Int] = Array(7, 8, 9)

上述例子是經過函數filter對分區中全部數據進行過濾。

3. mapValues

針對(key,value)型數據中的Value進行操做，而不對Key進行處理。即：

mapValues顧名思義就是輸入函數應用於RDD中Kev-Value的Value，原RDD中的Key保持不變，與新的Value一塊兒組成新的RDD中的元素。所以，該函數只適用於元素爲KV對的RDD。

scala> val a = sc.parallelize(List("Hadoop","HBase","Hive","Spark"), 2)
a: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[12] at parallelize at :27

scala> val b = a.map(x => (x.length,x) )
b: org.apache.spark.rdd.RDD[(Int, String)] = MapPartitionsRDD[13] at map at :29

scala> b.mapValues(""+_+"").collect
res14: Array[(Int, String)] = Array((6,Hadoop), (5,HBase), (4,Hive), (5,Spark))

4. mapWith:

mapWith是map的另一個變種，map只須要一個輸入函數，而mapWith有兩個輸入函數。

eg: 把partition index 乘以10，而後加上2做爲新的RDD的元素.(3 是將十個數分爲三個區)

scala> val x  = sc.parallelize(List(1,2,3,4,5,6,7,8,9,10),3)

scala> x.mapWith( a => a*10 )( (a,b)=>(b+2)).collect
res16: Array[Int] = Array(2, 2, 2, 12, 12, 12, 22, 22, 22, 22)

5. flatMap：

將原來RDD中的每一個元素經過函數f轉換爲新的元素，並將生成的RDD的每一個集合中的元素合併爲一個集合，內部建立FlatMappedRDD(this,sc.clean() )。即：

與map相似，區別是原RDD中的元素經map處理後只能生成一個元素，而原RDD中的元素經flatmap處理後可生成多個元素來構建新RDD。

eg：對原RDD中的每一個元素x產生y個元素（從1到y，y爲元素x的值）。

scala> val a = sc.parallelize(1 to 4,2)

scala> val b = a.flatMap(x => 1 to x )

scala> a.collect
res17: Array[Int] = Array(1, 2, 3, 4)                                           

scala> b.collect
res18: Array[Int] = Array(1, 1, 2, 1, 2, 3, 1, 2, 3, 4)

6. flatMapWith：

flatMapWith與mapWith很相似，都是接收兩個函數，一個函數把partitionIndex做爲輸入，輸出是一個新類型A；另一個函數是以二元組（T,A）做爲輸入，輸出爲一個序列，這些序列裏面的元素組成了新的RDD。

scala> val a = sc.parallelize(List(1,2,3,4,5,6,7,8,9),3)

scala> a.collect
res0: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)                             

scala> a.flatMapWith(x => x,true)((x,y)=>List(y,x)).collect
res1: Array[Int] = Array(0, 1, 0, 2, 0, 3, 1, 4, 1, 5, 1, 6, 2, 7, 2, 8, 2, 9)

7. flatMapWithValues：

flatMapValues相似於mapValues，不一樣的在於flatMapValues應用於元素爲KV對的RDD中Value。每一個一元素的Value被輸入函數映射爲一系列的值，而後這些值再與原RDD中的Key組成一系列新的KV對。

scala> val a = sc.parallelize( List((1,2),(3,4),(3,6)) )

scala> a.collect
res2: Array[(Int, Int)] = Array((1,2), (3,4), (3,6))

scala> val b = a.flatMapValues( x => x.to(5))

scala> b.collect
res3: Array[(Int, Int)] = Array((1,2), (1,3), (1,4), (1,5), (3,4), (3,5))

上述例子中原RDD中每一個元素的值被轉換爲一個序列（從其當前值到5），好比第一個KV對(1,2), 其值2被轉換爲2，3，4，5。而後其再與原KV對中Key組成一系列新的KV對(1,2),(1,3),(1,4),(1,5)。

8. reduce：

reduce將RDD中元素兩兩傳遞給輸入函數，同時產生一個新的值，新產生的值與RDD中下一個元素再被傳遞給輸入函數直到最後只有一個值爲止。

eg：對元素求和。

scala> val a = sc.parallelize(1 to 10 )

scala> a.reduce( (x,y) => x + y )
res5: Int = 55

9. reduceByKey

顧名思義，reduceByKey就是對元素爲KV對的RDD中Key相同的元素的Value進行reduce，所以，Key相同的多個元素的值被reduce爲一個值，而後與原RDD中的Key組成一個新的KV對。

eg：對Key相同的元素的值求和，所以Key爲3的兩個元素被轉爲了(3,10)。

scala> val a = sc.parallelize(List((1,2),(3,4),(3,6)))

scala> a.reduceByKey((x,y)=>x+y).collect
res6: Array[(Int, Int)] = Array((1,2), (3,10))

10. cartesian：

對兩個RDD內的全部元素進行笛卡爾積操做(耗內存)，內部實現返回CartesianRDD。

scala> val a = sc.parallelize(List(1,2,3))

scala> val b = sc.parallelize(List(4,5,6))

scala> val c = a.cartesian(b)

scala> c.collect
res15: Array[(Int, Int)] = Array((1,4), (1,5), (1,6), (2,4), (3,4), (2,5), (2,6), (3,5), (3,6))

11. Sample：

sample將RDD這個集合內的元素進行採樣，獲取全部元素的子集。用戶能夠設定是否有有放回的抽樣，百分比，隨機種子，進而決定採樣方式。

      內部實現： SampledRDD(withReplacement,fraction,seed)。
        函數參數設置：
            withReplacement=true，表示有放回的抽樣。
            withReplacement=false，表示無放回的抽樣。

根據fraction指定的比例，對數據進行採樣，能夠選擇是否用隨機數進行替換，seed用於指定隨機數生成器種子。

scala> val a = sc.parallelize(1 to 100,3)

scala> a.sample(false,0.1,0).count
res16: Long = 12

scala> a.sample(false,0.1,0).collect
res17: Array[Int] = Array(10, 47, 55, 73, 76, 84, 87, 88, 91, 92, 95, 98)

scala> a.sample(true,0.7,scala.util.Random.nextInt(10000)).count
res19: Long = 75

scala> a.sample(true,0.7,scala.util.Random.nextInt(10000)).collect
res20: Array[Int] = Array(1, 3, 3, 3, 5, 6, 9, 9, 9, 9, 10, 10, 15, 17, 20, 23, 23, 27, 28, 31, 32, 32, 34, 35, 36, 36, 36, 36, 38, 39, 41, 42, 42, 43, 45, 47, 49, 49, 50, 50, 51, 51, 54, 55, 55, 57, 57, 57, 57, 57, 59, 59, 61, 61, 63, 67, 72, 74, 76, 76, 80, 80, 81, 81, 81, 82, 83, 85, 87, 88, 90, 93, 95, 96, 97, 97, 99, 100)

12. union:

使用union函數時須要保證兩個RDD元素的數據類型相同，返回的RDD數據類型和被合併的RDD元素數據類型相同。並不進行去重操做，保存全部的元素，若是想去重，可使用distinct()。同時，spark還提供更爲簡潔的使用union的API，即經過++符號至關於union函數操做。

eg： a 與 b 的聯合

scala> val a = sc.parallelize(List(("A",1),("B",2),("c",3),("A",4),("C",5) ))

scala> val b = sc.parallelize(List(("A",5),("B",6),("A",4),("C",9) ))

scala> a.union(b).collect
res22: Array[(String, Int)] = Array((A,1), (B,2), (c,3), (A,4), (C,5), (A,5), (B,6), (A,4), (C,9))

去重複：

scala> val d = sc.parallelize(List(("A",5),("B",6),("A",5) ))

scala> d.distinct.collect
res25: Array[(String, Int)] = Array((B,6), (A,5))

13. groupBy：

將元素經過函數生成相應的Key，數據就轉化爲Key-Value格式，以後將Key相同的元素分爲一組。

eg：根據數據集中的每一個元素的K值對數據分組

scala> val a = sc.parallelize(List(("A",1),("B",2),("c",3),("A",4),("C",5) ))

scala> a.groupByKey().collect
res21: Array[(String, Iterable[Int])] = Array((B,CompactBuffer(2)), (A,CompactBuffer(1, 4)), (C,CompactBuffer(5)), (c,CompactBuffer(3)))

14. join：

join對兩個須要鏈接的RDD進行cogroup函數操做，將相同key的數據能偶放到一個分區，在cgroup操做以後造成新RDD對每一個key下的元素進行笛卡爾積的操做，返回的結果在展平，對應key下的全部元組造成一個集合。最後返回 RDD[(K， (V， W))]。

eg：a與b兩個數據鏈接，至關於表的關聯

scala> val a = sc.parallelize(List(("A",1),("B",2),("c",3),("A",4),("C",5) ))

scala> val b = sc.parallelize(List(("A",5),("B",6),("A",4),("C",9) ))

scala> a.join(b).collect
res23: Array[(String, (Int, Int))] = Array((B,(2,6)), (A,(1,5)), (A,(1,4)), (A,(4,5)), (A,(4,4)), (C,(5,9)))

15. cache：

cache將RDD元素從磁盤緩存到內存。至關於 persist(MEMORY_ONLY) 函數的
功能。

16. persist：

persist函數對RDD進行緩存操做，數據緩存在哪裏，由StorageLevel這個枚舉類型進行肯定。DISK 表明磁盤，MEMORY 表明內存， SER 表明數據是否進行序列化存儲。

函數定義： persist(newLevel:StorageLevel)

StorageLevel 是枚舉類型，表明存儲模式。

MEMORY_AND_DISK_SER 表明數據能夠存儲在內存和磁盤，而且以序列化的方式存儲，其餘同理。

二：Action：行動算子

1. foreach：

foreach對RDD中的每一個元素都應用f函數操做，不返回 RDD 和 Array，而是返回Uint。

scala> val a = sc.parallelize(List(1,2,3,4,5,6,7,8,9),3)

scala> a.foreach(println(_))
4
5
6
7
8
9
1
2
3

2. saveAsTextFile：

函數將數據輸出，存儲到 HDFS 的指定目錄。

函數的內部實現，其內部經過調用 saveAsHadoopFile 進行實現：

this.map(x => (NullWritable.get(), new Text(x.toString)))

.saveAsHadoopFile[TextOutputFormat[NullWritable, Text]](path)

將 RDD 中的每一個元素映射轉變爲 (null， x.toString)，而後再將其寫入 HDFS。

3. collect：

collect至關於toArray，不過已通過時不推薦使用，collect將分佈式的RDD返回爲一個單機的scala Array數據，在這個數組上運用 scala 的函數式操做。

4. count：

count返回整個RDD的元素個數。

scala> val a = sc.parallelize(1 to 10 )

scala> a.collect
res9: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)                         

scala> a.count
res10: Long = 10

更多學習：

http://www.tuicool.com/articles/ZfeQrq7

http://my.oschina.net/lgscofield/blog/497145

http://www.tuicool.com/articles/2iQVr2

《Spark大數據分析實戰》