Spark RDD-行动算子

2.4 Action

行动算子：触发运算，在 Executor 执行，如果想直接在 Driver 端看到结果可以使用 collect 和 foreach 都可以将数据拉取到 Driver 端。

2.4.1 reduce(func) 案例

1. 作用：通过 func 函数聚集 RDD 中的所有元素，先聚合分区内数据，再聚合分区间数据。

 

2. 需求：创建一个 RDD，将所有元素聚合得到结果

 

（1）创建一个 RDD[Int]

scala> val rdd1 = sc.makeRDD(1 to 10,2)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[85] at makeRDD at <console>:24

（2）聚合 RDD[Int]所有元素

scala> rdd1.reduce(_+_)
res50: Int = 55

（3）创建一个 RDD[String]

scala> val rdd2 = sc.makeRDD(Array(("a",1),("a",3),("c",3),("d",5)))
rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[86] at makeRDD at <console>:24

（4）聚合 RDD[String]所有数据

scala> rdd2.reduce((x,y)=>(x._1 + y._1,x._2 + y._2))
res51: (String, Int) = (adca,12)

测试：

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

scala> rdd.reduce(_+_)
res17: Int = 15

2.4.2 collect() 案例

从 Executor 端拉取数据到 Driver 端操作：一般用在测试环境中。

1. 作用：在驱动程序中，以数组的形式返回数据集的所有元素。

 

2. 需求：创建一个 RDD，并将 RDD 内容收集到 Driver 端打印

 

（1）创建一个 RDD

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

（2）将结果收集到 Driver 端

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

2.4.3 count() 案例

1. 作用：返回 RDD 中元素的个数

 

2. 需求：创建一个 RDD，统计该 RDD 的条数

 

（1）创建一个 RDD

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

（2）统计该 RDD 的条数

scala> rdd.count
res1: Long = 10

2.4.4 first() 案例

1. 作用：返回 RDD 中的第一个元素

 

2. 需求：创建一个 RDD，返回该 RDD 中的第一个元素

 

（1）创建一个 RDD

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

（2）统计该 RDD 的条数

scala> rdd.first
res2: Int = 1

2.4.5 take(n) 案例

1. 作用：返回一个由 RDD 的前 n 个元素组成的数组

 

2. 需求：创建一个 RDD，统计该 RDD 的条数

 

（1）创建一个 RDD

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

（2）统计该 RDD 的条数

scala> rdd.take(3)
res10: Array[Int] = Array(2, 5, 4)

 2.4.6 takeSample(WithReplacement,num,[seed])

返回一个数组，该数组由从数据集中随机采样的 num 个元素组成，可以选择是否随机数

替换不足的部分，seed 用于指定随机数生成器种子。

2.4.7 takeOrdered(n) 案例

1. 作用：返回该 RDD 排序后的前 n 个元素组成的数组

--相当于先执行 sortBy() 再执行 take()

 

2. 需求：创建一个 RDD，统计该 RDD 的条数

 

（1）创建一个 RDD

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

（2）统计该 RDD 的条数

scala> rdd.takeOrdered(3)
res18: Array[Int] = Array(2, 3, 4)

2.4.8 aggregate 案例

1. 参数：(zeroValue: U)(seqOp: (U, T) ⇒ U, combOp: (U, U) ⇒ U)

 

2. 作用：aggregate 函数将每个分区里面的元素通过 seqOp 和初始值进行聚合，然后用

combine 函数将每个分区的结果和初始值(zeroValue)进行 combine 操作。这个函数最终返回

的类型不需要和 RDD 中元素类型一致。

 

3. 需求：创建一个 RDD，将所有元素相加得到结果

 

（1）创建一个 RDD

scala> var rdd1 = sc.makeRDD(1 to 10,2)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[88] at makeRDD at <console>:24

（2）将该 RDD 所有元素相加得到结果

scala> rdd.aggregate(0)(_+_,_+_)
res22: Int = 55

2.4.9 fold(num)(func) 案例

1. 作用：折叠操作，aggregate 的简化操作，seqop 和 combop 一样。

 

2. 需求：创建一个 RDD，将所有元素相加得到结果

 

（1）创建一个 RDD

scala> var rdd1 = sc.makeRDD(1 to 10,2)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[88] at makeRDD at <console>:24

（2）将该 RDD 所有元素相加得到结果

scala> rdd.fold(0)(_+_)
res24: Int = 55

2.4.10 saveAsTextFile(path)

作用：将数据集的元素以 textfile 的形式保存到 HDFS 文件系统或者其他支持的文件系统，

对于每个元素，Spark 将会调用 toString 方法，将它装换为文件中的文本

开发中常用的读入（textFile）和 写出（saveAsTextFile）

 

 

2.4.11 saveAsSequenceFile(path)

作用：将数据集中的元素以 Hadoop sequencefile 的格式保存到指定的目录下，可以使

HDFS 或者其他 Hadoop 支持的文件系统。

 

 

2.4.12 saveAsObjectFile(path)

作用：用于将 RDD 中的元素序列化成对象，存储到文件中。

 

 

2.4.13 countByKey() 案例

1. 作用：针对(K,V)类型的 RDD，返回一个(K,Int)的 map，表示每一个 key 对应的元素个数。

 

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[95] at parallelize at <console>:24

（2）统计每种 key 的个数

scala> rdd.countByKey
res63: scala.collection.Map[Int,Long] = Map(3 -> 2, 1 -> 3, 2 -> 1)

测试：

scala> sc.parallelize(Array(1,2,1,2,3,3,3)).map((_,1)).countByKey()
res19: scala.collection.Map[Int,Long] = Map(1 -> 2, 2 -> 2, 3 -> 3) 

在开发中可以搭配 Sample 查看数据是否倾斜

2.4.14 foreach(func) 案例

1. 作用：在数据集的每一个元素上，运行函数 func 进行更新。

　　源码：

  def foreach(f: T => Unit): Unit = withScope {
    val cleanF = sc.clean(f)
    sc.runJob(this, (iter: Iterator[T]) => iter.foreach(cleanF))
  }

 

2. 需求：创建一个 RDD，对每个元素进行打印

 

（1）创建一个 RDD

scala> var rdd = sc.makeRDD(1 to 5,2)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[107] at makeRDD at <console>:24

（2）对该 RDD 每个元素进行打印

scala> rdd.foreach(println(_))
3
4
5
1
2

 测试：

scala> rdd.foreach(println)
2
1
4
5
3