初始化工作是在Driver端进行的,而实际运行程序是在Executor端进行的,这就涉及到了跨进程通信,是需要序列化的。
class User extends Serializable {
var name: String = _
}
class Test04 {
Logger.getLogger("org").setLevel(Level.ERROR)
@Test
def test(): Unit = {
val conf: SparkConf = new SparkConf().setAppName("SparkCore").setMaster("local[*]")
val sc: SparkContext = new SparkContext(conf)
val rdd01: RDD[(Int, String)] = sc.makeRDD(Array((111, "aaa"), (222, "bbbb"), (333, "ccccc")), 3)
val user01: User = new User()
user01.name = "list"
val user02: User = new User()
user02.name = "lisi"
val userRdd01: RDD[User] = sc.makeRDD(List(user01, user02))
// 没有序列化(java.io.NotSerializableException: day04.User)
userRdd01.foreach(user => println(user.name))
sc.stop()
}
}
参考地址: https://github.com/EsotericSoftware/kryo
Java的序列化能够序列化任何的类。但是比较重,序列化后对象的体积也比较大。
Spark出于性能的考虑,Spark2.0开始支持另外一种Kryo序列化机制。Kryo速度是Serializable的10倍。当RDD在Shuffle数据的时候,简单数据类型、数组和字符串类型已经在Spark内部使用Kryo来序列化。
import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.junit.Test
class Test04 {
Logger.getLogger("org").setLevel(Level.ERROR)
@Test
def test(): Unit = {
val conf: SparkConf = new SparkConf().setAppName("SparkCore").setMaster("local[*]")
// 替换默认的序列化机制
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
// 注册需要使用kryo序列化的自定义类
.registerKryoClasses(Array(classOf[Search]))
val sc: SparkContext = new SparkContext(conf)
val rdd: RDD[String] = sc.makeRDD(Array("hello world", "hello", "world"))
val search: Search = new Search("hello")
val result: RDD[String] = rdd.filter(search.isMatch)
println(result.collect().toList)
}
}
// 关键字封装在一个类里面
// 需要自己先让类实现序列化 之后才能替换使用kryo序列化
class Search(val query: String) extends Serializable {
def isMatch(s: String): Boolean = {
s.contains(query)
}
}
RDD只支持粗粒度转换,即在大量记录上执行的单个操作。将创建RDD的一系列Lineage(血统)记录下来,以便恢复丢失的分区。RDD的Lineage会记录RDD的元数据信息和转换行为,当该RDD的部分分区数据丢失时,它可以根据这些信息来重新运算和恢复丢失的数据分区。
rdd03.toDebugString
@Test
def test(): Unit = {
val conf: SparkConf = new SparkConf().setAppName("SparkCore").setMaster("local[*]")
val sc: SparkContext = new SparkContext(conf)
val rdd01: RDD[String] = sc.textFile("input/1.txt")
println(rdd01.toDebugString)
println("rdd01===")
val rdd02: RDD[String] = rdd01.flatMap(_.split(" "))
println(rdd02.toDebugString)
println("rdd02====")
val rdd03: RDD[(String, Int)] = rdd02.map((_, 1))
println(rdd03.toDebugString)
println("rdd03====")
val rdd04: RDD[(String, Int)] = rdd03.reduceByKey(_ + _)
println(rdd04.toDebugString)
sc.stop()
}
RDD之间的关系可以从两个维度来理解:一个是RDD是从哪些RDD转换而来,也就是 RDD的parent RDD(s)是什么(血缘); 另一个就是RDD依赖于parent RDD(s)的哪些Partition(s),这种关系就是RDD之间的依赖(依赖)。
RDD和它依赖的父RDD(s)的依赖关系有两种不同的类型,即窄依赖(NarrowDependency)和宽依赖(ShuffleDependency)。
一对一、多对一
一对多,会引起Shuffle
DAG有向无环图
DAG(Directed Acyclic Graph)有向无环图是由点和线组成的拓扑图形,该图形具有方向,不会闭环。
DAG记录了RDD的转换过程和任务的阶段。
RDD任务切分
RDD任务切分中间分为:Application、Job、Stage和Task
Application
:初始化一个SparkContext即生成一个Application;
Job
:一个Action算子就会生成一个Job;
Stage
:Stage等于宽依赖的个数加1;
Task
:一个Stage阶段中,最后一个RDD的分区个数就是Task的个数。
注意:Application
->Job
->Stage
->Task
每一层都是1对n的关系。
@Test
def Test(): Unit = {
val conf: SparkConf = new SparkConf().setAppName("SparkCore").setMaster("local[*]")
// 1、Application:初始化一个SparkContext即生成一个Application
val sc: SparkContext = new SparkContext(conf)
val lineRdd: RDD[String] = sc.textFile("input/1.txt")
val rdd01: RDD[String] = lineRdd.flatMap(_.split(" "))
val rdd02: RDD[(String, Int)] = rdd01.map((_, 1))
// 3、Stage:reduceByKey算子会有宽依赖,stage阶段+1。一共2个stage
val resultRdd: RDD[(String, Int)] = rdd02.reduceByKey(_ + _)
// 2、Job:一个Action算子就会生成一个Job。一共2个Job
resultRdd.collect().foreach(println)
resultRdd.saveAsTextFile("output")
Thread.sleep(Long.MaxValue)
sc.stop()
}
Application个数:
// 1、Application:初始化一个SparkContext即生成一个Application
val sc: SparkContext = new SparkContext(conf)
Job个数
Stage个数
Task数量
RDD通过Cache或者Persist方法将前面的计算结果缓存,默认情况下会把数据以序列化的形式缓存在JVM的堆内存中。但是并不是这两个方法被调用时立即缓存,而是触发后面的action算子时,该RDD将会被缓存在计算节点的内存中,并供后面重用。
// cache底层调用的就是persist方法,缓存级别默认用的是MEMORY_ONLY
wortToOneRdd.cache()
// 可以更改缓存级别
wortToOneRdd.persist(StorageLevel.MEMORY_AND_DISK_2)
案例:
val wordRdd: RDD[String] = lineRdd.flatMap(line => line.split(" "))
val wortToOneRdd: RDD[(String, Int)] = wordRdd.map(word => (word, 1))
// 打印血缘关系(缓存前)
println(wortToOneRdd.toDebugString)
// 数据缓存
// cache底层调用的就是persist方法,缓存级别默认用的是MEMORY_ONLY
wortToOneRdd.cache()
// 可以更改缓存级别
// wortToOneRdd.persist(StorageLevel.MEMORY_AND_DISK_2)
wortToOneRdd.collect().foreach(println)
// 打印血缘关系(缓存后)
println(wortToOneRdd.toDebugString)
缓存枚举参数
默认的存储级别都是仅在内存存储一份。在存储级别的末尾加上“_2”表示持久化的数据存为两份。
SER:表示序列化。
缓存有可能丢失,或者存储于内存的数据由于内存不足而被删除,RDD的缓存容错机制保证了即使缓存丢失也能保证计算的正确执行。通过基于RDD的一系列转换,丢失的数据会被重算,由于RDD的各个Partition是相对独立的,因此只需要计算丢失的部分即可,并不需要重算全部Partition。
自带缓存采用reduceByKey
// 采用reduceByKey,自带缓存
val wordByKeyRDD: RDD[(String, Int)] = wordToOneRdd.reduceByKey(_+_)
检查点:是通过将RDD中间结果写入磁盘
原因: 由于血缘依赖过长会造成容错成本过高,这样就不如在中间阶段做检查点容错,如果检查点之后有节点出现问题,可以从检查点开始重做血缘,减少了开销。
检查点存储路径: Checkpoint的数据通常是存储在HDFS等容错、高可用的文件系统。
存储格式为: 二进制的文件。
检查点切断血缘: 在Checkpoint的过程中,该RDD的所有依赖于父RDD中的信息将全部被移除。
检查点触发时间: 对RDD进行Checkpoint操作并不会马上被执行,必须执行Action操作才能触发。但是检查点为了数据安全,会从血缘关系的最开始执行一遍。
// 设置检查点数据存储路径:
sc.setCheckpointDir("./checkpoint1")
// 调用检查点方法:
wordToOneRdd.checkpoint()
代码:
val rdd: RDD[String] = sc.textFile("input/1.txt")
// 业务逻辑
val rdd01: RDD[String] = rdd.flatMap(line => line.split(" "))
val rdd02: RDD[(String, Long)] = rdd01.map(word => (word, System.currentTimeMillis()))
// 增加缓存,避免再重新跑一个job做checkpoint
rdd02.cache()
// 数据检查点:针对wordToOneRdd做检查点计算
rdd02.checkpoint()
// 会立即启动一个新的job来专门的做checkpoint运算(一共会有2个job)
rdd02.collect().foreach(println)
// 再次触发2次执行逻辑,用来对比
rdd02.collect().foreach(println)
rdd02.collect().foreach(println)
执行结果:
通过页面http://localhost:4040/jobs
查看DAG图。可以看到检查点切断了血缘依赖关系。
只增加checkpoint,没有增加Cache缓存打印 | 第1个job执行完,触发了checkpoint,第2个job运行checkpoint,并把数据存储在检查点上。第3、4个job,数据从检查点上直接读取。 |
增加checkpoint,也增加Cache缓存打印 | 第1个job执行完,数据就保存到Cache里面了,第2个job运行checkpoint,直接读取Cache里面的数据,并把数据存储在检查点上。第3、4个job,数据从检查点上直接读取。 |
如果检查点数据存储到HDFS集群,要注意配置访问集群的用户名。否则会报访问权限异常。
// 设置访问HDFS集群的用户名
System.setProperty("HADOOP_USER_NAME", "atguigu")
// 需要设置路径.需要提前在HDFS集群上创建/checkpoint路径
sc.setCheckpointDir("hdfs://hadoop102:8020/checkpoint")
// 数据检查点:针对wordToOneRdd做检查点计算
rdd02.checkpoint()
Spark目前支持Hash分区、Range分区和用户自定义分区。Hash分区为当前的默认分区。分区器直接决定了RDD中分区的个数、RDD中每条数据经过Shuffle后进入哪个分区和Reduce的个数。
val conf: SparkConf = new SparkConf().setAppName("SparkCore").setMaster("local[*]")
val sc: SparkContext = new SparkContext(conf)
// 数据源处理
val rdd: RDD[(Int, Int)] = sc.makeRDD(List((1, 1), (2, 2), (3, 3)))
// 打印分区器
println(rdd.partitioner)
// 使用HashPartitioner对RDD进行重新分区
val rdd02: RDD[(Int, Int)] = rdd.partitionBy(new HashPartitioner(2))
// 打印分区器
println(rdd02.partitioner)
sc.stop()
Hash分区
HashPartitioner分区的原理:对于给定的key,计算其hashCode,并除以分区的个数取余,如果余数小于0,则用余数+分区的个数(否则加0),最后返回的值就是这个key所属的分区ID。
HashPartitioner分区弊端:可能导致每个分区中数据量的不均匀,极端情况下会导致某些分区拥有RDD的全部数据。
Ranger分区
RangePartitioner作用:将一定范围内的数映射到某一个分区内,尽量保证每个分区中数据量均匀,而且分区与分区之间是有序的,一个分区中的元素肯定都是比另一个分区内的元素小或者大,但是分区内的元素是不能保证顺序的。简单的说就是将一定范围内的数映射到某一个分区内。
实现过程为:
第一步:先从整个RDD中采用水塘抽样算法,抽取出样本数据,将样本数据排序,计算出每个分区的最大key值,形成一个Array[KEY]类型的数组变量rangeBounds;
第二步:判断key在rangeBounds中所处的范围,给出该key值在下一个RDD中的分区id下标;该分区器要求RDD中的KEY类型必须是可以排序的
分布式共享只写变量(Executor和Executor之间不能读数据)
// 累加器定义(SparkContext.accumulator(initialValue)方法)
val sum: LongAccumulator = sc.longAccumulator("sum")
// 累加器添加数据(累加器.add方法)
sum.add(count)
// 累加器获取数据(累加器.value)
sum.value
val dataRdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("a", 3), ("a", 4)))
// 设置新的累加器
val accSum: LongAccumulator = sc.longAccumulator("sum")
dataRdd.foreach(line => {
// 使用累加器累加
accSum.add(line._2)
})
// 获取累加器,累加后的值
println(accSum.value)
累加器要放在行动算子中
val value: RDD[Unit] = dataRdd.map {
case (a, count) => {
accSum.add(count)
}
}
//假如放在map中,调用两次行动算子,map执行两次,导致最终累加器的值翻倍
mapRDD.collect()
mapRDD.collect()
分布式共享只读变量
广播变量用来高效分发较大的对象。向所有工作节点发送一个较大的只读值,以供一个或多个Spark Task操作使用。
步骤:
// 声明广播变量
val bdStr: Broadcast[Int] = sc.broadcast(num)
// 使用广播变量
bdstr.value
val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5, 6), 3)
// 需要广播的值
val num: Int = 1
// 声明广播变量
val bdStr: Broadcast[Int] = sc.broadcast(num)
// 使用广播变量
val rdd02: RDD[Int] = rdd.filter(lin => {
lin.equals(bdStr.value)
})
rdd02.foreach(println)