Spark RDD-转化算子

迷南。 2024-04-20 09:09 80阅读 0赞

# **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 创建 ###

后面详解

## 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)

测试：

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）使每个元素跟所在分区形成一个元组组成一个新的 RDD

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))

测试：

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

测试：

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()案例解析

![1604514-20190706210006473-676291474.png][]

（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))

测试：

#### 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()案例分析

![1604514-20190706211150853-14132020.png][]

图 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. 数据结构：时间戳，省份，城市，用户，广告，中间字段使用空格分割。

![1604514-20190707020904556-727060913.png][]

样本如下：

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()
      }
    }

####  ####

转载于:https://www.cnblogs.com/LXL616/p/11142909.html

[1604514-20190706210006473-676291474.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/f53f1708bfe64e848e5394f5f1cbfe88.png
[1604514-20190706211150853-14132020.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/30ec6a1260bd43499a97893b4b96f3a0.png
[1604514-20190707020904556-727060913.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/d3c07ec201ce48ba866bdce91d43f315.png