五、RDD 行动算子 - 《SparkCore》

// TODO 行动算子<br />    // 所谓的行动算子，其实就是触发作业(Job)执行的方法<br />    // 底层代码调用的是环境对象的runJob方法<br />    // 底层代码中会创建ActiveJob，并提交执行。<br />1)** reduce**     ➢ 函数签名    def reduce(f: (T, T) => T): T <br />➢ 函数说明     聚集 RDD 中的所有元素，先聚合分区内数据，再聚合分区间数据<br />**2) collect     **➢ 函数签名     def collect(): Array[T] <br />➢ 函数说明     在驱动程序中，以数组 Array 的形式返回数据集的所有元素   rdd.collect().foreach(println)<br />**3) count     **➢ 函数签名     def count(): Long <br />➢ 函数说明     返回 RDD 中元素的个数<br />**4) first     **➢ 函数签名     def first(): T <br />➢ 函数说明    返回 RDD 中的第一个元素<br />**5) take     **➢ 函数签名     def take(num: Int): Array[T] <br />➢ 函数说明     返回一个由 RDD 的前 n 个元素组成的数组<br />**6) takeOrdered     **➢ 函数签名     def takeOrdered(num: Int)(implicit ord: Ordering[T]): Array[T] <br />➢ 函数说明     返回该 RDD 排序后的前 n 个元素组成的数组<br />**7) aggregate     **➢ 函数签名     def aggregate[U: ClassTag](zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U): U <br />➢ 函数说明     分区的数据通过初始值和分区内的数据进行聚合，然后再和初始值进行分区间的数据聚合

val rdd = sc.makeRDD(List(1,2,3,4),2)
// TODO - 行动算子
//10 + 13 + 17 = 40
// aggregateByKey : 初始值只会参与分区内计算
// aggregate : 初始值会参与分区内计算,并且和参与分区间计算
//val result = rdd.aggregate(10)(_+_, _+_)  //10 + 13 + 17 = 40
val result = rdd.fold(10)(_+_)   // 10 + 13 + 17 = 40
println(result)

8) fold ➢ 函数签名 def fold(zeroValue: T)(op: (T, T) => T): T
➢ 函数说明折叠操作，aggregate 的简化版操作
9) countByKey ➢ 函数签名 def countByKey(): Map[K, Long]
➢ 函数说明统计每种 key 的个数

val rdd: RDD[(Int, String)] = sc.makeRDD(List(
  (1, "a"), (1, "a"), (1, "a"), (2, "b"), (3, "c"), (3, "c")))
// 统计每种 key 的个数
val result: collection.Map[Int, Long] = rdd.countByKey()

10) save 相关算子 ➢ 函数说明将数据保存到不同格式的文件中
➢ 函数签名
def saveAsTextFile(path: String): Unit
def saveAsObjectFile(path: String): Unit
def saveAsSequenceFile(
path: String,
codec: Option[Class[_ <: CompressionCodec]] = None): Unit

// 保存成 Text 文件
rdd.saveAsTextFile("output")
// 序列化成对象保存到文件
rdd.saveAsObjectFile("output1")
// 保存成 Sequencefile 文件
rdd.map((_,1)).saveAsSequenceFile("output2")

11) foreach ➢ 函数说明分布式遍历 RDD 中的每一个元素，调用指定函数
➢ 函数签名
def foreach(f: T => Unit): Unit = withScope {
val cleanF = sc.clean(f)
sc.runJob(this, (iter: Iterator[T]) => iter.foreach(cleanF)) }

val rdd = sc.makeRDD(List(1,2,3,4))
    // TODO 行动算子
    // reduce
    val i: Int = rdd.reduce(_+_)
    println(i)
    // collect : 方法会将不同分区的数据按照分区顺序采集到Driver端内存中，形成数组
//    val ints: Array[Int] = rdd.collect()
//    println(ints.mkString(","))
    // count : 数据源中数据的个数
    val cnt = rdd.count()
    println(cnt)
    // first : 获取数据源中数据的第一个
    val first = rdd.first()
    println(first)
    // take : 获取N个数据
    val ints: Array[Int] = rdd.take(3)
    println(ints.mkString(","))
    // takeOrdered : 数据排序后，取N个数据
    val rdd1 = sc.makeRDD(List(4,2,3,1))
    val ints1: Array[Int] = rdd1.takeOrdered(3)
    println(ints1.mkString(","))