【Spark Core】行动算子

• 0.行动算子概述
• 1.reduce
• 2.collect
• 3. count
• 4.first
• 5.take
• 6.takeOrdered
• 7.aggregate
• 8.fold
• 9. countByKey
• 10.save
• 11 foreach
• 12.算子的本质

• 行动算子触发任务执行，sc.runjob()，创建activeJob，提交并执行。
• 行动算子直接出结果，而不是RDD

1.函数签名

def reduce(f: (T, T) => T): T

2.函数说明
聚集RDD中的所有元素，先聚合分区内数据，再聚合分区间数据

package _02_action

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

object _01_reduce {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("mapPartitions").setMaster("local[2]")
    val sc = new SparkContext(conf)
    val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
    // 聚合数据
    val reduceResult: Int = rdd.reduce(_+_)
    println(reduceResult) //10
  }
}

1.函数签名

def collect(): Array[T]

• 以数组Array的形式收集RDD的所有元素到驱动程序（Driver）中
• 以分区为单位，从0号分区将数据按顺序收集采集到Driver

package _02_action

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

object _02_collect {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("mapPartitions").setMaster("local[2]")
    val sc = new SparkContext(conf)

    val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))

    // 收集数据到Driver
    rdd.collect().foreach(println)
    //1 2 3 4
  }
}

1.函数签名

def count(): Long

2.函数说明
返回RDD中元素的个数

val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
// 返回RDD中元素的个数
val countResult: Long = rdd.count()

1.函数签名

def first(): T

2.函数说明
返回RDD中的第一个元素

val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))

// 返回RDD中元素的个数
val firstResult: Int = rdd.first()
println(firstResult)

1.函数签名

def take(num: Int): Array[T]

2.函数说明
返回一个由RDD的前n个元素组成的数组

vval rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))

// 返回RDD中元素的个数
val takeResult: Array[Int] = rdd.take(2)
println(takeResult.mkString(","))

1.函数签名

def takeOrdered(num: Int)(implicit ord: Ordering[T]): Array[T]

2.函数说明
返回该RDD排序后的前n个元素组成的数组

val rdd: RDD[Int] = sc.makeRDD(List(1,3,2,4))

// 返回RDD中元素的个数
val result: Array[Int] = rdd.takeOrdered(2) //1 2

1.函数签名

def aggregate[U: ClassTag](zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U): U

2.函数说明

• 分区的数据通过初始值和分区内的数据进行聚合，然后再和初始值进行分区间的数据聚合
• 哪个分区先计算完谁在前面
• 初始值只参与1次分区间计算,多个分区只参与一次
• 聚合结果的数据类型和zeroValue的类型是一样的

package _02_action

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

object _07_aggregate {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("mapPartitions").setMaster("local[2]")
    val sc = new SparkContext(conf)

    val list1 = List(1,2,3,4,5,6,7,8,9,10)


    val rdd1: RDD[Int] = sc.makeRDD(list1,4)
    val value = rdd1.mapPartitionsWithIndex((index, iter) => {
      iter.map(x => {
        (index, x)
      })
    })
    value.collect().foreach(println)
    //(0,1)
    //(0,2)
    //(1,3)
    //(1,4)
    //(1,5)
    //(2,6)
    //(2,7)
    //(3,8)
    //(3,9)
    //(3,10)

    val str: String = rdd1.aggregate("+")(_ + _, _ + _)
    println(str)
    //++345+12+8910+67

    //todo aggregate
    // zeroValue : 参与一次分组内计算，做初始值
    //             参与一次分组间计算，做初始值

    //todo aggregateByKey 是行动算子
    // zeroValue: 只参与一次分组内计算，做初始值
  }
}

1.函数签名

def fold(zeroValue: T)(op: (T, T) => T): T

2.函数说明
aggregate的简化版操作

val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))

val foldResult: Int = rdd.fold(0)(_+_)

1.函数签名

def countByKey(): Map[K, Long]

package _02_action

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

object _09_countByKey {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("kv").setMaster("local[1]")
    val sparkContext = new SparkContext(conf)
    val dataRDD1 = sparkContext.makeRDD(List(("a",1),("b",2),("c",3),("a",3),("c",3)))

    val nums: collection.Map[String, Long] = dataRDD1.countByKey()


    dataRDD1.countByValue().foreach(println)
    //((b,2),1)
    //((a,3),1)
    //((c,3),2)  相同的元素才算一个
    //((a,1),1)

    nums.foreach(println)
    //(a,2)
    //(b,1)
    //(c,2)

  }
}

将rdd中数据输出到外部

package _02_action

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

object _10_save {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("kv").setMaster("local[1]")
    val sparkContext = new SparkContext(conf)
    val rdd = sparkContext.makeRDD(List(("a",1),("b",2),("c",3),("a",3),("c",3)))

    // 保存成Text文件
    rdd.saveAsTextFile("output")

    // 序列化成对象保存到文件
    rdd.saveAsObjectFile("output1")

    // 保存成Sequencefile文件
   //  rdd.map((_,1)).saveAsSequenceFile("output2")
  }
}

1.函数签名

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

2.函数说明
分布式遍历RDD中的每一个元素，调用指定函数

val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))

// 1.收集后打印
rdd.map(num=>num).collect().foreach(println)

println("****************")

// 2.分布式打印
rdd.foreach(println)

• collect先采集到Driver，这个foreach是内存中的集合的遍历方法
• foreach将数据发给不同的executor端，进行打印，是分布式打印
  从上面的打印结果可以看出来，收集后，由于是按照分区序号收集的，因此是按顺序打印的；而分布式打印，是并行的，哪个分区在前是不一定的
  （1）collect后打印：

（2）直接调用foreach算子打印：

• 算子是RDD的方法，和Scala集合对象的方法不同
• 集合对象的方法都是在同一个节点的内存中完成的
• RDD的算子可以将计算逻辑发送到Executor端（分布式节点）内存执行
• 为了区分不同的处理效果，所以将RDD的方法称之为算子
• RDD的算子外部的操作都是在Driver端执行的，而方法内部的逻辑代码是在Executor中执行的。