Spark-Core-阿里云开发者社区

一、RDD 编程

1、RDD序列化

初始化工作是在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()
  }
}

1.2 Kryo序列化框架

参考地址: 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)
  }
}

2、RDD依赖关系

2.1 查看血缘关系

RDD只支持粗粒度转换，即在大量记录上执行的单个操作。将创建RDD的一系列Lineage（血统）记录下来，以便恢复丢失的分区。RDD的Lineage会记录RDD的元数据信息和转换行为，当该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()
}

2.2 查看依赖关系

RDD之间的关系可以从两个维度来理解：一个是RDD是从哪些RDD转换而来，也就是 RDD的parent RDD(s)是什么（血缘）; 另一个就是RDD依赖于parent RDD(s)的哪些Partition(s)，这种关系就是RDD之间的依赖（依赖）。

RDD和它依赖的父RDD（s）的依赖关系有两种不同的类型，即窄依赖（NarrowDependency）和宽依赖（ShuffleDependency）。

2.3 窄依赖

一对一、多对一

窄依赖表示每一个父RDD的Partition最多被子RDD的一个Partition使用（一对一、多对一）。
窄依赖我们形象的比喻为独生子女。

2.4 宽依赖

一对多，会引起Shuffle

宽依赖表示同一个父RDD的Partition被多个子RDD的Partition依赖（只能是一对多），会引起Shuffle。
总结：宽依赖我们形象的比喻为超生。
具有宽依赖的transformations包括：sort、reduceByKey、groupByKey、join和调用rePartition函数的任何操作。
宽依赖对Spark去评估一个transformations有更加重要的影响，比如对性能的影响。
在不影响业务要求的情况下，要尽量避免使用有宽依赖的转换算子，因为有宽依赖，就一定会走shuffle，影响性能。

2.5 Stage任务划分

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数量

如果存在shuffle过程，系统会自动进行缓存，UI界面显示skipped的部分。
从Stage中看有2个Task。

3、RDD 持久化

3.1 Cache缓存

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(_+_)

3.2 CheckPoint检查点

检查点：是通过将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，数据从检查点上直接读取。

3.3 缓存和检查点区别

Cache缓存只是将数据保存起来，不切断血缘依赖。Checkpoint检查点切断血缘依赖。
Cache缓存的数据通常存储在磁盘、内存等地方，可靠性低。Checkpoint的数据通常存储在HDFS等容错、高可用的文件系统，可靠性高。
建议对checkpoint()的RDD使用Cache缓存，这样checkpoint的job只需从Cache缓存中读取数据即可，否则需要再从头计算一次RDD。
如果使用完了缓存，可以通过unpersist()方法释放缓存。

3.4 检查点存储到HDFS集群

如果检查点数据存储到HDFS集群，要注意配置访问集群的用户名。否则会报访问权限异常。

// 设置访问HDFS集群的用户名
System.setProperty("HADOOP_USER_NAME", "atguigu")
// 需要设置路径.需要提前在HDFS集群上创建/checkpoint路径
sc.setCheckpointDir("hdfs://hadoop102:8020/checkpoint")
//  数据检查点：针对wordToOneRdd做检查点计算
rdd02.checkpoint()

4、键值对RDD数据分区

Spark目前支持Hash分区、Range分区和用户自定义分区。Hash分区为当前的默认分区。分区器直接决定了RDD中分区的个数、RDD中每条数据经过Shuffle后进入哪个分区和Reduce的个数。

只有Key-Value类型的RDD才有分区器，非Key-Value类型的RDD分区的值是None
每个RDD的分区ID范围：0~numPartitions-1，决定这个值是属于那个分区的。

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类型必须是可以排序的

1）我们假设有100万条数据要分4个区
2）从100万条中抽100个数（1,2,3, …… 100）
3）对100个数进行排序，然后均匀的分为4段
4）获取100万条数据，每个值与4个分区的范围比较，放入合适分区

二、累加器

分布式共享只写变量（Executor和Executor之间不能读数据）

累加器用来把Executor端变量信息聚合到Driver端。在Driver中定义的一个变量，在Executor端的每个task都会得到这个变量的一份新的副本，每个task更新这些副本的值后，传回Driver端进行合并计算。
注意：Executor端的任务不能读取累加器的值（例如：在Executor端调用sum.value，获取的值不是累加器最终的值）。因此我们说，累加器是一个分布式共享只写变量。

//  累加器定义（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)

累加器要放在行动算子中

因为转换算子执行的次数取决于job的数量，如果一个spark应用有多个行动算子，那么转换算子中的累加器可能会发生不止一次更新，导致结果错误。
所以，如果想要一个无论在失败还是重复计算时都绝对可靠的累加器，我们必须把它放在foreach()这样的行动算子中。
对于在行动算子中使用的累加器，Spark只会把每个Job对各累加器的修改应用一次。

val value: RDD[Unit] = dataRdd.map {
  case (a, count) => {
    accSum.add(count)
  }
}
//假如放在map中，调用两次行动算子，map执行两次，导致最终累加器的值翻倍
mapRDD.collect()
mapRDD.collect()

三、广播变量

分布式共享只读变量

广播变量用来高效分发较大的对象。向所有工作节点发送一个较大的只读值，以供一个或多个Spark Task操作使用。

步骤：

调用SparkContext.broadcast（广播变量）创建出一个广播对象，任何可序列化的类型都可以这么实现。
通过广播变量.value，访问该对象的值。
广播变量只会被发到各个节点一次，作为只读值处理（修改这个值不会影响到别的节点）。

//  声明广播变量
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)

Spark-Core

一、RDD 编程

1、RDD序列化

1.2 Kryo序列化框架

2、RDD依赖关系

2.1 查看血缘关系

2.2 查看依赖关系

2.3 窄依赖

2.4 宽依赖

2.5 Stage任务划分

3、RDD 持久化

3.1 Cache缓存

3.2 CheckPoint检查点

3.3 缓存和检查点区别

3.4 检查点存储到HDFS集群

4、键值对RDD数据分区

二、累加器

三、广播变量

热门文章

最新文章

相关电子书

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

Spark-Core

一、RDD 编程

1、RDD序列化

1.2 Kryo序列化框架

2、RDD依赖关系

2.1 查看血缘关系

2.2 查看依赖关系

2.3 窄依赖

2.4 宽依赖

2.5 Stage任务划分

3、RDD 持久化

3.1 Cache缓存

3.2 CheckPoint检查点

3.3 缓存和检查点区别

3.4 检查点存储到HDFS集群

4、键值对RDD数据分区

二、累加器

三、广播变量

热门文章

最新文章

相关电子书