- 微信
- 微博
  
  分享文章到微博
- 复制链接
  
  复制链接到剪贴板

Spark RDD常用算子整理

lwq1228 发表于 2021/03/19 16:32:20 2021/03/19

【摘要】 RDD算子分为转换算子和行动算子，转换算子根据数据处理方式的不同将算子整体上分为 Value 类型、双 Value 类型和 Key-Value类型，转换算子不会提交作业，而行动算子会提交作业。

一、转换算子

1、Value类型

1.1、map

函数签名：

def map[U: ClassTag](f: T => U): RDD[U]

函数说明：

将处理的数据逐条进行映射转换，这里的转换可以是类型的转换，也可以是值的转换。

使用样例：

object Spark_Rdd_Operator_Transform01 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(1, 2, 3, 4))
    // 使用map转换算子
    val mapRdd = rdd.map(_ * 2)
    //采集数据并打印
    mapRdd.collect().foreach(println)
    // 关闭环境
    sc.stop()
  }
}

1.2、mapPartitions

函数签名：

def mapPartitions[U: ClassTag](
      f: Iterator[T] => Iterator[U],
      preservesPartitioning: Boolean = false): RDD[U]

函数说明：

将待处理的数据以分区为单位发送到计算节点进行处理，这里的处理是指可以进行任意的处理，哪怕是过滤数据。

使用样例：

object Spark_Rdd_Operator_Transform02 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(1, 2, 3, 4), 2)
    // 使用mapPartitions转换算子
    val mapRdd = rdd.mapPartitions(iter => {
      println(">>>>>>>>>>>>>")
      iter.map(_ * 2)
    })
    //采集数据并打印
    mapRdd.collect().foreach(println)
    // 关闭环境
    sc.stop()
  }
}

map 和 mapPartitions 的区别

数据处理角度：Map算子是分区内一个数据一个数据的执行，类似于串行操作。而mapPartitions算子是以分区为单位进行批处理操作。
功能的角度：Map算子主要目的将数据源中的数据进行转换和改变。但是不会减少或增多数据。MapPartitions算子需要传递一个迭代器，返回一个迭代器，没有要求的元素的个数保持不变，所以可以增加或减少数据
性能的角度：Map算子因为类似于串行操作，所以性能比较低，而是mapPartitions算子类似于批处理，所以性能较高。但是mapPartitions算子会长时间占用内存，那么这样会导致内存可能不够用，出现内存溢出的错误。所以在内存有限的情况下，不推荐使用。使用map操作。

1.3、mapPartitionsWithIndex

函数签名：

def mapPartitionsWithIndex[U: ClassTag](
      f: (Int, Iterator[T]) => Iterator[U],
      preservesPartitioning: Boolean = false): RDD[U]

函数说明：

将待处理的数据以分区为单位发送到计算节点进行处理，这里的处理是指可以进行任意的处理，哪怕是过滤数据，在处理时同时可以获取当前分区索引。

使用样例：

object Spark_Rdd_Operator_Transform03 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(1, 2, 3, 4), 2)
    // 使用mapPartitionsWithIndex转换算子获取指定分区的数据
    val mapRdd = rdd.mapPartitionsWithIndex((index, iter) => {
      if (index == 1) {
        iter
      }
      else {
        Nil.iterator
      }
    })
    //采集数据并打印
    mapRdd.collect().foreach(println)
    // 关闭环境
    sc.stop()
  }
}

1.4、flatMap

函数签名：

def flatMap[U: ClassTag](f: T => TraversableOnce[U]): RDD[U]

函数说明：

将处理的数据进行扁平化后再进行映射处理，所以算子也称之为扁平映射

使用样例：

object Spark_Rdd_Operator_Transform04 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(List(1, 2), 3, List(4, 5)))
    // 转换算子,使用模式匹配
    val flatRdd = rdd.flatMap {
      case list: List[_] => list
      case dat => List(dat)
    }
    //采集数据并打印
    flatRdd.collect().foreach(println)
    // 关闭环境
    sc.stop()
  }
}

1.5、glom

函数签名：

def glom(): RDD[Array[T]]

函数说明：

将同一个分区的数据直接转换为相同类型的内存数组进行处理，分区不变

使用样例：

object Spark_Rdd_Operator_Transform05 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(1, 2, 3, 4), 2)
    val glomRdd = rdd.glom()
    // 分区内取最大值
    val maxRdd = glomRdd.map(array => array.max)
    // 分区间最大值求和
    println(maxRdd.collect().sum)
    // 关闭环境
    sc.stop()
  }
}

1.6、groupBy

函数签名：

def groupBy[K](f: T => K)(implicit kt: ClassTag[K]): RDD[(K, Iterable[T])]

函数说明：

将数据根据指定的规则进行分组, 分区默认不变，但是数据会被打乱重新组合，我们将这样的操作称之为shuffle。极限情况下，数据可能被分在同一个分区中。
一个组的数据在一个分区中，但是并不是说一个分区中只有一个组。

使用样例：

object Spark_Rdd_Operator_Transform06 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List("Hello", "hive", "hbase", "Hadoop"), 1)
    // 数据分组
    val groupRdd = rdd.groupBy(_.charAt(0))
    // 打印数据
    groupRdd.collect().foreach(println)
    // 关闭环境
    sc.stop()
  }
}

1.7、filter

函数签名：

def filter(f: T => Boolean): RDD[T]

函数说明：

将数据根据指定的规则进行筛选过滤，符合规则的数据保留，不符合规则的数据丢弃。当数据进行筛选过滤后，分区不变，但是分区内的数据可能不均衡，生产环境下，可能会出现数据倾斜。

使用样例：

object Spark_Rdd_Operator_Transform07 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(1, 2, 3, 4), 1)
    // 数据过滤（取偶数）
    val filterRdd = rdd.filter(_ % 2 == 0)
    // 打印数据
    filterRdd.collect().foreach(println)
    // 关闭环境
    sc.stop()
  }
}

1.8、sample

函数签名：

def sample(
      withReplacement: Boolean,
      fraction: Double,
      seed: Long = Utils.random.nextLong): RDD[T]

函数说明：

根据指定的规则从数据集中抽取数据

使用样例：

object Spark_Rdd_Operator_Transform08 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    val rdd = sc.makeRDD(List(1, 2, 3, 4), 1)
    // 抽取数据不放回（伯努利算法）
    // 伯努利算法：又叫 0、 1 分布。例如扔硬币，要么正面，要么反面。
    // 具体实现：根据种子和随机算法算出一个数和第二个参数设置几率比较，小于第二个参数要，大于不要
    // 第一个参数：抽取的数据是否放回， false：不放回
    // 第二个参数：抽取的几率，范围在[0,1]之间,0：全不取； 1：全取；
    // 第三个参数：随机数种子
    val dataRdd1 = rdd.sample(false, 0.5)
    // 打印数据
    dataRdd1.collect().foreach(println)

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

    // 抽取数据放回（泊松算法）
    // 第一个参数：抽取的数据是否放回， true：放回； false：不放回
    // 第二个参数：重复数据的几率，范围大于等于 0.表示每一个元素被期望抽取到的次数
    // 第三个参数：随机数种子
    val dataRdd2 = rdd.sample(true, 2)
    // 打印数据
    dataRdd2.collect().foreach(println)
    // 关闭环境
    sc.stop()
  }
}

1.9、distinct

函数签名：

def distinct(): RDD[T]
def distinct(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]

函数说明：

将数据集中重复的数据去重

使用样例：

object Spark_Rdd_Operator_Transform09 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(1, 2, 3, 4, 1, 2), 1)
    val dataRdd1 = rdd.distinct()
    // 打印数据
    dataRdd1.collect().foreach(println)

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

    val dataRdd2 = rdd.distinct(2)
    // 打印数据
    dataRdd2.collect().foreach(println)
    // 关闭环境
    sc.stop()
  }
}

1.10、coalesce

函数签名：

def coalesce(numPartitions: Int, shuffle: Boolean = false,
               partitionCoalescer: Option[PartitionCoalescer] = Option.empty)
              (implicit ord: Ordering[T] = null)
      : RDD[T]

函数说明：

根据数据量缩减分区，用于大数据集过滤后，提高小数据集的执行效率当spark程序中，存在过多的小任务的时候，可以通过coalesce方法，收缩合并分区，减少分区的个数，减小任务调度成本。

使用样例：

object Spark_Rdd_Operator_Transform10 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(1, 2, 3, 4, 1, 2), 6)
    val dataRdd = rdd.coalesce(2)
    dataRdd.saveAsTextFile("output")
    // 关闭环境
    sc.stop()
  }
}

1.11、repartition

函数签名：

def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]

函数说明：

该操作内部其实执行的是coalesce操作，参数shuffle的默认值为true。无论是将分区数多的RDD转换为分区数少的RDD，还是将分区数少的RDD转换为分区数多的RDD，repartition操作都可以完成，因为无论如何都会经shuffle过程。

使用样例：

object Spark_Rdd_Operator_Transform11 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(1, 2, 3, 4, 1, 2), 2)
    val dataRdd = rdd.repartition(4)
    dataRdd.saveAsTextFile("output")
    // 关闭环境
    sc.stop()
  }
}

1.12、sortBy

函数签名：

def sortBy[K](
      f: (T) => K,
      ascending: Boolean = true,
      numPartitions: Int = this.partitions.length)
      (implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]

函数说明：

该操作用于排序数据。在排序之前，可以将数据通过f函数进行处理，之后按照f函数处理的结果进行排序，默认为升序排列。排序后新产生的RDD的分区数与原RDD的分区数一
致。中间存在shuffle的过程。

使用样例：

object Spark_Rdd_Operator_Transform12 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(1, 2, 3, 4, 1, 2), 2)
    val dataRdd = rdd.sortBy(num => num, false, 4)
    dataRdd.collect().foreach(println)
    // 关闭环境
    sc.stop()
  }
}

2、双 Value 类型

2.1、intersection

函数签名：

def intersection(other: RDD[T]): RDD[T]

函数说明：

对源RDD和参数RDD求交集后返回一个新的RDD

使用样例：

object Spark_Rdd_Operator_Transform13 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd1 = sc.makeRDD(List(1, 2, 3, 4))
    val rdd2 = sc.makeRDD(List(3, 4, 5, 6))
    // 数据取交集
    val rdd = rdd1.intersection(rdd2)
    // 打印数据
    rdd.collect().foreach(println)
    // 关闭环境
    sc.stop()
  }
}

2.2、union

函数签名：

def union(other: RDD[T]): RDD[T]

函数说明：

对源 RDD和参数RDD求并集后返回一个新的RDD

使用样例：

object Spark_Rdd_Operator_Transform14 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd1 = sc.makeRDD(List(1, 2, 3, 4))
    val rdd2 = sc.makeRDD(List(3, 4, 5, 6))
    // 数据取交集
    val rdd = rdd1.union(rdd2)
    // 打印数据
    rdd.collect().foreach(println)
    // 关闭环境
    sc.stop()
  }
}

2.3、subtract

函数签名：

def subtract(other: RDD[T]): RDD[T]

函数说明：

以一个RDD元素为主，去除两个RDD中重复元素，将其他元素保留下来。求差集

使用样例：

object Spark_Rdd_Operator_Transform15 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd1 = sc.makeRDD(List(1, 2, 3, 4))
    val rdd2 = sc.makeRDD(List(3, 4, 5, 6))
    // 数据取交集
    val rdd = rdd1.subtract(rdd2)
    // 打印数据
    rdd.collect().foreach(println)
    // 关闭环境
    sc.stop()
  }
}

2.4、zip

函数签名：

def zip[U: ClassTag](other: RDD[U]): RDD[(T, U)]

函数说明：

将两个RDD中的元素，以键值对的形式进行合并。其中，键值对中的Key为第1个RDD中的元素，Value为第2个RDD中的相同位置的元素。

使用样例：

object Spark_Rdd_Operator_Transform16 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd1 = sc.makeRDD(List(1, 2, 3, 4))
    val rdd2 = sc.makeRDD(List(3, 4, 5, 6))
    // 数据取交集
    val rdd = rdd1.zip(rdd2)
    // 打印数据
    rdd.collect().foreach(println)
    // 关闭环境
    sc.stop()
  }
}

3、Key - Value 类型

3.1、partitionBy

函数签名：

def partitionBy(partitioner: Partitioner): RDD[(K, V)]

函数说明：

将数据按照指定Partitioner重新进行分区。Spark默认的分区器是HashPartitioner。

使用样例：

object Spark_Rdd_Operator_Transform17 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(Array((1, "aaa"), (2, "bbb"), (3, "ccc")), 3)
    // 数据分组
    val rdd2 = rdd.partitionBy(new HashPartitioner(2))
    // 打印数据
    rdd2.saveAsTextFile("output")
    // 关闭环境
    sc.stop()
  }
}

3.2、reduceByKey

函数签名：

def reduceByKey(func: (V, V) => V): RDD[(K, V)]
def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)]
def reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)]

函数说明：

可以将数据按照相同的Key对Value进行聚合。

使用样例：

object Spark_Rdd_Operator_Transform18 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(("a", 1), ("b", 2), ("c", 3)))
    // 数据聚合
    val dataRdd1 = rdd.reduceByKey(_ + _)
    dataRdd1.saveAsTextFile("output1")

    val dataRdd2 = rdd.reduceByKey(_ + _, 2)
    dataRdd2.saveAsTextFile("output2")
    // 关闭环境
    sc.stop()
  }
}

3.3、groupByKey

函数签名：

def groupByKey(): RDD[(K, Iterable[V])]
def groupByKey(numPartitions: Int): RDD[(K, Iterable[V])]
def groupByKey(partitioner: Partitioner): RDD[(K, Iterable[V])]

函数说明：

将数据源的数据根据key对value进行分组。

使用样例：

object Spark_Rdd_Operator_Transform19 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(("a", 1), ("b", 2), ("c", 3)))
    // 数据根据key分组
    val dataRdd1 = rdd.groupByKey()
    val dataRdd2 = rdd.groupByKey(2)
    val dataRdd3 = rdd.groupByKey(new HashPartitioner(2))
    dataRdd1.saveAsTextFile("output1")
    dataRdd2.saveAsTextFile("output2")
    dataRdd3.saveAsTextFile("output3")
    // 关闭环境
    sc.stop()
  }
}

reduceByKey 和 groupByKey 的区别：

从shuffle的角度：reduceByKey和groupByKey都存在shuffle的操作，但是reduceByKey可以在shuffle前对分区内相同key的数据进行预聚合（combine）功能，这样会减少落盘的数据量，而groupByKey只是进行分组，不存在数据量减少的问题，reduceByKey性能比较高。
从功能的角度：reduceByKey其实包含分组和聚合的功能。GroupByKey只能分组，不能聚合，所以在分组聚合的场合下，推荐使用reduceByKey，如果仅仅是分组而不需要聚合。那么还是只能使用groupByKey

3.4、aggregateByKey

函数签名：

def aggregateByKey[U: ClassTag](zeroValue: U)(seqOp: (U, V) => U, combOp: (U, U) => U): RDD[(K, U)]
def aggregateByKey[U: ClassTag](zeroValue: U, numPartitions: Int)(seqOp: (U, V) => U, combOp: (U, U) => U): RDD[(K, U)]
def aggregateByKey[U: ClassTag](zeroValue: U, partitioner: Partitioner)(seqOp: (U, V) => U, combOp: (U, U) => U): RDD[(K, U)]

函数说明：

将数据根据不同的规则进行分区内计算和分区间计算

使用样例1：

object Spark_Rdd_Operator_Transform20 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(("a", 1), ("a", 2), ("c", 3)))
    // 数据根据key分组
    val dataRdd = rdd.aggregateByKey(0)(_ + _, _ + _)
    dataRdd.collect().foreach(println)
    // 关闭环境
    sc.stop()
  }
}

使用样例2：

object Spark_Rdd_Operator_Transform20 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 取出每个分区内相同 key 的最大值然后分区间相加
    // aggregateByKey 算子是函数柯里化，存在两个参数列表
    // 1. 第一个参数列表中的参数表示初始值
    // 2. 第二个参数列表中含有两个参数
    // 2.1 第一个参数表示分区内的计算规则
    // 2.2 第二个参数表示分区间的计算规则
    val rdd = sc.makeRDD(List(("a", 1), ("a", 2), ("c", 3), ("b", 4), ("c", 5), ("c", 6)), 2)

    // 0:("a",1),("a",2),("c",3) => (a,10)(c,10)
    // => (a,10)(b,10)(c,20)
    // 1:("b",4),("c",5),("c",6) => (b,10)(c,10)
    val dataRdd = rdd.aggregateByKey(10)(
      (x, y) => math.max(x, y),
      (x, y) => x + y
    )
    dataRdd.collect().foreach(println)
    // 关闭环境
    sc.stop()
  }
}

3.5、foldByKey

函数签名：

def foldByKey(zeroValue: V)(func: (V, V) => V): RDD[(K, V)]
def foldByKey(zeroValue: V, numPartitions: Int)(func: (V, V) => V): RDD[(K, V)]
def foldByKey(zeroValue: V, partitioner: Partitioner)(func: (V, V) => V): RDD[(K, V)]

函数说明：

当分区内计算规则和分区间计算规则相同时，aggregateByKey就可以简化为foldByKey

使用样例：

object Spark_Rdd_Operator_Transform21 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(("a", 1), ("a", 2), ("c", 3)))
    // 计算结果
    val dataRdd = rdd.foldByKey(0)(_ + _)
    dataRdd.collect().foreach(println)
    // 关闭环境
    sc.stop()
  }
}

3.6、combineByKey

函数签名：

def combineByKey[C](
      createCombiner: V => C,
      mergeValue: (C, V) => C,
      mergeCombiners: (C, C) => C,
      numPartitions: Int): RDD[(K, C)]

函数说明：

最通用的对key-value型rdd进行聚集操作的聚集函数（aggregation function）。类似于aggregate()，combineByKey()允许用户返回值的类型与输入不一致。

使用样例：

object Spark_Rdd_Operator_Transform22 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98)), 1)

    // 求每个 key 的平均值
    val combineRdd = rdd.combineByKey(
      (_, 1),
      (acc: (Int, Int), v) => (acc._1 + v, acc._2 + 1),
      (acc1: (Int, Int), acc2: (Int, Int)) => (acc1._1 + acc2._1, acc1._2 + acc2._2)
    )

    // 打印数据
    combineRdd.collect().foreach(println)
    // 关闭环境
    sc.stop()
  }
}

reduceByKey、 foldByKey、 aggregateByKey、 combineByKey 的区别

reduceByKey: 相同key的第一个数据不进行任何计算，分区内和分区间计算规则相同
foldByKey: 相同key的第一个数据和初始值进行分区内计算，分区内和分区间计算规则相同
aggregateByKey：相同key的第一个数据和初始值进行分区内计算，分区内和分区间计算规则可以不相同
combineByKey：当计算时，发现数据结构不满足要求时，可以让第一个数据转换结构。分区内和分区间计算规则不相同。

3.7、sortByKey

函数签名：

def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length)
      : RDD[(K, V)]

函数说明：

在一个(K,V)的 RDD 上调用， K 必须实现 Ordered 接口(特质)，返回一个按照 key 进行排序的

使用样例：

object Spark_Rdd_Operator_Transform23 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(("a", 1), ("b", 2), ("c", 3)))

    // 按照key排序
    val sortRdd1 = rdd.sortByKey(true)
    val sortRdd2 = rdd.sortByKey(false)

    // 打印数据
    sortRdd1.collect().foreach(println)
    println("*******")
    sortRdd2.collect().foreach(println)
    // 关闭环境
    sc.stop()
  }
}

3.8、join

函数签名：

def join[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (V, W))]

函数说明：

在类型为(K,V)和(K,W)的 RDD 上调用，返回一个相同 key 对应的所有元素连接在一起的(K,(V,W))的 RDD

使用样例：

object Spark_Rdd_Operator_Transform24 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd1 = sc.makeRDD(Array((1, "a"), (2, "b"), (3, "c")))
    val rdd2 = sc.makeRDD(Array((1, 4), (2, 5), (3, 6)))
    rdd1.join(rdd2).collect().foreach(println)

    // 关闭环境
    sc.stop()
  }
}

3.9、leftOuterJoin

函数签名：

def leftOuterJoin[W](other: RDD[(K, W)],partitioner: Partitioner): RDD[(K, (V, Option[W]))]

函数说明：

类似于 SQL 语句的左外连接

使用样例：

object Spark_Rdd_Operator_Transform25 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd1 = sc.makeRDD(Array((1, "a"), (2, "b"), (3, "c"), (4, "d")))
    val rdd2 = sc.makeRDD(Array((1, 4), (2, 5), (3, 6)))
    rdd1.leftOuterJoin(rdd2).collect().foreach(println)

    // 关闭环境
    sc.stop()
  }
}

3.10、rightOuterJoin

函数签名：

def rightOuterJoin[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (Option[V], W))]

函数说明：

类似于 SQL 语句的右外连接

使用样例：

object Spark_Rdd_Operator_Transform26 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd1 = sc.makeRDD(Array((1, "a"), (2, "b"), (3, "c"), (4, "d")))
    val rdd2 = sc.makeRDD(Array((1, 4), (2, 5), (3, 6)))
    rdd1.rightOuterJoin(rdd2).collect().foreach(println)

    // 关闭环境
    sc.stop()
  }
}

3.11、fullOuterJoin

函数签名：

def fullOuterJoin[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (Option[V], Option[W]))]

函数说明：

类似于 SQL 语句的全外连接

使用样例：

object Spark_Rdd_Operator_Transform27 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd1 = sc.makeRDD(Array((1, "a"), (2, "b"), (3, "c"), (4, "d")))
    val rdd2 = sc.makeRDD(Array((1, 4), (2, 5), (3, 6)))
    rdd1.fullOuterJoin(rdd2).collect().foreach(println)

    // 关闭环境
    sc.stop()
  }
}

3.12、cogroup

函数签名：

def cogroup[W](other: RDD[(K, W)]): RDD[(K, (Iterable[V], Iterable[W]))]

函数说明：

在类型为(K,V)和(K,W)的 RDD 上调用，返回一个(K,(Iterable<V>,Iterable<W>))类型的 RDD

使用样例：

object Spark_Rdd_Operator_Transform28 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd1 = sc.makeRDD(Array((1, "a"), (2, "b"), (3, "c"), (4, "d")))
    val rdd2 = sc.makeRDD(Array((1, 4), (2, 5), (3, 6)))
    rdd1.cogroup(rdd2).collect().foreach(println)

    // 关闭环境
    sc.stop()
  }
}

二、行动算子

2.1、reduce

函数签名：

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

函数说明：

聚集 RDD 中的所有元素，先聚合分区内数据，再聚合分区间数据

使用样例：

object Spark_Rdd_Operator_Action01 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(1, 2, 3, 4))
    // 聚合函数
    val result = rdd.reduce(_ + _)
    // 数据并打印
    println(result)
    // 关闭环境
    sc.stop()
  }
}

2.2、collect

函数签名：

def collect(): Array[T]

函数说明：

在驱动程序中，以数组 Array 的形式返回数据集的所有元素

使用样例：

object Spark_Rdd_Operator_Action02 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(1, 2, 3, 4))
    // 采集数据并打印
    rdd.collect().foreach(println)
    // 关闭环境
    sc.stop()
  }
}

2.3、count

函数签名：

def count(): Long

函数说明：

返回 RDD 中元素的个数

使用样例：

object Spark_Rdd_Operator_Action03 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(1, 2, 3, 4))
    val count = rdd.count()
    // 打印数据
    println(count)
    // 关闭环境
    sc.stop()
  }
}

2.4、first

函数签名：

def first(): T

函数说明：

返回 RDD 中的第一个元素

使用样例：

object Spark_Rdd_Operator_Action04 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(1, 2, 3, 4))
    val first = rdd.first()
    // 打印数据
    println(first)
    // 关闭环境
    sc.stop()
  }
}

2.5、take

函数签名：

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

函数说明：

返回一个由 RDD 的前 n 个元素组成的数组

使用样例：

object Spark_Rdd_Operator_Action05 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(1, 2, 3, 4))
    val data = rdd.take(2)
    // 打印数据
    data.foreach(println)
    // 关闭环境
    sc.stop()
  }
}

2.6、takeOrdered

函数签名：

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

函数说明：

返回该 RDD 排序后的前 n 个元素组成的数组

使用样例：

object Spark_Rdd_Operator_Action06 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(4, 3, 1, 2))
    val data = rdd.takeOrdered(2)
    // 打印数据
    data.foreach(println)
    // 关闭环境
    sc.stop()
  }
}

2.7、aggregate

函数签名：

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

函数说明：

分区的数据通过初始值和分区内的数据进行聚合，然后再和初始值进行分区间的数据聚合

使用样例：

object Spark_Rdd_Operator_Action07 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(1, 2, 3, 4), 4)
    val data = rdd.aggregate(10)(_ + _, _ + _)
    // 打印数据
    println(data)
    // 关闭环境
    sc.stop()
  }
}

2.8、fold

函数签名：

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

函数说明：

折叠操作， aggregate的简化版操作

使用样例：

object Spark_Rdd_Operator_Action08 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(1, 2, 3, 4), 4)
    val data = rdd.fold(10)(_ + _)
    // 打印数据
    println(data)
    // 关闭环境
    sc.stop()
  }
}

2.9、countByKey

函数签名：

def countByKey(): Map[K, Long]

函数说明：

统计每种 key 的个数

使用样例：

object Spark_Rdd_Operator_Action09 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List((1, "a"), (1, "a"), (1, "a"), (2, "b"), (3, "c"), (3, "c")))
    val countRdd = rdd.countByKey()
    // 打印数据
    countRdd.foreach(println)
    // 关闭环境
    sc.stop()
  }
}

2.10、save 相关算子

函数签名：

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

函数说明：

将数据保存到不同格式的文件中

使用样例：

object Spark_Rdd_Operator_Action10 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List((1, "a"), (1, "a"), (1, "a"), (2, "b"), (3, "c"), (3, "c")), 2)
    // 保存成 Text 文件
    rdd.saveAsTextFile("output1")
    // 序列化成对象保存到文件
    rdd.saveAsObjectFile("output2")
    // 保存成 Sequencefile 文件
    rdd.saveAsSequenceFile("output3")
    // 关闭环境
    sc.stop()
  }
}

2.11、foreach

函数签名：

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

函数说明：

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

使用样例：

object Spark_Rdd_Operator_Action11 {
  def main(args: Array[String]): Unit = {
    // 创建Spark上下文环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // 创建RDD
    val rdd = sc.makeRDD(List(1, 2, 3, 4))
    // 收集后打印
    rdd.map(num => num).collect().foreach(println)
    
    println("****************")
    // 分布式打印
    rdd.foreach(println)
    // 关闭环境
    sc.stop()
  }
}

点赞
收藏
关注作者

0/1000

抱歉，系统识别当前为高风险访问，暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称，即可参与社区互动！

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

确认取消

加入云驻计划，成为创作者

华为云周边好礼
免费体验产品
特殊身份标识
线下官方门票
内部专家零距离
与10000+优质创作者共同成长

立即加入

Spark RDD常用算子整理

一、转换算子

1、Value类型

1.1、map

1.2、mapPartitions

1.3、mapPartitionsWithIndex

1.4、flatMap

1.5、glom

1.6、groupBy

1.7、filter

1.8、sample

1.9、distinct

1.10、coalesce

1.11、repartition

1.12、sortBy

2、双 Value 类型

2.1、intersection

2.2、union

2.3、subtract

2.4、zip

3、Key - Value 类型

3.1、partitionBy

3.2、reduceByKey

3.3、groupByKey

3.4、aggregateByKey

3.5、foldByKey

3.6、combineByKey

3.7、sortByKey

3.8、join

3.9、leftOuterJoin

3.10、rightOuterJoin

3.11、fullOuterJoin

3.12、cogroup

二、行动算子

2.1、reduce

2.2、collect

2.3、count

2.4、first

2.5、take

2.6、takeOrdered

2.7、aggregate

2.8、fold

2.9、countByKey

2.10、save 相关算子

2.11、foreach

全部回复

设置昵称

关于作者

目录

加入云驻计划，成为创作者

推荐阅读

相关产品