SparkRDD：知识点总结

SPARK 2017-01-18

学习目标

  • 掌握RDD是什么
  • 掌握RDD的五大特性
  • 掌握如何创建RDD
  • 掌握RDD的常用算子操作
  • 掌握RDD的缓存机制
  • RDD容错机制之checkpoint

什么是RDD

  • RDD（Resilient Distributed Dataset）叫做弹性分布式数据集，是Spark中最基本的数据抽象，它代表一个不可变、可分区、里面的元素可并行计算的集合.

    • Dataset：一个数据集，简单的理解为集合，用于存放数据的
    • Distributed：它的数据分布式存储，并且可以做分布式的计算
    • Resilient：它表示的是数据可以保存在磁盘，也可以保存在内存中

RDD的五大特性

  • A list of partitions

    • 每个RDD都有一个分区列表

  • A function for computing each split

    • 一个计算每个分区的函数。

  • A list of dependencies on other RDDs

    • 一个RDD依赖其他多个RDD，这个特性很重要，rdd的容错机制就是根据这个特性而来的

  • Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)

    • 可选项：针对于 k-v 键值对的RDD才具有该分区特性

  • Optionally, a list of preferred locations to compute each split on (e.g. block locations for an HDFS file)

    • 可选项 ：数据本地性，数据最优，选择尽量存储在worker节点上的数据节点。

创建rdd方式

  • 1、由一个已经存在的Scala集合创建。

    scala> val rdd1 = sc.parallelize(Array(1,2,3,4,5,6,7,8))
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:24

  • 2、由外部存储系统的数据集创建，包括本地的文件系统，还有所有Hadoop支持的数据集，比如HDFS、Cassandra、HBase等

    scala> val rdd2 = sc.textFile("/words.txt")
rdd2: org.apache.spark.rdd.RDD[String] = /words.txt MapPartitionsRDD[2] at textFile at <console>:24

  • 3、由一个rdd转化成一个新的RDD

    scala> val rdd3=rdd2.flatMap(_.split(" "))
rdd3: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[4] at flatMap at <console>:26

RDD的算子分类

  • transformation：转换操作，将一个rdd转换生成一个新的rdd(flatMap/map/reduceByKey),它属于懒加载，延迟执行，并不会立即触发任务的执行。
  • action :此时才会真正的触发任务的计算。

掌握RDD的常用算子操作

  • 练习1：map、filter

    //通过并行化生成rdd
val rdd1 = sc.parallelize(List(5, 6, 4, 7, 3, 8, 2, 9, 1, 10))
//对rdd1里的每一个元素乘2然后排序
val rdd2 = rdd1.map(_ * 2).sortBy(x => x, true)
//过滤出大于等于十的元素
val rdd3 = rdd2.filter(_ >= 10)
//将元素以数组的方式在客户端显示
rdd3.collect

  • 练习2：flatmap

    val rdd1 = sc.parallelize(Array("a b c", "d e f", "h i j"))
//将rdd1里面的每一个元素先切分再压平
val rdd2 = rdd1.flatMap(_.split(' '))
rdd2.collect

  • 练习3：交集、并集

    val rdd1 = sc.parallelize(List(5, 6, 4, 3))
val rdd2 = sc.parallelize(List(1, 2, 3, 4))
//求并集
val rdd3 = rdd1.union(rdd2)
//求交集
val rdd4 = rdd1.intersection(rdd2)
//去重
rdd3.distinct.collect
rdd4.collect

  • 练习4：join、groupbykey

    val rdd1 = sc.parallelize(List(("tom", 1), ("jerry", 3), ("kitty", 2)))
val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 1), ("shuke", 2)))
//求jion
val rdd3 = rdd1.join(rdd2)
rdd3.collect
//求并集
val rdd4 = rdd1 union rdd2
rdd4.collect
//按key进行分组
val rdd5 = rdd4.groupByKey
rdd5.collect

  • 练习5：cogroup

    val rdd1 = sc.parallelize(List(("tom", 1), ("tom", 2), ("jerry", 3), ("kitty", 2)))
val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 1), ("shuke", 2)))
//cogroup
val rdd3 = rdd1.cogroup(rdd2)
//注意cogroup与groupByKey的区别
rdd3.collect

  • 练习6：reduce

    val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5))
//reduce聚合
val rdd2 = rdd1.reduce(_ + _)
rdd2.collect

  • 练习7：reducebykey、sortByKey

    val rdd1 = sc.parallelize(List(("tom", 1), ("jerry", 3), ("kitty", 2),  ("shuke", 1)))
val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 3), ("shuke", 2), ("kitty", 5)))
val rdd3 = rdd1.union(rdd2)
//按key进行聚合
val rdd4 = rdd3.reduceByKey(_ + _)
rdd4.collect
//按value的降序排序
val rdd5 = rdd4.map(t => (t._2, t._1)).sortByKey(false).map(t => (t._2, t._1))
rdd5.collect

rdd中的2种依赖关系

  • 窄依赖

    • 窄依赖指的是每一个父RDD的Partition最多被子RDD的一个Partition使用
    • 总结：窄依赖我们形象的比喻为独生子女

  • 宽依赖

    • 宽依赖指的是多个子RDD的Partition会依赖同一个父RDD的Partition
    • 总结：宽依赖我们形象的比喻为超生

Lineage(血统)

  • RDD的Lineage会记录RDD的元数据信息和转换行为，当该RDD的部分分区数据丢失时，它可以根据这些信息来重新运算和恢复丢失的数据分区。
  • Spark具备容错机制

RDD的缓存方式

  • cache:直接将rdd中的数据，保存在内存中，其本质是persist(StorageLevel.MEMOEY_ONLY)
  • persist:可以有丰富的缓存级别
  • 当rdd设置了缓存之后，如果下面有需要用到该RDD的数据的时候，就不要重复计算，可以直接从缓存中获取得到。

  • 源码解析：

     /** Persist this RDD with the default storage level (`MEMORY_ONLY`). */
 def persist(): this.type = persist(StorageLevel.MEMORY_ONLY)

 /** Persist this RDD with the default storage level (`MEMORY_ONLY`). */
 def cache(): this.type = persist()
    /**
* Various [[org.apache.spark.storage.StorageLevel]] defined and utility functions for creating
* new storage levels.
*/
object StorageLevel {
  val NONE = new StorageLevel(false, false, false, false)
  val DISK_ONLY = new StorageLevel(true, false, false, false)
  val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)
  val MEMORY_ONLY = new StorageLevel(false, true, false, true)
  val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)
  val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
  val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)
  val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
  val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)
  val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
  val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)
  val OFF_HEAP = new StorageLevel(true, true, true, false, 1)

RDD容错机制之checkpoint

  • 会对数据进行一个持久化操作，保存在hdfs

  • 使用的时候：

    • 需要sc.setCheckpointdir 来设置一个检查点目录
    • 对需要缓存的rdd调用checkpoint
    • 注意：同样在执行数据缓存的时候，需要有对应的action算子操作，才会真正触发持久化操作。
  • 在做checkpoint操作的时候，此时会先执行对应触发action算子的rdd结果，计算完成之后又会开辟一个新的job来计算你设置了checkpoint的rdd的结果。
  • 在设置了checkpoint之后，对应这个rdd会改变之前的依赖关系，如果当前数据丢失了，只有重头计算得到。

  • 如何使用checkpoint

    • 可以对要做checkPoint的rdd,先进行一个cache
    • 再做一个checkpint操作

  • 数据恢复的一般顺序

    • 内存---------->checkpoint------------>重新计算得到

每一个成功的背后都有无数个无人知晓的黑夜。

因为

夜晚，是超越对手的最佳时机。

===================== 码农1024 =====================#蔺光岭#


本文由 蔺光岭 创作，采用 知识共享署名 4.0，可自由转载、引用，但需署名作者且注明文章出处。

还不快抢沙发

添加新评论