Spark RDD-轉化算子

時間 2019-12-01

標籤 spark rdd 轉化算子欄目 Spark 简体版

原文原文鏈接

2.RDD 編程

2.1 編程模型

　　在 Spark 中，RDD 被表示爲對象，經過對象上的方法調用來對 RDD 進行轉換。通過

一系列的 transformations 定義 RDD 以後，就能夠調用 actions 觸發 RDD 的計算，action 可

以是嚮應用程序返回結果(count, collect 等)，或者是向存儲系統保存數據(saveAsTextFile

等)。在 Spark 中，只有遇到 action，纔會執行 RDD 的計算(即延遲計算)，這樣在運行時可

以經過管道的方式傳輸多個轉換。

　　要使用 Spark，開發者須要編寫一個 Driver 程序，它被提交到集羣以調度運行

Worker，以下圖所示。Driver 中定義了一個或多個 RDD，並調用 RDD 上的 action，Worker

則執行 RDD 分區計算任務。

2.2 RDD 的建立

　　在 Spark 中建立 RDD 的建立方式能夠分爲三種：從集合中建立 RDD；從外部存儲創

建 RDD；從其餘 RDD 建立。

2.2.1 從集合中建立

從集合中建立 RDD，Spark 主要提供了兩種函數：parallelize 和 makeRDD

1）使用 parallelize()從集合建立

scala> val rdd = sc.parallelize(Array(1,2,3,4,5,6,7,8))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:24

2）使用 makeRDD()從集合建立

scala> val rdd1 = sc.makeRDD(Array(1,2,3,4,5,6,7,8))
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[1] at makeRDD at <console>:24

2.2.2 由外部存儲系統的數據集建立

包括本地的文件系統，還有全部 Hadoop 支持的數據集，好比 HDFS、Cassandra、HBase

等，以後詳細介紹。

scala> val rdd2= sc.textFile("hdfs://hadoop102:9000/RELEASE")
rdd2: org.apache.spark.rdd.RDD[String] = hdfs:// hadoop102:9000/RELEASE 
MapPartitionsRDD[4] at textFile at <console>:24

2.2.3 從其餘 RDD 建立

後面詳解java

2.3 RDD 的轉換（面試開發重點）

RDD 總體上分爲 Value 類型和 Key-Value 類型

2.3.1 Value 類型

2.3.1.1 map(func)案例

1. 做用：返回一個新的 RDD，該 RDD 由每個輸入元素通過 func 函數轉換後組成

2. 需求：建立一個 1-10 數組的 RDD，將全部元素*2 造成新的 RDD

（1）建立

scala> var source = sc.parallelize(1 to 10)
source: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[8] at parallelize at <console>:24

（2）打印

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

（3）將全部元素*2

scala> val mapadd = source.map(_ * 2)
mapadd: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[9] at map at <console>:26

（4）打印最終結果

scala> mapadd.collect()
res8: Array[Int] = Array(2, 4, 6, 8, 10, 12, 14, 16, 18, 20)

測試：es6

scala> val rdd = sc.parallelize(Array(1,2,3,4,5))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[8] at parallelize at <console>:24

scala> rdd.map(_+1).collect
res16: Array[Int] = Array(2, 3, 4, 5, 6)

scala> rdd.map((_,1)).collect
res17: Array[(Int, Int)] = Array((1,1), (2,1), (3,1), (4,1), (5,1))

2.3.1.2 mapPartitions(func) 案例

1. 做用：相似於 map，但獨立地在 RDD 的每個分片上運行，所以在類型爲 T 的 RDD

上運行時，func 的函數類型必須是 Iterator[T] => Iterator[U]。假設有 N 個元素，有 M 個分

區，那麼 map 的函數的將被調用 N 次,而 mapPartitions 被調用 M 次,一個函數一次處理全部

分區。

2. 需求：建立一個 RDD，使每一個元素*2 組成新的 RDD

（1）建立一個 RDD

scala> val rdd = sc.parallelize(Array(1,2,3,4))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[4] at parallelize at <console>:24

（2）使每一個元素*2 組成新的 RDD

scala> rdd.mapPartitions(x=>x.map(_*2))
res3: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[6] at mapPartitions at <console>:27

（3）打印新的 RDD

scala> res3.collect
res4: Array[Int] = Array(2, 4, 6, 8)

測試：面試

scala> rdd.partitions.size
res18: Int = 4

scala> rdd.mapPartitions(x => Iterator(x.mkString("|")))
res20: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[11] at mapPartitions at <console>:27

scala> rdd.mapPartitions(x => Iterator(x.mkString("|"))).collect
res21: Array[String] = Array(1, 2, 3, 4|5)

2.3.1.3 mapPartitionsWithIndex(func) 案例

1. 做用：相似於 mapPartitions，但 func 帶有一個整數參數表示分片的索引值，所以在類型

爲 T 的 RDD 上運行時，func 的函數類型必須是(Int, Interator[T]) => Iterator[U]；

2. 需求：建立一個 RDD，使每一個元素跟所在分區造成一個元組組成一個新的 RDD

（1）建立一個 RDD

scala> val rdd = sc.parallelize(Array(1,2,3,4))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[4] at parallelize at <console>:24

（2）使每一個元素跟所在分區造成一個元組組成一個新的 RDDshell

scala> val indexRdd = rdd.mapPartitionsWithIndex((index,items)=>(items.map((index,_))))
indexRdd: org.apache.spark.rdd.RDD[(Int, Int)] = MapPartitionsRDD[5] at mapPartitionsWithIndex at <console>:26

（3）打印新的 RDD

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

測試：apache

val rdd = sc.makeRDD(Array(1,2,3,4,5))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at makeRDD at <console>:24

scala> rdd.partitions.size
res1: Int = 4

scala> rdd.mapPartitions
mapPartitions   mapPartitionsWithIndex

scala> rdd.mapPartitionsWithIndex((x,y) => Iterator(x+":"+y.mkString("|"))).collect
res2: Array[String] = Array(0:1, 1:2, 2:3, 3:4|5)                               

scala> sc.makeRDD(Array(1,2,3,4),5).mapPartitionsWithIndex((x,y) => Iterator(x+":"+y.mkString("|"))).collect
res3: Array[String] = Array(0:, 1:1, 2:2, 3:3, 4:4)

2.3.1.4 flatMap(func) 案例

1. 做用：相似於 map，可是每個輸入元素能夠被映射爲 0 或多個輸出元素（因此 func 應

該返回一個序列，而不是單一元素）

2. 需求：建立一個元素爲 1-5 的 RDD，運用 flatMap 建立一個新的 RDD，新的 RDD 爲原

RDD 的每一個元素的擴展（1->1,2->1,2……5->1,2,3,4,5）

（1）建立

scala> val sourceFlat = sc.parallelize(1 to 5)
sourceFlat: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[12] at parallelize at <console>:24

（2）打印

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

（3）根據原 RDD 建立新 RDD（1->1,2->1,2……5->1,2,3,4,5）

scala> val flatMap = sourceFlat.flatMap(1 to _)
flatMap: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[13] at flatMap at <console>:26

（4）打印新 RDD

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

測試：編程

scala> val rdd = sc.makeRDD(Array(1,2,3,4,5))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at makeRDD at <console>:24 scala> rdd.flatMap(x => Array(x + 1)) res1: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[2] at flatMap at <console>:27 scala> rdd.flatMap(x => Array(x + 1)).collect res2: Array[Int] = Array(2, 3, 4, 5, 6)

2.3.1.5 map()和 mapPartition()的區別

1. map()：每次處理一條數據。

2. mapPartition()：每次處理一個分區的數據，這個分區的數據處理完後，原 RDD 中分區的數據才能釋放，可能致使 OOM。

3. 開發指導：當內存空間較大的時候建議使用 mapPartition()，以提升處理效率。

2.3.1.6 glom 案例

1. 做用：將每個分區造成一個數組，造成新的 RDD 類型時 RDD[Array[T]]

2. 需求：建立一個 4 個分區的 RDD，並將每一個分區的數據放到一個數組

（1）建立

scala> val rdd = sc.parallelize(1 to 16,4)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[65] at parallelize at <console>:24

（2）將每一個分區的數據放到一個數組並收集到 Driver 端打印

scala> rdd.glom().collect()
res25: Array[Array[Int]] = Array(Array(1, 2, 3, 4), Array(5, 6, 7, 8), Array(9, 10, 11, 12), Array(13,14, 15, 16))

2.3.1.7 groupBy(func)案例

1. 做用：分組，按照傳入函數的返回值進行分組。將相同的 key 對應的值放入一個迭代器。

2. 需求：建立一個 RDD，按照元素模以 2 的值進行分組。

（1）建立

scala> val rdd = sc.parallelize(1 to 4)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[65] at parallelize at <console>:24

（2）按照元素模以 2 的值進行分組

scala> val group = rdd.groupBy(_%2)
group: org.apache.spark.rdd.RDD[(Int, Iterable[Int])] = ShuffledRDD[2] at groupBy at <console>:26

（3）打印結果

scala> group.collect
res0: Array[(Int, Iterable[Int])] = Array((0,CompactBuffer(2, 4)), (1,CompactBuffer(1, 3)))

2.3.1.8 filter(func) 案例

1. 做用：過濾。返回一個新的 RDD，該 RDD 由通過 func 函數計算後返回值爲 true 的輸

入元素組成。

2. 需求：建立一個 RDD（由字符串組成），過濾出一個新 RDD（包含」xiao」子串）

（1）建立

scala> var sourceFilter = sc.parallelize(Array("xiaoming","xiaojiang","xiaohe","dazhi"))
sourceFilter: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[10] at parallelize at <console>:24

（2）打印

scala> sourceFilter.collect()
res9: Array[String] = Array(xiaoming, xiaojiang, xiaohe, dazhi)

（3）過濾出含」 xiao」子串的造成一個新的 RDD

scala> val filter = sourceFilter.filter(_.contains("xiao"))
filter: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[11] at filter at <console>:26

（4）打印新 RDD

scala> filter.collect()
res10: Array[String] = Array(xiaoming, xiaojiang, xiaohe)

測試：數組

scala> val rdd = sc.makeRDD(Array(1,2,3,4,5))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at makeRDD at <console>:24 scala> rdd.filter(_%2 == 0).collect res3: Array[Int] = Array(2, 4)

2.3.1.9 sample(withReplacement, fraction, seed) 案例

1. 做用：以指定的隨機種子隨機抽樣出數量爲 fraction 的數據，withReplacement 表示是抽

出的數據是否放回，true 爲有放回的抽樣，false 爲無放回的抽樣，seed 用於指定隨機數生

成器種子。

2. 需求：建立一個 RDD（1-10），從中選擇放回和不放回抽樣

（1）建立 RDD

scala> val rdd = sc.parallelize(1 to 10)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[20] at parallelize at <console>:24

（2）打印

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

（3）放回抽樣

scala> var sample1 = rdd.sample(true,0.4,2)
sample1: org.apache.spark.rdd.RDD[Int] = PartitionwiseSampledRDD[21] at sample at <console>:26

（4）打印放回抽樣結果

scala> sample1.collect()
res16: Array[Int] = Array(1, 2, 2, 7, 7, 8, 9)

（5）不放回抽樣

scala> var sample2 = rdd.sample(false,0.2,3)
sample2: org.apache.spark.rdd.RDD[Int] = PartitionwiseSampledRDD[22] at sample at <console>:26

（6）打印不放回抽樣結果

scala> sample2.collect()
res17: Array[Int] = Array(1, 9)

測試：網絡

scala> val rdd = sc.parallelize(1 to 20)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[4] at parallelize at <console>:24

scala> rdd.sample(true,0.3,2)
res4: org.apache.spark.rdd.RDD[Int] = PartitionwiseSampledRDD[5] at sample at <console>:27

scala> rdd.sample(true,0.3,2).collect
res6: Array[Int] = Array(1, 2, 3, 3, 4, 13, 13, 14, 15, 16, 17)

scala> rdd.sample(false,0.3,2).collect
res7: Array[Int] = Array(1, 3, 12, 15, 16, 19, 20)

2.3.1.10 distinct([numTasks])) 案例

1. 做用：對源 RDD 進行去重後返回一個新的 RDD。默認狀況下，只有 8 個並行任務來操

做，可是能夠傳入一個可選的 numTasks 參數改變它。

2. 需求：建立一個 RDD，使用 distinct()對其去重。

（1）建立一個 RDD

scala> val distinctRdd = sc.parallelize(List(1,2,1,5,2,9,6,1))
distinctRdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[34] at parallelize at <console>:24

（2）對 RDD 進行去重（不指定並行度）

scala> val unionRDD = distinctRdd.distinct()
unionRDD: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[37] at distinct at <console>:26

（3）打印去重後生成的新 RDD

scala> unionRDD.collect()
res20: Array[Int] = Array(1, 9, 5, 6, 2)

（4）對 RDD（指定並行度爲 2）

scala> val unionRDD = distinctRdd.distinct(2)
unionRDD: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[40] at distinct at <console>:26

（5）打印去重後生成的新 RDD

scala> unionRDD.collect()
res21: Array[Int] = Array(6, 2, 1, 9, 5)

測試：數據結構

scala> val rdd = sc.parallelize(Array(1,1,2))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[8] at parallelize at <console>:24

scala> rdd.distinct.collect
res9: Array[Int] = Array(1, 2)

2.3.1.11 coalesce(numPartitions) 案例

1. 做用：縮減分區數，用於大數據集過濾後，提升小數據集的執行效率。

2. 需求：建立一個 4 個分區的 RDD，對其縮減分區

（1）建立一個 RDD

scala> val rdd = sc.parallelize(1 to 16,4)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[54] at parallelize at <console>:24

（2）查看 RDD 的分區數

scala> rdd.partitions.size
res20: Int = 4

（3）對 RDD 從新分區

scala> val coalesceRDD = rdd.coalesce(3)
coalesceRDD: org.apache.spark.rdd.RDD[Int] = CoalescedRDD[55] at coalesce at <console>:26

（4）查看新 RDD 的分區數

scala> coalesceRDD.partitions.size
res21: Int = 3

2.3.1.12 repartition(numPartitions) 案例

1. 做用：根據分區數，從新經過網絡隨機洗牌全部數據。

2. 需求：建立一個 4 個分區的 RDD，對其從新分區

（1）建立一個 RDD

scala> val rdd = sc.parallelize(1 to 16,4)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[56] at parallelize at <console>:24

（2）查看 RDD 的分區數

scala> rdd.partitions.size
res22: Int = 4

（3）對 RDD 從新分區

scala> val rerdd = rdd.repartition(2)
rerdd: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[60] at repartition at <console>:26

（4）查看新 RDD 的分區數

scala> rerdd.partitions.size
res23: Int = 2

測試：app

scala> rdd.partitions.size
res14: Int = 4

scala> rdd.repartition(2).partitions.size
res15: Int = 2

2.3.1.13 coalesce 和 repartition 的區別

1. coalesce 從新分區，能夠選擇是否進行 shuffle 過程。由參數 shuffle: Boolean = false/true決定。

2. repartition 其實是調用的 coalesce，進行 shuffle。源碼以下：

def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {
　　coalesce(numPartitions, shuffle = true)
}

2.3.1.14 sortBy(func,[ascending], [numTasks]) 案例

1. 做用；使用 func 先對數據進行處理，按照處理後的數據比較結果排序，默認爲正序。

2. 需求：建立一個 RDD，按照不一樣的規則進行排序

（1）建立一個 RDD

scala> val rdd = sc.parallelize(List(2,1,3,4))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[21] at parallelize at <console>:24

（2）按照自身大小排序

scala> rdd.sortBy(x => x).collect()
res11: Array[Int] = Array(1, 2, 3, 4)

（3）按照與 3 餘數的大小排序

scala> rdd.sortBy(x => x%3).collect()
res12: Array[Int] = Array(3, 4, 1, 2)

測試：

scala> rdd.sortBy(_*2)
res16: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[24] at sortBy at <console>:28

scala> rdd.sortBy(_*2).collect
res17: Array[Int] = Array(1, 1, 2)

scala> rdd.sortBy(_*2,false).collect
res18: Array[Int] = Array(2, 1, 1)

2.3.1.15 repartitionAndSortWithinPartitions(partitioner) 案例

repartitionAndSortWithinPartitions 函數是 repartition 函數的變量，與 repartition 函數不一樣的是

repartitionAndSortWithin在給定的 partitioner 內部進行排序，性能比 repartition 要高。

2.3.1.15 pipe(command, [envVars]) 案例

1. 做用：管道，針對每一個分區，都執行一個 shell 腳本，返回輸出的 RDD。

注意：腳本須要放在 Worker 節點能夠訪問到的位置

2. 需求：編寫一個腳本，使用管道將腳本做用於 RDD 上。

（1）編寫一個腳本

Shell 腳本：

#!/bin/sh
echo "AA"
while read LINE; do
　　echo ">>>"${LINE}
done

[lxl@hadoop102 spark]$ chmod 777 pipe.sh

（2）建立一個只有一個分區的 RDD

scala> val rdd = sc.parallelize(List("hi","Hello","how","are","you"),1)
rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[50] at parallelize at <console>:24

（3）將腳本做用該 RDD 並打印

scala> rdd.pipe("/opt/module/spark/pipe.sh").collect()
res18: Array[String] = Array(AA, >>>hi, >>>Hello, >>>how, >>>are, >>>you)

（4）建立一個有兩個分區的 RDD

scala> val rdd = sc.parallelize(List("hi","Hello","how","are","you"),2)
rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[52] at parallelize at <console>:24

（5）將腳本做用該 RDD 並打印

scala> rdd.pipe("/opt/module/spark/pipe.sh").collect()
res19: Array[String] = Array(AA, >>>hi, >>>Hello, AA, >>>how, >>>are, >>>you)

測試：

scala> val rdd = sc.parallelize(Array(1,1,2))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[8] at parallelize at <console>:24

scala> rdd.pipe("/opt/module/spark/pipe.sh").collect
res26: Array[String] = Array(AA, AA, >>>1, AA, >>>1, AA, >>>2)

　scala> sc.parallelize(1 to 3,1).pipe("/opt/module/spark/pipe.sh").collect
　res27: Array[String] = Array(AA, >>>1, >>>2, >>>3)

2.3.2 雙 Value 類型交互

2.3.2.1 union(otherDataset) 案例

1. 做用：對源 RDD 和參數 RDD 求並集後返回一個新的 RDD

2. 需求：建立兩個 RDD，求並集

（1）建立第一個 RDD

scala> val rdd1 = sc.parallelize(1 to 5)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[23] at parallelize at <console>:24

（2）建立第二個 RDD

scala> val rdd2 = sc.parallelize(5 to 10)
rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[24] at parallelize at <console>:24

（3）計算兩個 RDD 的並集

scala> val rdd3 = rdd1.union(rdd2)
rdd3: org.apache.spark.rdd.RDD[Int] = UnionRDD[25] at union at <console>:28

（4）打印並集結果

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

測試：

scala> val rdd = sc.parallelize(Array(1,1,2))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[8] at parallelize at <console>:24

scala> rdd.union(sc.parallelize(2 to 5)).collect
res19: Array[Int] = Array(1, 1, 2, 2, 3, 4, 5)

2.3.2.2 subtract (otherDataset) 案例

1. 做用：計算差的一種函數，去除兩個 RDD 中相同的元素，不一樣的 RDD 將保留下來

2. 需求：建立兩個 RDD，求第一個 RDD 與第二個 RDD 的差集

（1）建立第一個 RDD

scala> val rdd = sc.parallelize(3 to 8)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[70] at parallelize at <console>:24

（2）建立第二個 RDD

scala> val rdd1 = sc.parallelize(1 to 5)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[71] at parallelize at <console>:24

（3）計算第一個 RDD 與第二個 RDD 的差集並打印

scala> rdd.subtract(rdd1).collect()
res27: Array[Int] = Array(8, 6, 7)

測試：

scala> val rdd = sc.parallelize(Array(1,1,2))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[8] at parallelize at <console>:24

scala> rdd.subtract(sc.parallelize(2 to 5)).collect
res21: Array[Int] = Array(1, 1)

2.3.2.3 intersection(otherDataset) 案例

1. 做用：對源 RDD 和參數 RDD 求交集後返回一個新的 RDD

2. 需求：建立兩個 RDD，求兩個 RDD 的交集

（1）建立第一個 RDD

scala> val rdd1 = sc.parallelize(1 to 7)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[26] at parallelize at <console>:24

（2）建立第二個 RDD

scala> val rdd2 = sc.parallelize(5 to 10)
rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[27] at parallelize at <console>:24

（3）計算兩個 RDD 的交集

scala> val rdd3 = rdd1.intersection(rdd2)
rdd3: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[33] at intersection at <console>:28

（4）打印計算結果

scala> rdd3.collect()
res19: Array[Int] = Array(5, 6, 7)

測試：

scala> val rdd = sc.parallelize(Array(1,1,2))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[8] at parallelize at <console>:24

scala> rdd.intersection(sc.parallelize(2 to 5)).collect
res22: Array[Int] = Array(2)

2.3.2.4 cartesian(otherDataset) 案例

1. 做用：笛卡爾積（儘可能避免使用）

2. 需求：建立兩個 RDD，計算兩個 RDD 的笛卡爾積

（1）建立第一個 RDD

scala> val rdd1 = sc.parallelize(1 to 3)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[47] at parallelize at <console>:24

（2）建立第二個 RDD

scala> val rdd2 = sc.parallelize(2 to 5)
rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[48] at parallelize at <console>:24

（3）計算兩個 RDD 的笛卡爾積並打印

scala> rdd1.cartesian(rdd2).collect()
res17: Array[(Int, Int)] = Array((1,2), (1,3), (1,4), (1,5), (2,2), (2,3), (2,4), (2,5), (3,2), (3,3), (3,4), (3,5))

測試：

scala> val rdd = sc.parallelize(Array(1,1,2))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[8] at parallelize at <console>:24

scala> rdd.cartesian(sc.parallelize(2 to 5)).collect
res23: Array[(Int, Int)] = Array((1,2), (1,3), (1,4), (1,5), (1,2), (1,3), (1,4), (1,5), (2,2), (2,3), (2,4), (2,5))

2.3.2.5 zip(otherDataset)案例

1. 做用：將兩個 RDD 組合成 Key/Value 形式的 RDD,這裏默認兩個 RDD 的 partition 數量

以及元素數量都相同，不然會拋出異常。

2. 需求：建立兩個 RDD，並將兩個 RDD 組合到一塊兒造成一個(k,v)RDD

（1）建立第一個 RDD

scala> val rdd1 = sc.parallelize(Array(1,2,3),3)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[1] at parallelize at <console>:24

（2）建立第二個 RDD（與 1 分區數相同）

scala> val rdd2 = sc.parallelize(Array("a","b","c"),3)
rdd2: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[2] at parallelize at <console>:24

（3）第一個 RDD 組合第二個 RDD 並打印

scala> rdd1.zip(rdd2).collect
res1: Array[(Int, String)] = Array((1,a), (2,b), (3,c))

（4）第二個 RDD 組合第一個 RDD 並打印

scala> rdd2.zip(rdd1).collect
res2: Array[(String, Int)] = Array((a,1), (b,2), (c,3))

（5）建立第三個 RDD（與 1,2 分區數不一樣）

scala> val rdd3 = sc.parallelize(Array("a","b","c"),2)
rdd3: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[5] at parallelize at <console>:24

（6）第一個 RDD 組合第三個 RDD 並打印

scala> rdd1.zip(rdd3).collect
java.lang.IllegalArgumentException: Can't zip RDDs with unequal numbers of partitions: List(3, 2)
 at org.apache.spark.rdd.ZippedPartitionsBaseRDD.getPartitions(ZippedPartitionsRDD.scala:57)
 at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:252)
 at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:250)
 at scala.Option.getOrElse(Option.scala:121)
 at org.apache.spark.rdd.RDD.partitions(RDD.scala:250)
 at org.apache.spark.SparkContext.runJob(SparkContext.scala:1965)
 at org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:936)
 at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
 at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
 at org.apache.spark.rdd.RDD.withScope(RDD.scala:362)
 at org.apache.spark.rdd.RDD.collect(RDD.scala:935)
 ... 48 elided

2.3.3 Key-Value 類型

2.3.3.1 partitionBy 案例

1. 做用：對 pairRDD 進行分區操做，若是原有的 partionRDD 和現有的 partionRDD 是一致

的話就不進行分區，不然會生成 ShuffleRDD，即會產生 shuffle 過程。

2. 需求：建立一個 4 個分區的 RDD，對其從新分區

（1）建立一個 RDD

scala> val rdd = sc.parallelize(Array((1,"aaa"),(2,"bbb"),(3,"ccc"),(4,"ddd")),4)
rdd: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[44] at parallelize at <console>:24

（2）查看 RDD 的分區數

scala> rdd.partitions.size
res24: Int = 4

（3）對 RDD 從新分區

scala> var rdd2 = rdd.partitionBy(new org.apache.spark.HashPartitioner(2))
rdd2: org.apache.spark.rdd.RDD[(Int, String)] = ShuffledRDD[45] at partitionBy at <console>:26

（4）查看新 RDD 的分區數

scala> rdd2.partitions.size
res25: Int = 2

測試：

scala> val rdd = sc.parallelize(1 to 20)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[4] at parallelize at <console>:24

scala> import org.apache.spark._

scala> rdd.map((_,1)).partitionBy(new org.apache.spark.HashPartitioner(4)) //導包後紅色字段代碼能夠省略
res12: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[13] at partitionBy at <console>:28

scala> rdd.map((_,1)).partitionBy(new org.apache.spark.HashPartitioner(4)).partitions.size
res13: Int = 4

2.3.3.2 reduceByKey(func, [numTasks]) 案例

1. 在一個(K,V)的 RDD 上調用，返回一個(K,V)的 RDD，使用指定的 reduce 函數，將相同

key 的值聚合到一塊兒，reduce 任務的個數能夠經過第二個可選的參數來設置。

2. 需求：建立一個 pairRDD，計算相同 key 對應值的相加結果

（1）建立一個 pairRDD

scala> val rdd = sc.parallelize(List(("female",1),("male",5),("female",5),("male",2)))
rdd: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[46] at parallelize at <console>:24

（2）計算相同 key 對應值的相加結果

scala> val reduce = rdd.reduceByKey((x,y) => x+y)
reduce: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[47] at reduceByKey at <console>:26

（3）打印結果

scala> reduce.collect()
res29: Array[(String, Int)] = Array((female,6), (male,7))

測試：

scala> val rdd = sc.parallelize(Array(1,1,2))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:24

scala> rdd.map((_,1)).reduceByKey(_+_)
res0: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[2] at reduceByKey at <console>:27

scala> rdd.map((_,1)).reduceByKey(_+_).collect
res1: Array[(Int, Int)] = Array((1,2), (2,1))                                   

scala> rdd.map((_,1)).reduceByKey((x,y) => x+y).collect
res2: Array[(Int, Int)] = Array((1,2), (2,1))

2.3.3.3 groupByKey 案例

1. 做用：groupByKey 也是對每一個 key 進行操做，但只生成一個 seq。

2. 需求：建立一個 pairRDD，將相同 key 對應值聚合到一個 seq 中，並計算相同 key 對應

值的相加結果。

（1）建立一個 pairRDD

scala> val words = Array("one", "two", "two", "three", "three", "three")
words: Array[String] = Array(one, two, two, three, three, three)
scala> val wordPairsRDD = sc.parallelize(words).map(word => (word, 1))
wordPairsRDD: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[4] at map at <console>:26

（2）將相同 key 對應值聚合到一個 Seq 中

scala> val group = wordPairsRDD.groupByKey()
group: org.apache.spark.rdd.RDD[(String, Iterable[Int])] = ShuffledRDD[5] at groupByKey at <console>:28

（3）打印結果

scala> group.collect()
res1: Array[(String, Iterable[Int])] = Array((two,CompactBuffer(1, 1)), (one,CompactBuffer(1)), (three,CompactBuffer(1, 1, 1)))

（4）計算相同 key 對應值的相加結果

scala> group.map(t => (t._1, t._2.sum))
res2: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[6] at map at <console>:31

（5）打印結果

scala> res2.collect()
res3: Array[(String, Int)] = Array((two,2), (one,1), (three,3))

測試：

scala> val rdd = sc.parallelize(Array(1,1,2))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:24                          

scala> rdd.map((_,1)).groupByKey.collect
res3: Array[(Int, Iterable[Int])] = Array((1,CompactBuffer(1, 1)), (2,CompactBuffer(1)))

2.3.3.4 reduceByKey 和 groupByKey 的區別

1. reduceByKey：按照 key 進行聚合，在 shuffle 以前有 combine（預聚合）操做，返回結果

是 RDD[k,v]。

2. groupByKey：按照 key 進行分組，直接進行 shuffle。

3. 開發指導：reduceByKey 比 groupByKey，建議使用。可是須要注意是否會影響業務邏

輯。

2.3.3.5 aggregateByKey 案例

參數：
(
zeroValue:U,[partitioner: Partitioner] //一個分區，一個 key
)
(
seqOp: (U, V) => U, //分區內遇到
combOp: (U, U) => U
)

1. 做用：在 kv 對的 RDD 中，，按 key 將 value 進行分組合並，合併時，將每一個 value 和初

始值做爲 seq 函數的參數，進行計算，返回的結果做爲一個新的 kv 對，而後再將結果按照

key 進行合併，最後將每一個分組的 value 傳遞給 combine 函數進行計算（先將前兩個 value

進行計算，將返回結果和下一個 value 傳給 combine 函數，以此類推），將 key 與計算結果

做爲一個新的 kv 對輸出。

2. 參數描述：

（1）zeroValue：給每個分區中的每個 key 一個初始值；

（2）seqOp：函數用於在每個分區中用初始值逐步迭代 value；

（3）combOp：函數用於合併每一個分區中的結果。

3. 需求：建立一個 pairRDD，取出每一個分區相同 key 對應值的最大值，而後相加

4. 需求分析

aggregateByKey()案例解析

（1）建立一個 pairRDD

scala> val rdd = sc.parallelize(List(("a",3),("a",2),("c",4),("b",3),("c",6),("c",8)),2)
rdd: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[0] at parallelize at <console>:24

（2）取出每一個分區相同 key 對應值的最大值，而後相加

scala> val agg = rdd.aggregateByKey(0)(math.max(_,_),_+_)
agg: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[1] at aggregateByKey at <console>:26

（3）打印結果

scala> agg.collect()
res0: Array[(String, Int)] = Array((b,3), (a,3), (c,12))

測試：

scala> val scores = Array(("Fred", 88), ("Fred", 95), ("Fred", 91), ("Wilma", 93), ("Wilma", 95), ("Wilma", 98))
scores: Array[(String, Int)] = Array((Fred,88), (Fred,95), (Fred,91), (Wilma,93), (Wilma,95), (Wilma,98))

scala> val input = sc.parallelize(scores)
input: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[9] at parallelize at <console>:26

scala> input.aggregateByKey((0,0))((u,v) => (u._1+v,u._2+1),(u1,u2) => (u1._1+u2._1,u1._2+u2._2))
res6: org.apache.spark.rdd.RDD[(String, (Int, Int))] = ShuffledRDD[11] at aggregateByKey at <console>:29

scala> input.aggregateByKey((0,0))((u,v) => (u._1+v,u._2+1),(u1,u2) => (u1._1+u2._1,u1._2+u2._2)).collect
res7: Array[(String, (Int, Int))] = Array((Wilma,(286,3)), (Fred,(274,3)))

2.3.3.6 foldByKey 案例

參數：(zeroValue: V)(func: (V, V) => V): RDD[(K, V)]

1. 做用：aggregateByKey 的簡化操做，seqop 和 combop 相同

2. 需求：建立一個 pairRDD，計算相同 key 對應值的相加結果

（1）建立一個 pairRDD

scala> val rdd = sc.parallelize(List((1,3),(1,2),(1,4),(2,3),(3,6),(3,8)),3)
rdd: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[91] at parallelize at <console>:24

（2）計算相同 key 對應值的相加結果

scala> val agg = rdd.foldByKey(0)(_+_)
agg: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[92] at foldByKey at <console>:26

（3）打印結果

scala> agg.collect()
res61: Array[(Int, Int)] = Array((3,14), (1,9), (2,3))

2.3.3.7 combineByKey[C] 案例

參數：
(
createCombiner: V => C, //分區內使用，第一次遇到某個 key 的時
mergeValue: (C, V) => C, //分區內使用，非第一次遇到某個 key 時
mergeCombiners: (C, C) => C //將多個分區的結果進行合併
)

1. 做用：針對相同 K，將 V 合併成一個集合。

2. 參數描述：

（1）createCombiner: combineByKey() 會遍歷分區中的全部元素，所以每一個元素的鍵要麼尚未遇到過，

要麼就和以前的某個元素的鍵相同。若是這是一個新的元素,combineByKey()會使用一個叫做

createCombiner()的函數來建立那個鍵對應的累加器的初始值

（2）mergeValue: 若是這是一個在處理當前分區以前已經遇到的鍵，它會使用 mergeValue()方法將該鍵的

累加器對應的當前值與這個新的值進行合併

（3）mergeCombiners: 因爲每一個分區都是獨立處理的，所以對於同一個鍵能夠有多個累加器。若是有兩

個或者更多的分區都有對應同一個鍵的累加器，就須要使用用戶提供的 mergeCombiners() 方法將各個分

區的結果進行合併。

3. 需求：建立一個 pairRDD，根據 key 計算每種 key 的均值。（先計算每一個 key 出現的次數

以及能夠對應值的總和，再相除獲得結果）

4. 需求分析：

combineByKey()案例分析

　　　　　　　　　　　　　　　　圖 2- combineByKey 案例分析

（1）建立一個 pairRDD

scala> val input = sc.parallelize(Array(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98)),2)
input: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[52] at parallelize at <console>:26

（2）將相同 key 對應的值相加，同時記錄該 key

scala> val combine = input.combineByKey((_,1),(acc:(Int,Int),v)=>(acc._1+v,acc._2+1),(acc1:(Int,Int),acc2:(Int,Int)) => (acc1._1+acc2._1,acc1._2+acc2._2))
combine: org.apache.spark.rdd.RDD[(String, (Int, Int))] = ShuffledRDD[5] at combineByKey at <console>:28

（3）打印合並後的結果

scala> combine.collect
res5: Array[(String, (Int, Int))] = Array((b,(286,3)), (a,(274,3)))

（4）計算平均值

scala> val result = combine.map{case (key,value) => (key,value._1/value._2.toDouble)}
result: org.apache.spark.rdd.RDD[(String, Double)] = MapPartitionsRDD[54] at map at <console>:30

（5）打印結果

scala> result.collect()
res33: Array[(String, Double)] = Array((b,95.33333333333333), (a,91.33333333333333))

測試：

scala> val scores = Array(("Fred", 88), ("Fred", 95), ("Fred", 91), ("Wilma", 93), ("Wilma", 95), ("Wilma", 98))
scores: Array[(String, Int)] = Array((Fred,88), (Fred,95), (Fred,91), (Wilma,93), (Wilma,95), (Wilma,98))

scala> val input = sc.parallelize(scores)
input: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[9] at parallelize at <console>:26

scala> input.combineByKey(x => (x,1),(a,b) => (b+a._1,a._2+1),(c1,c2) => (c1._1+c2._1,c1._2+c2._2)).collect
<console>:29: error: missing parameter type
       input.combineByKey(x => (x,1),(a,b) => (b+a._1,a._2+1),(c1,c2) => (c1._1+c2._1,c1._2+c2._2)).collect
                                      ^
<console>:29: error: missing parameter type
       input.combineByKey(x => (x,1),(a,b) => (b+a._1,a._2+1),(c1,c2) => (c1._1+c2._1,c1._2+c2._2)).collect
                                                               ^
<console>:29: error: missing parameter type
       input.combineByKey(x => (x,1),(a,b) => (b+a._1,a._2+1),(c1,c2) => (c1._1+c2._1,c1._2+c2._2)).collect
                                                                  ^

scala> input.combineByKey(x => (x,1),(a:(Int,Int),b) => (b+a._1,a._2+1),(c1:(Int,Int),c2:(Int,Int)) => (c1._1+c2._1,c1._2+c2._2)).collect
res5: Array[(String, (Int, Int))] = Array((Wilma,(286,3)), (Fred,(274,3)))

2.3.3.8 sortByKey([ascending], [numTasks]) 案例

1. 做用：在一個(K,V)的 RDD 上調用，K 必須實現 Ordered 接口，返回一個按照 key 進行

排序的(K,V)的 RDD

2. 需求：建立一個 pairRDD，按照 key 的正序和倒序進行排序

（1）建立一個 pairRDD

scala> val rdd = sc.parallelize(Array((3,"aa"),(6,"cc"),(2,"bb"),(1,"dd")))
rdd: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[14] at parallelize at <console>:24

（2）按照 key 的正序

scala> rdd.sortByKey(true).collect()
res9: Array[(Int, String)] = Array((1,dd), (2,bb), (3,aa), (6,cc))

（3）按照 key 的倒序

scala> rdd.sortByKey(false).collect()
res10: Array[(Int, String)] = Array((6,cc), (3,aa), (2,bb), (1,dd))

測試：

scala> rdd.collect
res12: Array[Int] = Array(1, 1, 2)

scala> rdd.map((_,1))
res9: org.apache.spark.rdd.RDD[(Int, Int)] = MapPartitionsRDD[13] at map at <console>:27

scala> rdd.map((_,1)).collect
res10: Array[(Int, Int)] = Array((1,1), (1,1), (2,1))

scala> rdd.map((_,1)).sortByKey(true).collect
res11: Array[(Int, Int)] = Array((1,1), (1,1), (2,1))

2.3.3.9 mapValues 案例

1. 針對於(K,V)形式的類型只對 V 進行操做

2. 需求：建立一個 pairRDD，並將 value 添加字符串"|||"

（1）建立一個 pairRDD

scala> val rdd3 = sc.parallelize(Array((1,"a"),(1,"d"),(2,"b"),(3,"c")))
rdd3: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[67] at parallelize at <console>:24

（2）對 value 添加字符串"|||"

scala> rdd3.mapValues(_+"|||").collect()
res26: Array[(Int, String)] = Array((1,a|||), (1,d|||), (2,b|||), (3,c|||))

測試：

scala> rdd.collect
res12: Array[Int] = Array(1, 1, 2)

scala> rdd.map((_,1)).mapValues(_*2).collect
res13: Array[(Int, Int)] = Array((1,2), (1,2), (2,2))

scala> rdd.map((_,1)).mapValues(_*2).collect  //也可使用模式匹配
res13: Array[(Int, Int)] = Array((1,2), (1,2), (2,2))

2.3.3.10 join(otherDataset, [numTasks]) 案例

1. 做用：在類型爲(K,V)和(K,W)的 RDD 上調用，返回一個相同 key 對應的全部元素對在

一塊兒的(K,(V,W))的 RDD（內鏈接？）

2. 需求：建立兩個 pairRDD，並將 key 相同的數據聚合到一個元組。

（1）建立第一個 pairRDD

scala> val rdd = sc.parallelize(Array((1,"a"),(2,"b"),(3,"c")))
rdd: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[32] at parallelize at <console>:24

（2）建立第二個 pairRDD

scala> val rdd1 = sc.parallelize(Array((1,4),(2,5),(3,6)))
rdd1: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[33] at parallelize at <console>:24

（3）join 操做並打印結果

scala> rdd.join(rdd1).collect()
res13: Array[(Int, (String, Int))] = Array((1,(a,4)), (2,(b,5)), (3,(c,6)))

測試：

scala> val rdd1 = sc.parallelize(1 to 10).map((_,1))
rdd1: org.apache.spark.rdd.RDD[(Int, Int)] = MapPartitionsRDD[57] at map at <console>:25

scala> val rdd2 = sc.parallelize(5 to 15).map((_,1))
rdd2: org.apache.spark.rdd.RDD[(Int, Int)] = MapPartitionsRDD[59] at map at <console>:25

scala> rdd1.join(rdd2).collect
res28: Array[(Int, (Int, Int))] = Array((8,(1,1)), (9,(1,1)), (5,(1,1)), (6,(1,1)), (10,(1,1)), (7,(1,1)))

2.3.3.11 cogroup(otherDataset, [numTasks]) 案例

1. 做用：在類型爲(K,V)和(K,W)的 RDD 上調用，返回一個(K,(Iterable,Iterable))類型的 RDD

2. 需求：建立兩個 pairRDD，並將 key 相同的數據聚合到一個迭代器。

（1）建立第一個 pairRDD

scala> val rdd = sc.parallelize(Array((1,"a"),(2,"b"),(3,"c")))
rdd: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[37] at parallelize at <console>:24

（2）建立第二個 pairRDD

scala> val rdd1 = sc.parallelize(Array((1,4),(2,5),(3,6)))
rdd1: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[38] at parallelize at <console>:24

（3）cogroup 兩個 RDD 並打印結果

scala> rdd.cogroup(rdd1).collect()
res14: Array[(Int, (Iterable[String], Iterable[Int]))] = 
Array((1,(CompactBuffer(a),CompactBuffer(4))), (2,(CompactBuffer(b),CompactBuffer(5))), (3,(CompactBuffer(c),CompactBuffer(6))))

測試：

scala> val rdd1 = sc.parallelize(Array(0,1,1,3,3)).map((_,1))
rdd1: org.apache.spark.rdd.RDD[(Int, Int)] = MapPartitionsRDD[76] at map at <console>:25

scala> val rdd2 = sc.parallelize(Array(0,2,1,4,3)).map((_,1))
rdd2: org.apache.spark.rdd.RDD[(Int, Int)] = MapPartitionsRDD[78] at map at <console>:25

scala> rdd1.cogroup(rdd2).collect
res34: Array[(Int, (Iterable[Int], Iterable[Int]))] = Array((4,(CompactBuffer(),CompactBuffer(1))), (0,(CompactBuffer(1),CompactBuffer(1))), (1,(CompactBuffer(1, 1),CompactBuffer(1))), (2,(CompactBuffer(),CompactBuffer(1))), (3,(CompactBuffer(1, 1),CompactBuffer(1))))

2.3.4 案例實操

1. 數據結構：時間戳，省份，城市，用戶，廣告，中間字段使用空格分割。

樣本以下：

1516609143867 6 7 64 16
1516609143869 9 4 75 18
1516609143869 1 7 87 12

2. 需求：統計出每個省份廣告被點擊次數的 TOP3

3. 實現過程：

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

//需求：統計出每個省份廣告被點擊次數的 TOP3
object Practice {
  def main(args: Array[String]): Unit = {

    //1.初始化 spark 配置信息並創建與 spark 的鏈接
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Test")
    val sc = new SparkContext(sparkConf)

    //2.讀取數據生成 RDD：TS，Province，City，User，AD
    val line = sc.textFile("E:\\IDEAWorkSpace\\SparkTest\\src\\main\\resources\\agent.log")

    //3.按照最小粒度聚合：((Province,AD),1)
    val provinceAdAndOne = line.map { x =>
      val fields: Array[String] = x.split(" ")
      ((fields(1), fields(3)), 1)
    }

    //4.計算每一個省中每一個廣告被點擊的總數：((Province,AD),sum)
    val provinceAdToSum = provinceAdAndOne.reduceByKey(_ + _)

    //5.將省份做爲 key，廣告加點擊數爲 value：(Province,(AD,sum))
    val provinceToAdSum = provinceAdToSum.map(x => (x._1._1, (x._1._2, x._2)))

    //6.將同一個省份的全部廣告進行聚合(Province,List((AD1,sum1),(AD2,sum2)...))
    val provinceGroup = provinceToAdSum.groupByKey()

    //7.對同一個省份全部廣告的集合進行排序並取前 3 條，排序規則爲廣告點擊總數
    val provinceAdTop3 = provinceGroup.mapValues { x =>
      x.toList.sortWith((x, y) => x._2 > y._2).take(3)
    }

    //8.將數據拉取到 Driver 端並打印
    provinceAdTop3.collect().foreach(println)

    //9.關閉與 spark 的鏈接
    sc.stop()
  }
}