Spark RDD常用算子整理

一、转换算子

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,Iterable))类型的 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() } }

spark

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。