Spark—常用的RDD转换算子
本篇介绍了一些转换算子的基本用法
文章目录
一、转换算子
假设有两个分区数据【1, 2】, 【3, 4】
- RDD的计算一个分区内的数据是一个一个执行逻辑
只有前面一个分区中的数据全部的逻辑执行完毕后,才会执行下一个数据。
分区内数据的执行是有序的。即【1,2】为一个分区,那么数据 1 的逻辑执行完后才会执行2的逻辑 - 不同分区之间数据计算是无序的。即【1,2】【3,4】为不同分区,谁先执行不确定,体现并行计算
1. Value 类型
1.1 Map
代码如下(示例):
// 准备环境
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
val rdd = sc.makeRDD(List(1, 2, 3, 4))
// 转换函数
def mapFunction(num: Int): Int = {
num * 2
}
// 传入方法
val mapRDD1: RDD[Int] = rdd.map(mapFunction)
// 传入匿名函数
val mapRDD2 = rdd.map(
num => {
num * 2
}
)
// 至简原则
val mapRDD3: RDD[Int] = rdd.map(_ * 2)
mapRDD1.collect().foreach(data => print(data+"\t"))
println()
mapRDD2.collect().foreach(data => print(data+"\t"))
println()
mapRDD3.collect().foreach(data => print(data+"\t"))
/*
result:
2 4 6 8
2 4 6 8
2 4 6 8
*/
// 关闭
sc.stop()
1.2 mapPartitions
-
函数签名
def mapPartitions[U: classtag]( f: Iterator[T] => Iterator[U], preservesPartitioning: Boolean = false): RDD[U]
-
函数说明
将待处理的数据以分区为单位发送到计算节点进行处理,这里的处理是指可以进行任意的处理,哪怕是过滤数据。 -
mapPartitions 说明
1)可以以分区为单位进行数据转换操作
2)但是会将整个分区的数据加载到内存进行引用
3)如果处理完的数据是不会被释放掉,存在对象的引用
4)在内存较小,数据量较大的场合下,容易出现内存溢出
/*
获取每个数据分区的最大值
*/
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
// Todo 算子 - mapPartitions
val rdd = sc.makeRDD(List(1,2,3,4), 2)
// 【1,2】,【3,4】
// result: 【2】,【4】
// 将获取到的数据放到迭代器中,然后再返回一个迭代器,类似于开了一个缓冲区
val mpRDD = rdd.mapPartitions(
iter => {
println(">>>>>>>>>>") // 两个分区,所以会打印两次
// 因为返回的必须是一个迭代器,所以先进行包装成List再转换成迭代器
List(iter.max).iterator
}
)
mpRDD.collect().foreach(println)
sc.stop()
1.3 mapPartitionsWithIndex
将分区进行编号,索引
- 函数签名
def mapPartitionsWithIndex[U: classtag]( f: (Int, Iterator[T]) => Iterator[U], preservesPartitioning: Boolean = false): RDD[U]
- 函数说明
将待处理的数据以分区为单位发送到计算节点进行处理,这里的处理是指可以进行任意的处理,哪怕是过滤数据,在处理时同时可以获取当前分区索引。
/*
返回当前数据及其所在分区
*/
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
// Todo 算子 - mapPartitions
val rdd = sc.makeRDD(List(1,2,3,4))
val mpwRDD = rdd.mapPartitionsWithIndex(
(index, iter) => {
// 1, 2, 3, 4
// result: (0,1) (0,2) (1,3) (1,4)
iter.map(
num => {
(index, num) // 返回一个tuple
}
)
// iter.map((index, _))
}
)
// result: (0,1) (0,2) (1,3) (1,4)
mpwRDD.collect().foreach(data => print(data + "\t"))
1.4 flatMap
整体拆分成个体
- 函数签名
def flatMap[U: classtag](f: T => TraversableOnce[U]): RDD[U]
- 函数说明
将处理的数据进行扁平化后再进行映射处理,所以算子也称之为扁平映射
/*
将 List(List(1,2),3,List(4,5))进行扁平化操作
*/
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
// Todo 算子 - flatMap
// 此rdd类型为Any
val rdd = sc.makeRDD(List(List(1,2),3,List(4,5)))
// result: 1 2 3 4 5
val flatRDD = rdd.flatMap(
data => {
data match {
// 模式匹配,判断传进来的数据的类型
case list:List[_] => list
case dat => List(dat) // 封装成一个List
}
}
)
// result: 1 2 3 4 5
flatRDD.collect().foreach(data => print(data + "\t"))
sc.stop()
1.5 glom
个体合并成整体,封装成一个Array
/*
计算所有分区最大值求和(分区内取最大值,分区间最大值求和)
*/
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
// Todo 算子 - glom
val rdd : RDD[Int] = sc.makeRDD(List(1,2,3,4), 2)
// part: 【1,2】,【3,4】
// max: 【2】,【4】
// sum: 【6】
val glomRDD: RDD[Array[Int]] = rdd.glom()
// 此处返回的是每个分区中最大值
// 将同一个分区的数据直接转换为相同类型的内存数组进行处理,分区不变
val maxRDD: RDD[Int] = 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。极限情况下,数据可能被分在同一个分区中;一个组的数据在一个分区中,但是并不是说一个分区中只有一个组。
/*
将 List("Hello", "hive", "hbase", "Hadoop")根据单词首写字母进行分组。
*/
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
// Todo 算子 - groupBy
// // 分组和分区没有必然的关系,这里设置3个分区
val rdd = sc.makeRDD(List("Hello", "Spark", "Scala", "Hadoop"), 3)
// groupBy会将数据源中的每一个数据进行分组判断,根据返回的分组key进行分组
// 相同的key值的数据会放置在一个组中
/*
result:
(H,CompactBuffer(Hello, Hadoop))
(S,CompactBuffer(Spark, Scala))
*/
// 根据单词首写字母进行分组
val groupRDD = rdd.groupBy(_.charat(0))
groupRDD.collect().foreach(println)
sc.stop()
1.7 filter
- 函数签名
def filter(f: T => Boolean): RDD[T]
- 函数说明
将数据根据指定的规则进行筛选过滤,符合规则的数据保留,不符合规则的数据丢弃。
当数据进行筛选过滤后,分区不变,但是分区内的数据可能不均衡,生产环境下,可能会出现数据倾斜。
// 算子 - filter
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
val rdd = sc.makeRDD(List("hello","spark","hi","scala","hhh"))
// 筛选出以 'h' 开头的字符串
val filterRDD = rdd.filter(_.startsWith("h"))
// result: hello hi hhh
filterRDD.collect().foreach(data => print(data + "\t"))
sc.stop()
1.8 sample
明显出现数据倾斜时,通过抽取数据。判断倾斜的原因,例如根据key进行分组时,可以通过抽取数据,判断导致倾斜的key,进而在进行一开始的数据计算时,可以对key进行预处理,比如转换等操作,不然它们放在同一个组,便于提高效率
- 函数签名
def sample( withReplacement: Boolean, fraction: Double, seed: Long = Utils.random.nextLong): RDD[T]
- 函数说明
根据指定的规则从数据集中抽取数据
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
// Todo 算子 - filter
val rdd = sc.makeRDD(List(1,2,3,4,5,6,7,8,9,10))
/*
抽取不放回的场合:
根据随机种子和随机算法分别求每个数被抽取的概率
再和第二个参数(基准值)进行对比,如果小于基准值则要,大于不要
*/
// sample算子需要传递三个参数
// 1. 第一个参数表示,抽取数据后是否将数据返回 true(放回),false(丢弃)
// 2. 第二个参数表示,
// 如果抽取不放回的场合:数据源中每条数据被抽取的概率,基准值的概念
// 如果抽取放回的场合:表示数据源中的每条数据元素被期望抽取到的次数
// 3. 第三个参数表示,抽取数据时随机算法的种子
// 如果不传递第三个参数,那么使用的是当前系统时间
// 因为是随机,所以每次可能不同
// result1: 3,4,10
// result2: 1,2,4,7,8
println(rdd.sample(
false, // 不放回
0.4
//1
).collect().mkString(","))
// result: 1,1,1,2,3,3,5,5,5,6,6,7,7,8,9,9,9,10,10,10
println(rdd.sample(
true,
2
//1
).collect().mkString(","))
sc.stop()
1.9 distinct
➢ 函数签名
def distinct()(implicit ord: Ordering[T] = null): RDD[T]
def distinct(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]
// 源码
def distinct(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {
map(x => (x, null)).reduceByKey((x, y) => x, numPartitions).map(_._1)
}
➢ 函数说明
将数据集中重复的数据去重
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
// Todo 算子 - filter
val rdd = sc.makeRDD(List(1,2,3,4,1,2,3,4))
// scala对List的去重,底层实现的是HashSet
// 源码解析
// map(x => (x, null)).reduceByKey((x, _) => x, numPartitions).map(_._1)
// 1. 先映射成 (x, null) 格式
// (1, null),(2, null),(3, null),(4, null),(1, null),(2, null),(3, null),(4, null)
// 2. reduceByKey
// (1, null)(1, null)(1, null)
// (null, null) => null
// 3. 再取元组中第一个元素
// (1, null) => 1
val rdd1: RDD[Int] = rdd.distinct()
// result: 4,2,1,3
println(rdd1.collect().mkString(","))
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 方法,收缩合并分区,减少分区的个数,减小任务调度成本
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
// Todo 算子 - filter
val rdd = sc.makeRDD(List(1,2,3,4,5,6), 3)
// coalesce方法默认情况下不会将分区的数据打乱重新组合
// 这种情况下的缩减分区可能会导致数据不均衡,出现数据倾斜
// 如果想要让数据均衡,可以进行shuffle处理
//val newRDD: RDD[Int] = rdd.coalesce(2)
// 第一个参数:最终分区的数量,第二个参数:是否进行shuffle处理
val newRDD: RDD[Int] = rdd.coalesce(2,true)
newRDD.saveAsTextFile("output")
sc.stop()
1.11 repartition
- 函数签名
def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] // 底层源码 def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope { coalesce(numPartitions, shuffle = true) }
- 函数说明
该操作内部其实执行的是 coalesce 操作,参数 shuffle 的默认值为 true。无论是将分区数多的RDD 转换为分区数少的 RDD,还是将分区数少的 RDD 转换为分区数多的 RDD,repartition操作都可以完成,因为无论如何都会经 shuffle 过程。
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
// Todo 算子 - repartition
val rdd = sc.makeRDD(List(1,2,3,4,5,6), 2)
// coalesce算子可以扩大分区的,但是如果不进行shuffle操作,是没有意义,不起作用。
// 所以如果想要实现扩大分区的效果,需要使用shuffle操作
// spark提供了一个简化的操作
// 缩减分区:coalesce,如果想要数据均衡,可以采用shuffle
// 扩大分区:repartition, 底层代码调用的就是coalesce,而且肯定采用shuffle
//val newRDD: RDD[Int] = rdd.coalesce(3, true)
val newRDD: RDD[Int] = rdd.repartition(3)
newRDD.saveAsTextFile("output")
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 的过程
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
// Todo 算子 - sortBy
val rdd = sc.makeRDD(List(6,2,4,5,3,1), 2)
// 默认升序,分区数与原来保持一致
val newRDD: RDD[Int] = rdd.sortBy(num=>num)
newRDD.saveAsTextFile("output")
sc.stop()
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
// Todo 算子 - sortBy
val rdd = sc.makeRDD(List(("1", 1), ("11", 2), ("2", 3)), 2)
// sortBy方法可以根据指定的规则对数据源中的数据进行排序,默认为升序,第二个参数可以改变排序的方式
// sortBy默认情况下,不会改变分区。但是中间存在shuffle操作
val newRDD = rdd.sortBy(t=>t._1.toInt, false) // false降序
// result: (11,2),(2,3),(1,1)
println(newRDD.collect().mkString(","))
sc.stop()
2. 双 Value 类型
2.1 intersection(交集), union(并集), subtract(差集), 拉链(zip)
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
// Todo 算子 - 双Value类型
// 交集,并集和差集要求两个数据源数据类型保持一致
// 拉链操作两个数据源的类型可以不一致
val rdd1 = sc.makeRDD(List(1,2,3,4))
val rdd2 = sc.makeRDD(List(3,4,5,6))
val rdd7 = sc.makeRDD(List("3","4","5","6"))
// 交集 : 4,3
// 对源 RDD 和参数 RDD 求交集后返回一个新的 RDD
val rdd3: RDD[Int] = rdd1.intersection(rdd2)
//val rdd8 = rdd1.intersection(rdd7)
println(rdd3.collect().mkString(","))
// 并集 : 1,2,3,4,3,4,5,6
// 对源 RDD 和参数 RDD 求并集后返回一个新的 RDD
val rdd4: RDD[Int] = rdd1.union(rdd2)
println(rdd4.collect().mkString(","))
// 差集 : 2,1
// 以一个 RDD 元素为主,去除两个 RDD 中重复元素,将其他元素保留下来。求差集
val rdd5: RDD[Int] = rdd1.subtract(rdd2)
println(rdd5.collect().mkString(","))
// 拉链 : (1,3),(2,4),(3,5),(4,6)
// 将两个 RDD 中的元素,以键值对的形式进行合并。其中,键值对中的 Key 为第 1 个 RDD中的元素,Value 为第 2 个 RDD 中的相同位置的元素
/*
1. 如果两个rdd中的元素数量不同,则会报错,这里与scala中不同
Caused by: org.apache.spark.SparkException:
Can only zip RDDs with same number of elements in each partition
2. 如果两个rdd中的元素数量不同,则会报错
Can't zip RDDs with unequal numbers of partitions: List(2, 4)
*/
val rdd6: RDD[(Int, Int)] = rdd1.zip(rdd2)
val rdd8 = rdd1.zip(rdd7)
println(rdd6.collect().mkString(","))
sc.stop()
3. Key - Value 类型
3.1 partitionBy
- 函数签名
def partitionBy(partitioner: Partitioner): RDD[(K, V)]
- 函数说明
将数据按照指定 Partitioner 重新进行分区。Spark 默认的分区器是 HashPartitioner
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
// Todo 算子 - (Key - Value类型)
val rdd = sc.makeRDD(List(1,2,3,4),2)
// (2,1),(4,1)...
val mapRDD:RDD[(Int, Int)] = rdd.map((_,1))
// RDD => PairRDDFunctions
// 隐式转换(二次编译)
/*
源码:
implicit def rddToPairRDDFunctions[K, V](rdd: RDD[(K, V)])
(implicit kt: classtag[K], vt: classtag[V], ord: Ordering[K] = null): PairRDDFunctions[K, V] = {
new PairRDDFunctions(rdd)
}
*/
/*
源码:
class HashPartitioner(partitions: Int) extends Partitioner {
def getPartition(key: Any): Int = key match {
case null => 0
case _ => Utils.nonNegativeMod(key.hashCode, numPartitions)
}
}
*/
/*
源码:
def nonNegativeMod(x: Int, mod: Int): Int = {
val rawMod = x % mod
rawMod + (if (rawMod < 0) mod else 0)
}
*/
// partitionBy根据指定的分区规则对数据进行重分区 分区规则--> val rawMod = x % mod 模运算
val newRDD = mapRDD.partitionBy(new HashPartitioner(2))
/*
源码:
override def equals(other: Any): Boolean = other match {
case h: HashPartitioner =>
h.numPartitions == numPartitions
case _ =>
false
}
*/
/*
如果再次使用分区器,则会调用equals()方法,
对传进来的Partitioner进行非空校验比较,如果是分区器,则返回self,否则返回新的RDD
*/
newRDD.partitionBy(new HashPartitioner(2))
newRDD.saveAsTextFile("output")
sc.stop()
通过计算规则,即底层源码体现,模运算,1,3放在同一个分区,2,4放在同一个分区
3.2 reduceByKey
- 函数签名
def reduceByKey(func: (V, V) => V): RDD[(K, V)] def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)]
- 函数说明
可以将数据按照相同的 Key 对 Value 进行聚合
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
// Todo 算子 - (Key - Value类型) reduceByKey
val rdd = sc.makeRDD(List(
("a", 1), ("a", 2), ("a", 3), ("b", 4),("b", 3),("c", 5)
))
// reduceByKey : 相同的key的数据进行value数据的聚合操作
// scala语言中一般的聚合操作都是两两聚合,spark基于scala开发的,所以它的聚合也是两两聚合
// 【1,2,3】
// a: 【3,3】 => 【6】
// b: 【4,3】 => 【7】
// c: 【5】
// reduceByKey中如果key的数据只有一个,是不会参与运算的。
val reduceRDD: RDD[(String, Int)] = rdd.reduceByKey( (x:Int, y:Int) => {
println(s"x = ${x}, y = ${y}")
x + y
} )
// val reduceRDD: RDD[(String, Int)] = rdd.reduceByKey(_+_)
/*
result:
x = 1, y = 2 // a 第一次聚合的两个数
x = 3, y = 3 // a 第二次聚合的两个数,第一个数为上一次聚合的数
x = 4, y = 3 // b 第一次聚合的两个数
(b,7)
(a,6)
(c,5) // c 的key只有一个,所以直接输出,没有运算的过程
*/
reduceRDD.collect().foreach(println)
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 进行分组
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
// 算子 - groupByKey
val rdd = sc.makeRDD(List(
("a", 1), ("a", 2), ("a", 3), ("b", 4)
))
// groupByKey : 将数据源中的数据,相同key的数据分在一个组中,形成一个对偶元组
// 元组中的第一个元素就是key,
// 元组中的第二个元素就是相同key的value的集合
val groupRDD: RDD[(String, Iterable[Int])] = rdd.groupByKey()
groupRDD.collect().foreach(println)
println("----------------")
// groupBy没有指定对key分组,不会把value单独拿出来,这里会把整体作为一个分组元素
val groupRDD1: RDD[(String, Iterable[(String, Int)])] = rdd.groupBy(_._1)
groupRDD1.collect().foreach(println)
/*
result:
(b,CompactBuffer(4))
(a,CompactBuffer(1, 2, 3))
----------------
(b,CompactBuffer((b,4)))
(a,CompactBuffer((a,1), (a,2), (a,3)))
*/
sc.stop()
3.4 aggregateByKey
- 函数签名
def aggregateByKey[U: classtag](zerovalue: U)(seqOp: (U, V) => U, combOp: (U, U) => U): RDD[(K, U)]
- 函数说明
将数据根据不同的规则进行分区内计算和分区间计算
/*
取出每个分区内相同 key 的最大值然后分区间相加
*/
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
// Todo 算子 - (Key - Value类型)
val rdd = sc.makeRDD(List(
("a", 1), ("a", 2), ("a", 3), ("a", 4)
),2)
// aggregateByKey存在函数柯里化,有两个参数列表
// 第一个参数列表,需要传递一个参数,表示为初始值
// 主要用于当碰见第一个key的时候,和value进行分区内计算
// 第二个参数列表需要传递2个参数
// 第一个参数表示分区内计算规则
// 第二个参数表示分区间计算规则
// math.min(x, y)
// math.max(x, y)
// (a,【1,2】), (a, 【3,4】)
// (a, 2), (a, 4)
// result: (a, 6)
rdd.aggregateByKey(0)( // 初始值
(x, y) => math.max(x, y), // 分区内计算规则
(x, y) => x + y // 分区间计算规则
).collect.foreach(println)
sc.stop()
3.5 foldByKey
-
函数签名
def foldByKey(zerovalue: V)(func: (V, V) => V): RDD[(K, V)]
-
函数说明
当分区内计算规则和分区间计算规则相同时,aggregateByKey 就可以简化为foldByKey
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
// 算子 - foldByKey
val rdd = sc.makeRDD(List(
("a", 1), ("a", 2), ("b", 3),
("b", 4), ("b", 5), ("a", 6)
),2)
rdd.aggregateByKey(0)(_+_, _+_).collect.foreach(println)
println("-------------")
// 如果聚合计算时,分区内和分区间计算规则相同,spark提供了简化的方法
rdd.foldByKey(0)(_+_).collect.foreach(println)
/*
result:
(b,12)
(a,9)
-------------
(b,12)
(a,9)
*/
sc.stop()
3.6 combineByKey
- 函数签名
def combineByKey[C]( createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C): RDD[(K, C)]
- 函数说明
最通用的对 key-value 型 rdd 进行聚集操作的聚集函数(aggregation function)。类似于aggregate(),combineByKey()允许用户返回值的类型与输入不一致。
/*
求每个 key 的平均值
*/
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
// 算子 - combineByKey
val rdd = sc.makeRDD(List(
("a", 1), ("a", 2), ("b", 3),
("b", 4), ("b", 5), ("a", 6)
), 2)
// combineByKey : 方法需要三个参数
// 第一个参数表示:将相同key的第一个数据进行结构的转换,实现操作
// 第二个参数表示:分区内的计算规则
// 第三个参数表示:分区间的计算规则
// combineByKey()的方式
val newRDD: RDD[(String, (Int, Int))] = rdd.combineByKey(
v => (v, 1),
(t: (Int, Int), v) => {
(t._1 + v, t._2 + 1)
},
(t1: (Int, Int), t2: (Int, Int)) => {
(t1._1 + t2._1, t1._2 + t2._2)
}
)
// aggregateByKey()的方式
// val newRDD: RDD[(String, (Int, Int))] = rdd.aggregateByKey((0, 0))(
// (t, v) => {
// (t._1 + v, t._2 + 1)
// },
// (t1, t2) => {
// (t1._1 + t2._1, t1._2 + t2._2)
// }
// )
val resultRDD: RDD[(String, Int)] = newRDD.mapValues {
case (num, cnt) => {
num / cnt
}
}
resultRDD.collect().foreach(println)
/*
result:
(b,4)
(a,3)
*/
sc.stop()
3.7 join
-
函数签名
def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]
-
函数说明
在类型为(K,V)和(K,W)的 RDD 上调用,返回一个相同 key 对应的所有元素连接在一起的(K,(V,W))的 RDD
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
// Todo 算子 - (Key - Value类型)
val rdd1 = sc.makeRDD(List(
("a", 1), ("a", 2), ("c", 3)
))
val rdd2 = sc.makeRDD(List(
("a", 5), ("c", 6),("a", 4)
))
// join : 两个不同数据源的数据,相同的key的value会连接在一起,形成元组
// 如果两个数据源中key没有匹配上,那么数据不会出现在结果中
// 如果两个数据源中key有多个相同的,会依次匹配,可能会出现笛卡尔乘积,数据量会几何性增长,会导致性能降低。
val joinRDD: RDD[(String, (Int, Int))] = rdd1.join(rdd2)
joinRDD.collect().foreach(println)
/*
// b 没有匹配到,所以不会出现在结果中
result:
(a,(1,5))
(a,(1,4))
(a,(2,5))
(a,(2,4))
(c,(3,6))
*/
sc.stop()
3.8 leftOuterJoin & rightOuterJoin
- 函数签名
def leftOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (V, Option[W]))] = self.withScope { leftOuterJoin(other, defaultPartitioner(self, other)) } def rightOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (Option[V], W))] = self.withScope { rightOuterJoin(other, defaultPartitioner(self, other)) }
- 函数说明
类似于 sql 语句的左右外连接
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
val rdd1 = sc.makeRDD(List(
("a", 1), ("b", 2)//, ("c", 3)
))
val rdd2 = sc.makeRDD(List(
("a", 4), ("b", 5),("c", 6)
))
println("leftOuterJoin------------")
val leftJoinRDD = rdd1.leftOuterJoin(rdd2)
val rightJoinRDD = rdd1.rightOuterJoin(rdd2)
leftJoinRDD.collect().foreach(println)
println("rightOuterJoin------------")
rightJoinRDD.collect().foreach(println)
/*
leftOuterJoin------------
(b,(2,Some(5)))
(a,(1,Some(4)))
rightOuterJoin------------
(b,(Some(2),5))
(a,(Some(1),4))
(c,(None,6))
*/
sc.stop()
3.9 cogroup
- 函数签名
def cogroup[W](other: RDD[(K, W)]): RDD[(K, (Iterable[V], Iterable[W]))]
- 函数说明
在类型为(K,V)和(K,W)的 RDD 上调用,返回一个(K,(Iterable,Iterable))类型的 RDD
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
// Todo 算子 - (Key - Value类型)
val rdd1 = sc.makeRDD(List(
("a", 1), ("b", 2)//, ("c", 3)
))
val rdd2 = sc.makeRDD(List(
("a", 4), ("b", 5),("c", 6),("c", 7)
))
// cogroup : connect + group (分组,连接)
val cGrdD: RDD[(String, (Iterable[Int], Iterable[Int]))] = rdd1.cogroup(rdd2)
cGrdD.collect().foreach(println)
/*
result:
(b,(CompactBuffer(2),CompactBuffer(5)))
(a,(CompactBuffer(1),CompactBuffer(4)))
(c,(CompactBuffer(),CompactBuffer(6, 7)))
*/
sc.stop()