测试准备
pom文件
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.12</artifactId>
<version>3.0.0</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>3.2.0</version>
</dependency>
</dependencies>
<build>
<plugins>
<!-- 该插件用于将Scala代码编译成class文件 -->
<plugin>
<groupId>net.alchim31.maven</groupId>
<artifactId>scala-maven-plugin</artifactId>
<version>3.2.2</version>
<executions>
<execution>
<!-- 声明绑定到maven的compile阶段 -->
<goals>
<goal>compile</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<version>3.0.0</version>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
测试类
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.{SparkConf, SparkContext}
import org.junit.{After, Before, Test}
/**
* 初始化SparkContext
*/
@Before
def innit {
val conf = new SparkConf().setAppName("RDDTest").setMaster("local[*]")
sc = new SparkContext(conf)
//删除上一次产生的output文件,不然报错
val fs = FileSystem.get(new Configuration())
val path = new Path("output")
if(fs.exists(path)){
fs.delete(path)
}
}
/**
* 关闭SparkContext
*/
def close()= {
sc.stop
}
一.单值Value类型转换算子
map
函数签名
def map[U: ClassTag](f: T => U): RDD[U] = withScope {
// 检查函数中是否有闭包,如果有闭包确保闭包变量可以被序列化,才能发送给Task,否则报错
val cleanF = sc.clean(f)
// iter.map : 调用的就是scala集合中的map方法 集合(分区)中的每个元素,都调用一次函数!
new MapPartitionsRDD[U, T](this, (_, _, iter) => iter.map(cleanF))
}
特点: ①不会改变RDD的分区数
②对每个分区中的元素调用函数进行计算,计算的结果还在原先分区
练习:从服务器日志数据apache.log中获取用户请求URL资源路径
/**
* map
* 数据格式:"83.149.9.216 - - 17/05/2015:10:05:07 +0000 GET /presentations/logstash-monitorama-2013/plugin/notes/notes.js"
*/
@Test
def testMap {
val rdd = sc.textFile("input/apache.log").map(_.split(" ")(6))
rdd.saveAsTextFile("output")
}
mapPartitions
一个分区调用一次函数,批处理
函数签名
def mapPartitions[U: ClassTag](
f: Iterator[T] => Iterator[U],
preservesPartitioning: Boolean = false): RDD[U] = withScope {
val cleanedF = sc.clean(f)
// cleanedF(iter) 一个分区调用一次函数,批处理
new MapPartitionsRDD(
this,
(_: TaskContext, _: Int, iter: Iterator[T]) => cleanedF(iter),
preservesPartitioning)
}
练习:获取每个数据分区的最大值
/**
* mapPartitions
*/
@Test
def testMapPartition {
val rdd = sc.makeRDD(List(1,2,3,4,5,6,7,8),2).mapPartitions(iter => {
List(iter.max).toIterator
})
rdd.saveAsTextFile("output")
}
mapPartitions和map的区别
map算子每一次处理一条数据,而mapPartitions算子每一次将一个分区的数据当成一个整体进行数据处理。如果一个分区的数据没有完全处理完,那么所有的数据都不会释放,即使前面已经处理完的数据也不会释放。容易出现内存溢出,所以当内存空间足够大时,为了提高效率,推荐使用mapPartitions算子
mapPartitionsWithIndex
一个分区调用一次函数,批处理 ,可以获得的当前分区索引
函数签名
def mapPartitionsWithIndex[U: ClassTag](
f: (Int, Iterator[T]) => Iterator[U],
preservesPartitioning: Boolean = false): RDD[U] = withScope {
val cleanedF = sc.clean(f)
new MapPartitionsRDD(
this,
// cleanedF(index, iter) : 批处理,提供了分区的索引
(_: TaskContext, index: Int, iter: Iterator[T]) => cleanedF(index, iter),
preservesPartitioning)
}
练习:获取第二个数据分区的数据
/**
* mapPartitionsWithIndex
*/
@Test
def testMapPartitionIndex {
val rdd = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7), 3).mapPartitionsWithIndex {
case (2, iter) => iter
case _ => Nil.toIterator
}
println(rdd.collect().mkString(","))
}
flatMap
扁平映射
练习:将List(List(1,2),3,List(4,5))进行扁平化操作
/**
* flatMap
*/
@Test
def testflatMap {
val rdd= sc.makeRDD(List(List(1, 2), 3, List(4, 5)))
val rdd1 =rdd.flatMap{
case x:Int => List(x)
case list: List[_] =>list
}
rdd1.collect().foreach(println)
}
glom 同分区转数组
合并一个分区的数据到一个数组中
函数签名
def glom(): RDD[Array[T]] = withScope {
new MapPartitionsRDD[Array[T], T](this, (_, _, iter) => Iterator(iter.toArray))
}
练习:计算所有分区最大值求和(分区内取最大值,分区间最大值求和)
/**
* glom
*/
@Test
def testglom {
val rdd = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7), 3).glom()
rdd.collect().foreach(x=>println(x.toBuffer))
println(rdd.map(_.max).sum())
}
groupBy 分组
数据根据指定的规则进行分组, 分区默认不变,也可以指定分组后的分区数
一个组的数据在一个分区中,但是并不是说一个分区中只有一个组
函数签名
// 根据f函数的返回值作为key,对T类型进行分组
def groupBy[K](f: T => K)(implicit kt: ClassTag[K]): RDD[(K, Iterable[T])] = withScope {
groupBy[K](f, defaultPartitioner(this))
}
filter过滤
sample抽样
distinct 去重
去重,去重后会有shuffle! 会对去重后的数据重新分区!
默认使用HashPartitoner对去重后的数据进行分区!
去重的原理
map(x => (x, null)).reduceByKey((x, _) => x, numPartitions).map(_._1)
coalesce重分区
Coalesce的作用是将当前RDD的数据重新分区到指定分区数的新RDD中。
①使用Coalesce从父RDD的多个分区 核减(合并) 到 少的分区,不会产生shuffle
②使用Coalesce 从少的分区 合并到多的分区,指定shuffle才会增加分区,否则分区数不变也不会shuffle。
③默认,如果要合并的分区比当前的分区数大,还会采用当前的分区数替换要合并的分区数
可以传入shuffle=true,完成从 少的分区合并到大的分区。
将会产生shuffle。默认使用HashPatitioner将数据重新分区!
repartition重分区
repartition用于增加分区,本质调用了可以shuffle的coalesce!
使用场景:减少分区使用coalesce ,增加分区使用repartition
sortBy 单值排序
针对单Value排序,可以指定分区规则,和排序规则,默认升序。
顶层调用了sortByKey,有shuffle
pipe 调用脚本
管道,针对每个分区,都调用一次shell脚本,返回输出的RDD。
注意:shell脚本需要放在计算节点可以访问到的位置
二.双Value类型转换算子
就是两个单值RDD之间的操作
intersect 交集
结果分区数=上游RDD最大的分区数
有shuffle,取上游最大的分区数的RDD的分区器,,默认使用HashPartitioner
两个RDD类型需要一致
union 并集
结果分区数=上游所有RDD的分区数之和
没有shuffle,就是简单的合并。要想实现数学意义的并集,需要去重
两个RDD类型需要一致
substract 差集
结果分区数=调用substract算子的RDD的分区数
有shuffle
两个RDD类型需要一致
cartesian 笛卡尔积
结果分区数=上游RDD的分区数之乘积
没有shuffle
zip 拉链
两个RDD的分区数和每个分区内的元素数需要一致,不然抛异常。
两个RDD数据类型可以不同。
val rdd = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7), 4)
val rdd1 = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7), 4)
rdd.zip(rdd1).saveAsTextFile("output")
三.Key-Value类型转换算子
partitionBy(自定义分区)
使用HashPartitioern
val rdd = sc.makeRDD(List(("a", 1), ("b", 2), ("a", 1), ("a", 4), ("c", 1), ("b", 5)),3)
rdd.partitionBy(new HashPartitioner(3)).saveAsTextFile("output")
Spark默认使用的就是HashPartitioner如果重分区的分区器和当前RDD的分区器一样, 不进行任何的处理, 不会再次重分区。如下
hashrdd的分区器为HashPartition,result使用HashPartition分区,不会进行处理,不会报错。
val rdd = sc.makeRDD(List(("a", 1), ("b", 2), ("a", 1), ("a", 4), ("c", 1), ("b", 5)),3)
val hashrdd = rdd.partitionBy(new HashPartitioner(3))
val result = rdd.partitionBy(new HashPartitioner(4))
使用RangePartitioner,一般sorrBy中使用的是这个分区器
自定义分区器
需求:有以下数据,希望年龄相同的进入同一个区。
User("tom", 12), User("kobe", 12), User("mick", 22), User("jack", 23)
import org.apache.spark.{Partitioner, SparkConf, SparkContext}
/**
* @description: TODO
* @author: HaoWu
* @create: 2020年08月03日
*/
object MyPartitionerTest {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("RDDTest").setMaster("local[*]")
val sc = new SparkContext(conf)
val list = List(User("tom", 12), User("kobe", 12), User("mick", 22), User("jack", 23))
val result = sc.makeRDD(list).map {
case User(name, age) => (age, name)
}.partitionBy(new MyPartitioner(3))
result.saveAsTextFile("output")
}
}
/**
* User样例类
*/
case class User(name: String, age: Int)
/**
* 自定义分区器
*/
class MyPartitioner(num: Int) extends Partitioner {
//设置分区数
override def numPartitions: Int = num
//分区规则
override def getPartition(key: Any): Int = {
//判断是否为Int类型
if (!key.isInstanceOf[Int]) {
0
} else {
//Hash分区具有聚类的作用,相同age的会被分如同一个区
key.asInstanceOf[Int] % numPartitions
}
}
}
groupByKey
结果RDD默认Hash分区器
函数签名
def groupByKey(): RDD[(K, Iterable[V])] = self.withScope {
groupByKey(defaultPartitioner(self))
}
示例
val rdd = sc.makeRDD(List((1,2),(2,4),(3,2),(1,1),(2,2),(3,1)), 2)
val result = rdd.groupByKey()
//result结果
//{(1,Array(2,1)), (2,Array(4,2)), (3,Array(2,1))}
分组的结果
mapValues
针对K-V类型的RDD,对所有value集合每个元素做相同操作。返回新的K-V类型的RDD
场景:求TOP3 {(省份1,List(12,21,12,23,1221)), (省份2,List(12,21,12,23,1221))....}
这种groupBy后的结果可以采用mapValues对Value取TOP3.
示例
//求TOP1
val rdd2 = sc.makeRDD(List(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98))
val rddg: RDD[(String, Iterable[Int])] = rdd2.groupByKey()
val result10 = rddg.mapValues(iter => {
val i =iter.toList
i.sortBy(x=>x).reverse.take(1)
})
result10.coalesce(1).saveAsTextFile("output")
reduceByKey
产生shuffle,先区内聚合=>分区间聚合()
结果RDD默认是Hash分区器
函数签名
def reduceByKey(func: (V, V) => V): RDD[(K, V)] = self.withScope {
reduceByKey(defaultPartitioner(self), func)
}
示例:
val rdd = sc.makeRDD(List((1,2),(2,4),(3,2),(1,1),(2,2),(3,1)), 2)
val result = rdd.reduceByKey(_+_)
//result结果
//(1,3),(2,6),(3,3)
注意:reduceByKey算子和groupByKey在实现相同的业务功能时,reduceByKey存在预聚和(combiner)功能,所以性能比较高,推荐使用。前提是不影响业务逻辑,比如求平均值就不能区内求平均值
aggregateByKey
将数据根据不同的规则进行分区内计算和分区间计算
可以设置输出分区数,和自定义分区器
函数签名
//这是个柯理化函数。zeroValue:传入的初始值,
//(U,V)=>U:分区内计算,U就是zeroVlaue的类型,对区内每个相同key的Value就进行操作,然后再重新赋值给zeroValue
//(U,U)=>U:分区间计算,分区间相同key的value进行操作。
def aggregateByKey[U: ClassTag](zeroValue: U)(seqOp: (U, V) => U,
combOp: (U, U) => U): RDD[(K, U)] = self.withScope {
aggregateByKey(zeroValue, defaultPartitioner(self))(seqOp, combOp)
}
示例1:达到和reduceByKey的相同的效果
val result3 = rdd.aggregateByKey(0)((zeroValue, value) => zeroValue + value, _+_)
result3.saveAsTextFile("output")
//result结果
//(1,3),(2,6),(3,3)
示例2:取出每个分区内相同key的最大值然后分区间相加
解析:用Int.minValue做初始值,在区内找出每个key的最大的value然后分区间相加。
val rdd = sc.makeRDD(List((1, 2), (2, 4),(2,5), (3, 2), (1,4),(1, 1), (2, 2), (3, 1)), 2)
val result4 = rdd.aggregateByKey(Int.MinValue)((zeroValue, value) => zeroValue.max(value), (_ + _))
result4.saveAsTextFile("output")
//result结果
//(1,6),(3,3),(2,7)
示例3:分区内同时求出每个key的最大值和最小值,分区间合并
//取出每个分区内相同key的最大值和最小值,然后分区间相加 用模式匹配
val result7 = rdd.aggregateByKey((Int.MinValue, Int.MaxValue))({
//分区内,参数:zeroValue=(MinInt,MaxInt) ,区内相同key的vlaue和zeroValue迭代计算 ,返回值:(key,(最大值,最小值))
case (zeroValue, value) => (zeroValue._1.max(value), zeroValue._2.min(value))
}, {
//分区间,参数:相同key的value,zero1=(max1,min1),zero2=(max2,min2)
case (zero1, zero2) => (zero1._1+zero2._1,zero1._2+zero2._2)
})
foldByKey
aggregateByKey的特例
当分区内计算规则和分区间计算规则相同时,aggregateByKey就可以简化为foldByKey
示例:达到和reduceByKey的相同的效果
val rdd = sc.makeRDD(List((1, 2), (2, 4), (2, 5), (3, 2), (1, 4), (1, 1), (2, 2), (3, 1)), 2)
val result5 = rdd.foldByKey(0)(_ + _)
//result5结果
//(2,11),(1,7),(3,3)
combineByKey
def combineByKey[C](
createCombiner: V => C, //就是zeroValue初始值
mergeValue: (C, V) => C, //分区内计算规则
mergeCombiners: (C, C) => C): RDD[(K, C)] //分区间计算规则
aggregateByKey的增强版,aggregateByKey的zeroValue必须指定固定值,但是combineByKey的zeroValue可以通过key-value的第一个value获取。
示例:将数据List(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98))求每个key的平均值
val rdd2 = sc.makeRDD(List(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98)), 2)
val result8 = rdd2.combineByKey(
//1.初始值为第一个value构建出来,后续的计算从第二个value开始 ,V=>C
value => (value, 1),
//2.分区内计算,(C,V)=>C
(zeroValue: Tuple2[Int, Int], value: Int) => (zeroValue._1 + value, zeroValue._2 + 1),
//3.分区间计算,(C,C)=>C ,需要保证返回值和输入类型一致
(zero1: Tuple2[Int, Int], zero2: Tuple2[Int, Int]) => (zero1._1 + zero2._1, zero1._2 + zero2._2)
)
val result9 = result8.map {
case (key, value) => (key, value._1.toDouble / value._2)
}
result9.saveAsTextFile("output")
//result结果:(a,91.33333333333333),(b,95.33333333333333)
sortByKey
只能针对key对k-v数据进行排序,有shuffle
函数签名
//升序:true,降序:false
def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length)
: RDD[(K, V)]
join
类似SQL的内连接,左外连接,右外连接,满连接。
inner join、left join、right join、full join
cogroup
有shuffle,RDD内先分组,RDD间按照key相同在进行分组。
示例
val rdd2 = sc.makeRDD(List(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98)), 2)
val rdd5 = sc.makeRDD(List(("a", 18), ("b", 15), ("a", 71), ("b", 53), ("a", 45), ("b", 78)))
rdd2.cogroup(rdd5).coalesce(1).saveAsTextFile("output")
结果
(a,(CompactBuffer(88, 91, 95),CompactBuffer(18, 71, 45)))
(b,(CompactBuffer(95, 93, 98),CompactBuffer(15, 53, 78)))
四.行动算子
区分一个算子是转换算子还是行动算子?
RDD ---> 转换算子 ----> RDD: 返回RDD的算子是转换算子!
RDD ----> 行动算子 ---> 返回的一定不是RDD: 不返回RDD的算子就是行动算子!
转换算子一般是懒执行、行动算子,触发Job的提交运行!
特殊情况
sortBy、sortByKey
底层sortBy-->sortByKey-->new Rangpartitioner-->collect返回边界数组。
只要有调用RangPartitioner就会触发Job提交!
行动算子种类
算子 | 解释 | 备注 |
---|---|---|
reduce | 将RDD中的元素通过函数进行聚合 | |
collect | 将RDD的元素以Array形式返回到Driver端 | 如果数据量过大,此时Driver端就OOM。collect适合少量数据的返回! |
count | 返回RDD元素的个数 | |
max | 返回RDD元素的最大值 | |
min | 返回RDD元素的最小值 | |
avg | 返回RDD元素的平均值 | |
sum | 返回RDD元素的和 | |
first | 返回RDD元素的第一个元素 | |
take(num:Int) | 返回前N个元素 | 如果数据量过大,此时Driver端就OOM。适合少量数据的返回! |
takeOrdered | 返回排序后的前N个元素 | 如果数据量过大,此时Driver端就OOM。适合少量数据的返回! |
aggregate | 聚合 | zeroValue不仅在分区内参与运算,还在分区间参与运算 |
fold | aggregate的简化版 | zeroValue必须是T类型,分区内和分区间运算的函数一致! |
countByValue | 对RDD中的每个元素的个数统计 | |
countByKey | 对K-V类型的RDD中的每个key,统计k-v对的个数 |