Spark Core快速入门系列(8) | RDD 的持久化

2020-10-28 16:58:32 浏览数 (1)

每碰到一个 Action 就会产生一个 job, 每个 job 开始计算的时候总是从这个 job 最开始的 RDD 开始计算.

一. 代码及结果

  • 1. 代码
代码语言：javascript复制
package Day04
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 **
@author 不温卜火
 **
 * @create 2020-07-26 13:57
 **
 *         MyCSDN :https://buwenbuhuo.blog.csdn.net/
 */
object CacheDemo {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("Practice").setMaster("local[2]")
    val sc = new SparkContext(conf)

    val rdd1 = sc.parallelize(Array("ab", "bc"))
    val rdd2 = rdd1.flatMap(x => {
      println("flatMap...")
      x.split("")
    })
    val rdd3: RDD[(String, Int)] = rdd2.map(x => {
      (x, 1)
    })

    rdd3.collect.foreach(println)
    println("-----------")
    rdd3.collect.foreach(println)
  }
}
  • 2. 结果

说明:

  1. 每调用一次 collect, 都会创建一个新的 job, 每个 job 总是从它血缘的起始开始计算. 所以, 会发现中间的这些计算过程都会重复的执行.
  2. 原因是因为 rdd记录了整个计算过程. 如果计算的过程中出现哪个分区的数据损坏或丢失, 则可以从头开始计算来达到容错的目的.

二. RDD 数据的持久化

  每个 job 都会重新进行计算, 在有些情况下是没有必要, 如何解决这个问题呢?   Spark 一个重要能力就是可以持久化数据集在内存中. 当我们持久化一个 RDD 时, 每个节点都会存储他在内存中计算的那些分区, 然后在其他的 action 中可以重用这些数据. 这个特性会让将来的 action 计算起来更快(通常块 10 倍). 对于迭代算法和快速交互式查询来说, 缓存(Caching)是一个关键工具.   可以使用方法persist()或者cache()来持久化一个 RDD. 在第一个 action 会计算这个 RDD, 然后把结果的存储到他的节点的内存中. Spark 的 Cache 也是容错: 如果 RDD 的任何一个分区的数据丢失了, Spark 会自动的重新计算.   RDD 的各个 Partition 是相对独立的, 因此只需要计算丢失的部分即可, 并不需要重算全部 Partition   另外, 允许我们对持久化的 RDD 使用不同的存储级别.   例如: 可以存在磁盘上, 存储在内存中(堆内存中), 跨节点做复本.   可以给persist()来传递存储级别. cache()方法是使用默认存储级别(StorageLevel.MEMORY_ONLY)的简写方法.

代码语言：javascript复制
// rdd2.cache() // 等价于 rdd2.persist(StorageLevel.MEMORY_ONLY)
  • rdd2.persist(StorageLevel.MEMORY_ONLY)

说明：

  1. 第一个 job 会计算 RDD2, 以后的 job 就不用再计算了.
  2. 有一点需要说明的是, 即使我们不手动设置持久化, Spark 也会自动的对一些 shuffle 操作的中间数据做持久化操作(比如: reduceByKey). 这样做的目的是为了当一个节点 shuffle 失败了避免重新计算整个输入. 当时, 在实际使用的时候, 如果想重用数据, 仍然建议调用persist 或 cache

  本次的分享就到这里了

spark 缓存 存储

0 人点赞