SparkCore

sparkcore · 浏览次数 : 17

小编点评

**RDD阶段划分** * DAG(Directed Acyclic Graph)有向无环图是由点和线组成的拓扑图形。该图形具有方向,不会闭环。 * 应用程序、 job、 stage 和任务 Application之间的关系是 1 对 n。 **RDD 通过 Cache 或者 Persist 方法将前面的计算结果缓存** * 如果缓存丢失,或者存储于内存的数据由于内存不足而被删除,RDD 的缓存容错机制保证了即使缓存丢失也能保证计算的正确执行。 * 通过基于 RDD 的一系列转换,丢失的数据会被重算,由于 RDD 的各个 Partition 是相对独立的,因此只需要计算丢失的部分即可,并不需要重算全部 Partition。 **RDD 的检查点** * 在实际使用的时候,如果想重用数据,仍然建议调用 persist 或 cache。 * 检查点切断血缘依赖(执行过程中会切断血缘关系,重新建立新的血缘关系,等同于改变了数据源)。 * Cache 缓存的数据通常存储在磁盘、内存等地方,可靠性低。 **checkpoint** * checkpoint 将数据长久地保存在磁盘文件中进行数据重用,涉及到磁盘IO,性能较低,但是数据安全。 * 为了能够提高效率,一般情况下需要和cache联合使用(先用cache,再紧跟着用checkpoint)。

正文

SparkCore

RDD基础

定义

​ 在 Spark 的编程接口中,每一个数据集都被表示为一个对象,称为 RDD。RDD 是 Resillient Distributed Dataset(弹性分布式数据集)的简称,是一个只读的(不可变的)、分区的(分布式的)、容错的、延迟计算的、类型推断的和可缓存的记录集合。

结构

​ RDD 由以下五部分组成:

  • 一组 partition(分区),即组成整个数据集的块;

  • 每个 partition(分区)的计算函数(用于计算数据集中所有行的函数);

  • 所依赖的 RDD 列表(即父 RDD 列表);

  • (可选的)对于 key-value 类型的 RDD,则包含一个 Partitioner(默认是 HashPartitioner);

  • (可选的)每个 partition 数据驻留在集群中的位置(可选);如果数据存放在 HDFS 上,那么它就是块所在的位置。

特性

  • RDD 会优先使用内存;

  • 一旦创建不可修改;

  • 惰性执行;

  • 可缓存,可复用;

  • 可并行处理;

  • 强类型,单一类型数据;

  • 分区的;

  • 可指定分区优先使用的节点。

SC与SS

​ SparkContext 从 Spark 1.x 引入的,在 2.0 中引入 SparkSession之前,用来作为 Spark 和 PySpark 的入口点。使用 RDD 编程和连接到 Spark Cluster 的第一步就是创建SparkContext。SparkContext 是在 org.apache.spark 包中定义的,它用于在集群中通过编程方式创建 SparkRDD、累加器和广播变量。

​ 虽然 SparkContext 是 2.0 之前一个入口点,但它并没有被 SparkSession 完全取代,SparkContext 的许多特性仍然可用,并在 Spark 2.0 和以后的版本中使用。

​ SparkContext 构造函数在 2.0 中已经弃用,因此建议使用静态方法 getOrCreate()来创建 SparkContext。该函数用于获取或实例SparkContext,并将其注册为一个单例对象。

val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
//val sc = new SparkContext(sparkConf)
val sc = SparkContext.getOrCreate(sparkConf)

创建

  1. 从集合(内存)中创建RDD。Spark主要提供了两个方法:parallelize和makeRDD,从底层代码实现来讲,makeRDD 方法其实就是 parallelize 方法。
val rdd1 = sc.parallelize(
 List(1,2,3,4)
)

val rdd2 = sc.makeRDD(
 List(1,2,3,4)
)
  1. 从外部存储(文件)创建RDD。由外部存储系统的数据集创建 RDD 包括:本地的文件系统,所有 Hadoop 支持的数据集,比如 HDFS、HBase 等。
val fileRDD: RDD[String] = sc.textFile("input")
  1. 从已有的RD转换得到新的RDD。
// 字符转为大写,得到一个新的 RDD
val rdd5 = rdd4.map(line => line.toUpperCase)

并行度与分区

​ 默认情况下,Spark 可以将一个作业切分多个任务后,发送给 Executor 节点并行计算,而能够并行计算的任务数量我们称之为并行度。分区的数量等于任务的数量,不一定等于并行度。分区数量可以在构建 RDD 时指定。

val dataRDD: RDD[Int] =
 sc.makeRDD(
 List(1,2,3,4),
 4)

​ 读取内存数据时,数据分区规则的Spark 核心源码如下。i为第几个分区(从零开始),length为集合长度,numSlices为分区数量。start和end为某一个分区内要存储的集合元素的范围,左闭右开。

val start = ((i*length) / numSlices).toInt
val end = (((i + 1)*length) / numSlices).toInt

​ 读取文件时,spark读取文件的方式和hadoop类似,一行一行读取。数据读取时以偏移量为单位,区分内存分区中以一个元素为单位,数据读取是全闭。偏移量不会重复读取。如果数据源为多个文件,那么计算分区时以文件为单位进行分区。具体Spark核心源码如下。totalSize表示读取文件的字节数的总和,goalSize表示每个分区应该存放多少个字节。如果除不尽的话,比较剩余字节数占goalSize的多少,如果大于%10,则需要产生新的分区,参考Hadoop。

long goalSize = totalSize / (numSplits == 0 ? 1 : numSplits) 

操作RDD

​ RDD 操作分为两种类型:转换(Transformation)和动作(action)。转换(Transformation)是用来创建RDD的方法,而动作(action)是使用RDD的方法。

转换算子

map

​ 将处理的数据逐条进行映射转换,这里的转换可以是类型的转换,也可以是值的转换。

val dataRDD: RDD[Int] = sparkContext.makeRDD(List(1,2,3,4))
val dataRDD1: RDD[Int] = dataRDD.map(
 num => {
 	num * 2
 }
)
val dataRDD2: RDD[String] = dataRDD1.map(
 num => {
 	"" + num
 }
)     

​ map 和 foreach 的区别?

  1. 前者是 transformation 操作(不会立即执行),后者是 action 操作(会立即执行);
  2. 前者返回值是一个新 RDD,后者没有返回值。

mapPartitions

​ 将待处理的数据以分区为单位发送到计算节点进行处理,这里的处理是指可以进行任意的处理,哪怕是过滤数据。

val dataRDD1: RDD[Int] = dataRDD.mapPartitions(
 datas => {
 datas.filter(x => x!=2)
 }
)

​ 思考一个问题:map 和 mapPartitions 的区别?

  • 数据处理角度

​ Map 算子是分区内一个数据一个数据的执行,类似于串行操作。而 mapPartitions 算子是以分区为单位进行批处理操作。

  • 功能的角度

​ Map 算子主要目的将数据源中的数据进行转换和改变。但是不会减少或增多数据。MapPartitions 算子需要传递一个迭代器,返回一个迭代器,没有要求的元素的个数保持不变,所以可以增加或减少数据

  • 性能的角度

​ Map 算子因为类似于串行操作,所以性能比较低,而mapPartitions 算子类似于批处理,所以性能较高。但是 mapPartitions 算子会长时间占用内存,那么这样会导致内存可能不够用,出现内存溢出的错误。所以在内存有限的情况下,不推荐使用。使用 map 操作。

mapPartitionsWithIndex

​ 将待处理的数据以分区为单位发送到计算节点进行处理,这里的处理是指可以进行任意的处理,哪怕是过滤数据,在处理时同时可以获取当前分区索引

val mpiRDD = rdd.mapPartitionsWithIndex(
	(index, iter) => {
		if ( index == 1 ) {
			iter
		} else {
			Nil.iterator
		}
	}
)

flatMap

​ 对集合中每个元素进行操作然后再扁平化。,所以算子也称之为扁平映射。

val dataRDD = sparkContext.makeRDD(List(
 List(1,2),List(3,4)
),1)
val dataRDD1 = dataRDD.flatMap(
 list => list
)

glom

​ 将同一个分区的数据直接转换为相同类型的内存数组进行处理,分区不变 。

val rdd : RDD[Int] = sc.makeRDD(List(1,2,3,4), 2)
val glomRDD: RDD[Array[Int]] = rdd.glom()
val maxRDD: RDD[Int] = glomRDD.map(
	array => {
		array.max
	}
)

groupBy

​ 将数据根据指定的规则进行分组, 分区默认不变,但是数据会被打乱重新组合,我们将这样的操作称之为 shuffle。极限情况下,数据可能被分在同一个分区中,一个组的数据在一个分区中,但是并不是说一个分区中只有一个组。

val rdd : RDD[Int] = sc.makeRDD(List(1,2,3,4), 2)
val groupRDD = rdd.groupBy(num => {num%2})

filter

​ 将数据根据指定的规则进行筛选过滤,符合规则的数据保留,不符合规则的数据丢弃。当数据进行筛选过滤后,分区不变,但是分区内的数据可能不均衡,生产环境下,可能会出现数据倾斜

val rdd = sc.makeRDD(List(1,2,3,4))
val filterRDD: RDD[Int] = rdd.filter(num=>num%2!=0)

sample

​ 根据指定的规则从数据集中抽取数据。

val dataRDD = sparkContext.makeRDD(List(
 1,2,3,4
),1)
// 抽取数据不放回(伯努利算法)
// 伯努利算法:又叫 0、1 分布。例如扔硬币,要么正面,要么反面。
// 具体实现:根据种子和随机算法算出一个数和第二个参数设置几率比较,小于第二个参数要,大于不要
// 第一个参数:抽取的数据是否放回,false:不放回
// 第二个参数:抽取的几率,范围在[0,1]之间,0:全不取;1:全取;
// 第三个参数:随机数种子
val dataRDD1 = dataRDD.sample(false, 0.5)
// 抽取数据放回(泊松算法)
// 第一个参数:抽取的数据是否放回,true:放回;false:不放回
// 第二个参数:重复数据的几率,范围大于等于 0.表示每一个元素被期望抽取到的次数
// 第三个参数:随机数种子
val dataRDD2 = dataRDD.sample(true, 2)

distinct

​ 将数据集中重复的数据去重。

val dataRDD = sparkContext.makeRDD(List(
 1,2,3,4,1,2
),1)
val dataRDD1 = dataRDD.distinct()

coalesce

​ 根据数据量缩减分区,用于大数据集过滤后,提高小数据集的执行效率,当 spark 程序中,存在过多的小任务的时候,可以通过 coalesce 方法,收缩合并分区,减少分区的个数,减小任务调度成本。

val dataRDD = sparkContext.makeRDD(List(
 1,2,3,4,1,2
),6)
val dataRDD1 = dataRDD.coalesce(2)

repartition

​ 该操作内部其实执行的是 coalesce 操作,参数 shuffle 的默认值为 true。无论是将分区数多的RDD 转换为分区数少的 RDD,还是将分区数少的 RDD 转换为分区数多的 RDD,repartition操作都可以完成,因为无论如何都会经 shuffle 过程。repartition常用于增加分区,coalesce常用于减小分区

val dataRDD = sparkContext.makeRDD(List(
 1,2,3,4,1,2
),2)
val dataRDD1 = dataRDD.repartition(4)

sortBy

​ 该操作用于排序数据。在排序之前,可以将数据通过 f 函数进行处理,之后按照 f 函数处理的结果进行排序,默认为升序排列。排序后新产生的 RDD 的分区数与原 RDD 的分区数一致。中间存在 shuffle 的过程。

val rdd = sc.makeRDD(List(("1", 1), ("11", 2), ("2", 3)), 2)
val newRDD = rdd.sortBy(t=>t._1.toInt, false)

intersection

​ 对源 RDD 和参数 RDD 求交集后返回一个新的 RDD。

val dataRDD1 = sparkContext.makeRDD(List(1,2,3,4))
val dataRDD2 = sparkContext.makeRDD(List(3,4,5,6))
val dataRDD = dataRDD1.intersection(dataRDD2)

union

​ 对源 RDD 和参数 RDD 求并集后返回一个新的 RDD。

val dataRDD1 = sparkContext.makeRDD(List(1,2,3,4))
val dataRDD2 = sparkContext.makeRDD(List(3,4,5,6))
val dataRDD = dataRDD1.union(dataRDD2)

subtract

​ 以一个 RDD 元素为主,去除两个 RDD 中重复元素,将其他元素保留下来。求差集。

val dataRDD1 = sparkContext.makeRDD(List(1,2,3,4))
val dataRDD2 = sparkContext.makeRDD(List(3,4,5,6))
val dataRDD = dataRDD1.subtract(dataRDD2)

zip

​ 将两个 RDD 中的元素,以键值对的形式进行合并。其中,键值对中的 Key 为第 1 个 RDD中的元素,Value 为第 2 个 RDD 中的相同位置的元素。

val dataRDD1 = sparkContext.makeRDD(List(1,2,3,4))
val dataRDD2 = sparkContext.makeRDD(List(3,4,5,6))
val dataRDD = dataRDD1.subtract(dataRDD2)

partitionBy

​ 将数据按照指定 Partitioner 重新进行分区。Spark 默认的分区器是 HashPartitioner。

val rdd: RDD[(Int, String)] =
 sc.makeRDD(Array((1,"aaa"),(2,"bbb"),(3,"ccc")),3)
val rdd2: RDD[(Int, String)] =
 rdd.partitionBy(new HashPartitioner(2))

reduceByKey

​ 可以将数据按照相同的 Key 对 Value 进行聚合。

val rdd = sc.makeRDD(List(
	("a", 1), ("a", 2), ("a", 3), ("b", 4)
))
val reduceRDD: RDD[(String, Int)] = rdd.reduceByKey( (x:Int, y:Int) => {
	x + y
} )

groupByKey

​ 将数据源的数据根据 key 对 value 进行分组。

val rdd = sc.makeRDD(List(
	("a", 1), ("a", 2), ("a", 3), ("b", 4)
))
val groupRDD: RDD[(String, Iterable[Int])] = rdd.groupByKey()
//打印
//(a,CompactBuffer(1, 2, 3))
//(b,CompactBuffer(4))
  1. 思考一个问题:reduceByKey 和 groupByKey 的区别?
  • 从 shuffle的角度

​ reduceByKey 和 groupByKey 都存在 shuffle 的操作,但是 reduceByKey可以在 shuffle 前对分区内相同 key 的数据进行预聚合(combine)功能,这样会减少落盘的数据量,而 groupByKey 只是进行分组,不存在数据量减少的问题,reduceByKey 性能比较高。

  • 从功能的角度

​ reduceByKey 其实包含分组和聚合的功能。GroupByKey 只能分组,不能聚合,所以在分组聚合的场合下,推荐使用 reduceByKey,如果仅仅是分组而不需要聚合。那么还是只能使用 groupByKey。

  1. 思考一个问题:groupBy 和 groupByKey 的区别?

    (1)后者只能针对KV类型的数据进行分组,而前者没有限制.

​ (2)后者的返回值是一个迭代器,且迭代器中的元素是value,而前者迭代器中的元素没有改变。

​ (3)后者只能针对Key进行分组,前者没有限制。

aggregateByKey

​ 将数据根据不同的规则进行分区内计算和分区间计算。

val rdd = sc.makeRDD(List(
	("a", 1), ("a", 2), ("a", 3), ("a", 4)
),2)
// 第一个参数列表,需要传递一个参数,表示为初始值
// 	    主要用于当碰见第一个key的时候,和value进行分区内计算
// 第二个参数列表需要传递2个参数
//      第一个参数表示分区内计算规则
//      第二个参数表示分区间计算规则
rdd.aggregateByKey(10)(
	(x, y) => math.max(x, y),
	(x, y) => x + y
).collect.foreach(println)

foldByKey

​ 当分区内计算规则和分区间计算规则相同时,aggregateByKey 就可以简化为 foldByKey。

val rdd = sc.makeRDD(List(
	("a", 1), ("a", 2), ("b", 3),
	("b", 4), ("b", 5), ("a", 6)
),2)
rdd.foldByKey(0)((x, y) => x+y).collect.foreach(println)

combineByKey

​ 最通用的对 key-value 型 rdd 进行聚集操作的聚集函数(aggregation function)。

val rdd = sc.makeRDD(List(
	("a", 1), ("a", 2), ("b", 3),
	("b", 4), ("b", 5), ("a", 6)
),2)
// combineByKey : 方法需要三个参数
// 第一个参数表示:将相同key的第一个数据进行结构的转换,实现操作
// 第二个参数表示:分区内的计算规则
// 第三个参数表示:分区间的计算规则
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)
	}
)

​ 思考一个问题:reduceByKey、foldByKey、aggregateByKey、combineByKey 的区别?

  • reduceByKey

​ 相同 key 的第一个数据不进行任何计算,分区内和分区间计算规则相同。

  • FoldByKey

​ 相同 key 的第一个数据和初始值进行分区内计算,分区内和分区间计算规则相同 。

  • AggregateByKey

​ 相同 key 的第一个数据和初始值进行分区内计算,分区内和分区间计算规则可以不相同。

  • CombineByKey

​ 当计算时,发现数据结构不满足要求时,可以让第一个数据转换结构。分区内和分区间计算规则不相同。

sortByKey

​ 在一个(K,V)的 RDD 上调用,K 必须实现 Ordered 接口(特质),返回一个按照 key 进行排序的。

val dataRDD1 = sparkContext.makeRDD(List(("a",1),("b",2),("c",3)))
val sortRDD1: RDD[(String, Int)] = dataRDD1.sortByKey(true)
val sortRDD1: RDD[(String, Int)] = dataRDD1.sortByKey(false)

join

​ 在类型为(K,V)和(K,W)的 RDD 上调用,返回一个相同 key 对应的所有元素连接在一起的。

val rdd1 = sc.makeRDD(List(
	("a", 1), ("a", 2), ("c", 3)
))
val rdd2 = sc.makeRDD(List(
	("a", 5), ("c", 6),("a", 4)
))
val joinRDD: RDD[(String, (Int, Int))] = rdd1.join(rdd2)
joinRDD.collect().foreach(println)
// (a,(1,5))
// (a,(1,4))
// (a,(2,5))
// (a,(2,4))
// (c,(3,6))

leftOuterJoin

​ 类似于 SQL 语句的左外连接 。

val rdd1 = sc.makeRDD(List(
	("a", 1), ("b", 2)
))
val rdd2 = sc.makeRDD(List(
	("a", 4), ("b", 5), ("b", 3), ("c", 6)
))
val rightJoinRDD = rdd1.rightOuterJoin(rdd2)
rightJoinRDD.collect().foreach(println)
// (a,(Some(1),4))
// (b,(Some(2),5))
// (b,(Some(2),3))
// (c,(None,6))

cogroup

​ 在类型为(K,V)和(K,W)的 RDD 上调用,返回一个(K,(Iterable,Iterable))类型的 RDD。

val rdd1 = sc.makeRDD(List(
	("a", 1), ("b", 2)
))
val rdd2 = sc.makeRDD(List(
	("a", 4), ("b", 5), ("c", 7), ("c", 6)
))
val cgRDD: RDD[(String, (Iterable[Int], Iterable[Int]))] = rdd1.cogroup(rdd2)
cgRDD.collect().foreach(println)
// (a,(CompactBuffer(1),CompactBuffer(4)))
// (b,(CompactBuffer(2),CompactBuffer(5)))
// (c,(CompactBuffer(),CompactBuffer(6, 7)))

动作算子

reduce

​ 聚集 RDD 中的所有元素,先聚合分区内数据,再聚合分区间数据。

val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
// 聚合数据
val reduceResult: Int = rdd.reduce((x, y) => x+y)

collect

​ 在驱动程序中,以数组 Array 的形式返回数据集的所有元素。

val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
rdd.collect().foreach(println)

count

​ 返回 RDD 中元素的个数。

val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
val countResult: Long = rdd.count()

first

​ 返回 RDD 中的第一个元素。

val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
val firstResult: Int = rdd.first()
println(firstResult)

take

​ 返回一个由 RDD 的前 n 个元素组成的数组。

vval rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
val takeResult: Array[Int] = rdd.take(2)
println(takeResult.mkString(","))

takeOrdered

​ 返回该 RDD 排序后的前 n 个元素组成的数组。

import org.apache.spark.{SparkConf, SparkContext}
object RddTakeOrderedDemo {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("RddTakeOrderedDemo")
    val sc = new SparkContext(conf)
    val rdd = sc.makeRDD(1 to 10)
    val ord = new MyOrdering()
    rdd.takeOrdered(3)(ord).foreach(println)
    sc.stop()
  }
 
  class MyOrdering extends Ordering[Int] {
    override def compare(x: Int, y: Int): Int = {
      if (x < y) 1 else if (x == y) 0 else -1
    }
  }
}

aggregate

​ 分区的数据通过初始值和分区内的数据进行聚合,然后再和初始值进行分区间的数据聚合。

val rdd = sc.makeRDD(List(1,2,3,4),2)
val result = rdd.aggregate(10)((x, y) => x+y, (x, y) => x+y)
// 40

fold

​ 折叠操作,aggregate 的简化版操作。

val rdd = sc.makeRDD(List(1,2,3,4),2)
val result = rdd.fold(10)((x, y) => x+y)
// 40

countByKey

​ 统计每种 key 的个数。

val rdd = sc.makeRDD(List(
	("a", 1),("a", 2),("a", 3)
))
val stringToLong: collection.Map[String, Long] =rdd.countByKey()
println(stringToLong)

countByValue

​ 和countByKey类似。

def wordcount8(sc : SparkContext): Unit = {
    val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
    val words = rdd.flatMap(_.split(" "))
    val wordCount: collection.Map[String, Long] = words.countByValue()
}

mapValues

​ 不管Key,直接计算value。

val resultRDD: RDD[(String, Int)] = newRDD.mapValues {
    case (num, cnt) => {
        num / cnt
    }
}

RDD序列化

​ 从计算的角度, 算子以外的代码都是在 Driver 端执行, 算子里面的代码都是在 Executor端执行。那么在 scala 的函数式编程中,就会导致算子内经常会用到算子外的数据,这样就形成了闭包的效果,如果使用的算子外的数据无法序列化,就意味着无法传值给 Executor端执行,就会发生错误,所以需要在执行任务计算前,检测闭包内的对象是否可以进行序列化,这个操作我们称之为闭包检测

import org.apache.spark.{SparkConf, SparkContext}

object Spark07_RDD_Operator_Action {

    def main(args: Array[String]): Unit = {

        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
        val sc = new SparkContext(sparkConf)

        val rdd = sc.makeRDD(List[Int](1,2,3))

        val user = new User()

        // SparkException: Task not serializable
        // NotSerializableException: com.atguigu.bigdata.spark.core.rdd.operator.action.Spark07_RDD_Operator_Action$User

        // RDD算子中传递的函数是会包含闭包操作,那么就会进行检测功能
        // 闭包检测
        rdd.foreach(
            num => {
                println("age = " + (user.age + num))
            }
        )

        sc.stop()

    }
    //class User extends Serializable {
    // 样例类在编译时,会自动混入序列化特质(实现可序列化接口)
    //case class User() {
    class User {
        var age : Int = 30
    }
}

​ Java 的序列化能够序列化任何的类。但是比较重(字节多),序列化后,对象的提交也比较大。Spark 出于性能的考虑,Spark2.0 开始支持另外一种 Kryo 序列化机制。Kryo 速度是 Serializable 的 10 倍。当 RDD 在 Shuffle 数据的时候,简单数据类型、数组和字符串类型已经在 Spark 内部使用 Kryo 来序列化。注意:即使使用 Kryo 序列化,也要继承 Serializable 接口。

object serializable_Kryo {
 def main(args: Array[String]): Unit = {
 val conf: SparkConf = new SparkConf()
 .setAppName("SerDemo")
 .setMaster("local[*]")
 // 替换默认的序列化机制
 .set("spark.serializer", 
"org.apache.spark.serializer.KryoSerializer")
 // 注册需要使用 kryo 序列化的自定义类
 .registerKryoClasses(Array(classOf[Searcher]))
 val sc = new SparkContext(conf)
 val rdd: RDD[String] = sc.makeRDD(Array("hello world", "hello atguigu", 
"atguigu", "hahah"), 2)
 val searcher = new Searcher("hello")
 val result: RDD[String] = searcher.getMatchedRDD1(rdd)
 result.collect.foreach(println)
 }
}
case class Searcher(val query: String) {
 def isMatch(s: String) = {
 s.contains(query)
 }
 def getMatchedRDD1(rdd: RDD[String]) = {
 rdd.filter(isMatch) 
 }
 def getMatchedRDD2(rdd: RDD[String]) = {
 val q = query
 rdd.filter(_.contains(q))
 }
}

RDD依赖关系

​ 相邻的两个RDD的关系称之为依赖关系,多个连续的RDD的依赖关系,称之为血缘关系。

宽依赖:宽依赖表示同一个父(上游)RDD 的 Partition 被多个子(下游)RDD 的 Partition 依赖,会引起 Shuffle,总结:宽依赖我们形象的比喻为多生。

窄依赖:窄依赖表示每一个父(上游)RDD 的 Partition 最多被子(下游)RDD 的一个 Partition 使用,窄依赖我们形象的比喻为独生子女。

RDD阶段划分:DAG(Directed Acyclic Graph)有向无环图是由点和线组成的拓扑图形,该图形具有方向,不会闭环。例如,DAG 记录了 RDD 的转换过程和任务的阶段。

任务划分:RDD 任务切分中间分为:Application、Job、Stage 和 Task

  • Application:初始化一个 SparkContext 即生成一个 Application;

  • Job:一个 Action 算子就会生成一个 Job;

  • Stage:Stage 等于宽依赖(ShuffleDependency)的个数加 1;

  • Task:一个 Stage 阶段中,最后一个 RDD 的分区个数就是 Task 的个数。

注意:Application->Job->Stage->Task 每一层都是 1 对 n 的关系。

RDD持久化

Cache、Persist缓存

​ RDD的对象可以重用,但是数据无法重用,如果一个RDD需要重复使用,那么根据血缘关系,需要从头再次执行来获取数据。

​ RDD 通过 Cache 或者 Persist 方法将前面的计算结果缓存,默认情况下会把数据以缓存在 JVM 的堆内存中。但是并不是这两个方法被调用时立即缓存,而是触发后面的 action 算子时,该 RDD 将会被缓存在计算节点的内存中,并供后面重用。

val list = List("Hello Scala", "Hello Spark")
val rdd = sc.makeRDD(list)
val flatRDD = rdd.flatMap(_.split(" "))
val mapRDD = flatRDD.map(word=>{
	(word,1)
})
mapRDD.persist(StorageLevel.DISK_ONLY)
val reduceRDD: RDD[(String, Int)] = mapRDD.reduceByKey(_+_)
reduceRDD.collect().foreach(println)
println("**************************************")
val groupRDD = mapRDD.groupByKey()
groupRDD.collect().foreach(println)

​ 存储级别如下如所示。

​ 缓存有可能丢失,或者存储于内存的数据由于内存不足而被删除,RDD 的缓存容错机制保证了即使缓存丢失也能保证计算的正确执行。通过基于 RDD 的一系列转换,丢失的数据会被重算,由于 RDD 的各个 Partition 是相对独立的,因此只需要计算丢失的部分即可,并不需要重算全部 Partition。

​ Spark 会自动对一些 Shuffle 操作的中间数据做持久化操作(比如:reduceByKey)。这样做的目的是为了当一个节点 Shuffle 失败了避免重新计算整个输入。但是,在实际使用的时候,如果想重用数据,仍然建议调用 persist 或 cache。

CheckPoint检查点

​ 所谓的检查点其实就是通过将 RDD 中间结果写入磁盘。由于血缘依赖过长会造成容错成本过高,这样就不如在中间阶段做检查点容错,如果检查点之后有节点出现问题,可以从检查点开始重做血缘,减少了开销。对 RDD 进行checkpoint 操作并不会马上被执行,必须执行 Action 操作才能触发。

// 设置检查点路径
sc.setCheckpointDir("./checkpoint1")
// 创建一个 RDD,读取指定位置文件:hello atguigu atguigu
val lineRdd: RDD[String] = sc.textFile("input/1.txt")
// 业务逻辑
val wordRdd: RDD[String] = lineRdd.flatMap(line => line.split(" "))
val wordToOneRdd: RDD[(String, Long)] = wordRdd.map {
 word => {
 (word, System.currentTimeMillis())
 }
}
// 增加缓存,避免再重新跑一个 job 做 checkpoint
wordToOneRdd.cache()
// 数据检查点:针对 wordToOneRdd 做检查点计算
wordToOneRdd.checkpoint()
// 触发执行逻辑
wordToOneRdd.collect().foreach(println)

区别

  1. cache将数据临时存储在内存中进行数据重用。

  2. persist将数据临时存储在磁盘文件中进行数据重用,涉及到磁盘IO,性能较低,但是数据安全,如果作业执行完毕,临时保存的数据文件就会丢失。

  3. checkpoint将数据长久地保存在磁盘文件中进行数据重用,涉及到磁盘IO,性能较低,但是数据安全,为了保证数据安全,所一般情况下会独立执行作业。为了能够提高效率,一般情况下需要和cache联合使用(先用cache,再紧跟着用checkpoint)。

  • Cache 缓存只是将数据保存起来,不切断血缘依赖(会在血缘关系中添加新的依赖)。Checkpoint 检查点切断血缘依赖(执行过程中会切断血缘关系,重新建立新的血缘关系,等同于改变了数据源)。

  • Cache 缓存的数据通常存储在磁盘、内存等地方,可靠性低。Checkpoint 的数据通常存储在 HDFS 等容错、高可用的文件系统,可靠性高。

  • 建议对 checkpoint()的 RDD 使用 Cache 缓存,这样 checkpoint 的 job 只需从 Cache 缓存中读取数据即可,否则需要再从头计算一次 RDD。

与SparkCore相似的内容: