【Spark】RDD的行动算子

【Spark】RDD的行动算子

RDD的转换算子点击跳转

RDD的行动算子

• reduce
• collect
• count
• first
• take
• takeOrdered
• aggregate
• fold
• countByKey
• save 相关算子
• foreach

所谓的行动算子就是触发作业执行的方法

reduce

➢ 函数签名：def reduce(f: (T, T) => T): T
➢ 函数说明：聚集 RDD 中的所有元素，先聚合分区内数据，再聚合分区间数据

  @Test
  def reduce(): Unit = {
    val rdd = sc.makeRDD(List(1,2,3,4))
    //直接出结果，不像转换算子，从一个RDD变成了另一个RDD
    val i: Int = rdd.reduce(_+_)
    println(i)
  }

结果展示

跳转顶部

---

collect

➢ 函数签名：def collect(): Array[T]
➢ 函数说明：在驱动程序中，以数组 Array 的形式返回数据集的所有元素

  @Test
  def collect(): Unit = {
    val rdd = sc.makeRDD(List(1,2,3,4))
    //collect会将不同分区的数据按照分区顺序采集到Driver端内存中，形成数组
    val ints: Array[Int] = rdd.collect()
    println(ints.mkString(","))
  }

结果展示

跳转顶部

---

count

➢ 函数签名：def count(): Long
➢ 函数说明：返回 RDD 中元素的个数

  @Test
  def count(): Unit = {
    val rdd = sc.makeRDD(List(1, 2, 3, 4))
    //元素个数
    val cnt = rdd.count()
    println(cnt)
  }

结果展示

跳转顶部

---

first

➢ 函数签名：def first(): T
➢ 函数说明：返回 RDD 中的第一个元素

  @Test
  def first(): Unit = {
    val rdd = sc.makeRDD(List(1, 2, 3, 4))
    //返回第一个元素
    println(rdd.first())
  }

结果展示

跳转顶部

---

take

➢ 函数签名：def take(num: Int): Array[T]
➢ 函数说明：返回一个由 RDD 的前 n 个元素组成的数组

  @Test
  def takeOrdered(): Unit = {
    val rdd = sc.makeRDD(List(1, 2, 3, 4))
    //返回前几个元素
    println(rdd.take(2).mkString(","))
  }

结果展示

跳转顶部

---

takeOrdered

➢ 函数签名：def takeOrdered(num: Int)(implicit ord: Ordering[T]): Array[T]
➢ 函数说明：返回该 RDD 排序后的前 n 个元素组成的数组

  @Test
  def takeOrdered(): Unit = {
    val rdd = sc.makeRDD(List(1, 6, 2, 4))
    //返回排好序的前几个
    println(rdd.takeOrdered(2).mkString(","))
  }

结果展示

跳转顶部

---

aggregate

➢ 函数签名
def aggregate[U: ClassTag](zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U): U
➢ 函数说明：分区的数据通过初始值和分区内的数据进行聚合，然后再和初始值进行分区间的数据聚合

  @Test
  def aggregate(): Unit = {
    val rdd = sc.makeRDD(List(1, 2, 3, 4))
    println(rdd.aggregate(1)(_+_,_+_))
  }

结果展示

aggravat不仅仅会参与分区内的计算，还会参与分区间的计算

跳转顶部

---

fold

➢ 函数签名：def fold(zeroValue: T)(op: (T, T) => T): T
➢ 函数说明：折叠 *** 作，aggregate 的简化版 *** 作
的简化 *** 作

  @Test
  def fold(): Unit = {
    val rdd = sc.makeRDD(List(1, 2, 3, 4), 2)
    println(rdd.fold(1)(_ + _))
  }

结果展示

跳转顶部

---

countByKey

➢ 函数签名：def countByKey(): Map[K, Long]
➢ 函数说明：统计每种 key 的个数

  @Test
  def countByKey(): Unit = {
    val rdd = sc.makeRDD(List(("a",1),("b",1),("c",1),("a",1),("c",1)), 2)
    println(rdd.countByKey())
  }

结果展示

跳转顶部

---

save 相关算子

➢ 函数签名

def saveAsTextFile(path: String): Unit
def saveAsObjectFile(path: String): Unit
def saveAsSequenceFile(
	 path: String,
	 codec: Option[Class[_ <: CompressionCodec]] = None): Unit

➢ 函数说明：将数据保存到不同格式的文件中

  @Test
  def save(): Unit = {
    val rdd = sc.makeRDD(List(("a", 1), ("b", 1), ("c", 1), ("a", 1), ("c", 1)), 2)
    rdd.saveAsTextFile("output")
    rdd.saveAsObjectFile("output1")
    rdd.saveAsSequenceFile("output2")
  }

结果展示

saveAsSequenceFile要求数据格式必须为KV类型
跳转顶部

---

foreach

➢ 函数签名

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

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

  @Test
  def foreach(): Unit = {
    val rdd = sc.makeRDD(List(1, 2, 3, 4), 2)
    rdd.collect().foreach(println(_))
    println("#######################")
    rdd.foreach(println)
  }

结果展示

跳转顶部

---