共計 7346 個字符，預計需要花費 19 分鐘才能閱讀完成。
本篇內容主要講解“Spark RDD 算子分為哪幾類”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓丸趣 TV 小編來帶大家學習“Spark RDD 算子分為哪幾類”吧!
RDD 算子分類，大致可以分為兩類，即：
1. Transformation：轉換算子，這類轉換并不觸發提交作業，完成作業中間過程處理。
2. Action：行動算子，這類算子會觸發 SparkContext 提交 Job 作業。
一：Transformation：轉換算子
1. map：
將原來 RDD 的每個數據項通過 map 中的用戶自定義函數 f 映射轉變為一個新的元素。源碼中 map 算子相當于初始化一個 RDD，新 RDD 叫做 MappedRDD(this,sc.clean(f) )。即：
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：
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 = 559. 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(_))
32. 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到此，相信大家對“Spark RDD 算子分為哪幾類”有了更深的了解，不妨來實際操作一番吧！這里是丸趣 TV 網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！
