Spark RDD-行动算子

小鱼儿 2024-04-20 09:10 100阅读 0赞

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

![1604514-20190707032642467-261499854.png][]

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

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

[1604514-20190707032642467-261499854.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/76ce0be31baf4d2f9e674b7d2dddf6b3.png