RDD的五大特性以及五大特性在源码中的体现

首先提一下之前的内容，RDD的五大特性以及五大特性在源码中的体现，之前的博客有详细的讲解，这里简单说一下。详情参考：Spark之RDD

• ①A list of partitions 一个RDD是由一系列的Partition构成。
  对应源码方法：getPartitions
• ②A function for computing each split 对一个RDD做计算，实际上是对RDD里每个分区进行计算。
  对应源码方法：compute
• ③A list of dependencies on other RDDs RDD之间有转换，有依赖，它们之间有对应关系
  对应源码方法：getDependencies
• ④Optionally, a Partitioner for key-value RDDs 可选 对于key-value的RDD可指定一个partitioner，告诉它如何分区
  对应源码方法：partitioner
• ⑤Optionally, a list of preferred locations to compute each split on 对每个分片进行计算的时候，对每个split拿到PreferredLocations最佳位置，要运行的计算/执行最好在哪(几)个机器上运行
  对应源码方法：getPreferredLocations
  另外需要注意的是一个partition对应一个task，有多少partitions就有多少个task来执行。

那么问题来了，

• 1）上面①②③对应的三个方法到底是在driver端还是在executor端运行呢？？？
  带回答。。。。
  你以后写的Spark代码，哪些代码运行在driver端，哪些代码运行在executor端？自己都需要很清楚。比如有些代码在driver端执行可能内存不够，有些代码在executor端执行内存不够，针对不同的场景需要不同的处理方式，所以都需要很清楚。
  由上一节可以知道：一个Spark应用程序包含一个driver和多个executor。Driver program是一个进程，它运行应用程序application里面的main()函数，并在main函数里面创建SparkContext。Executor是在worker node上启动应用程序的进程，这个进程可以运行多个任务并将数据保存在内存或磁盘存储中。
• 2）上面①②③对应的三个方法的输入Input和输出Output是什么？
  ①getPartitions是没有输入，输出是Array[Partition]，一个数组，数组里面是Partition；
  ②compute输入是Partition，输出是Iterator[T]迭代器；
  （（迭代器）不是一个集合，迭代器Iterator提供了一种访问集合的方法，可以通过while或者for循环来实现对迭代器的遍历）
  ③getDependencies没有输入，输出是Seq[Dependency[_]]，一个数组，数组里面是一系列的依赖。

关于Stage

Each job gets divided into smaller sets of tasks called stages that depend on each other (similar to the map and reduce stages in MapReduce); you’ll see this term used in the driver’s logs.
每个action触发一个job，每个job被切分成小的任务集，这些小的任务集叫做stages，并且他们之间相互依赖（类似于MapReduce中的map和reduce阶段）。一个job可以有多个stage。
遇到一个action就会触发一个job，遇到shuffle就会触发一个stage。
访问：http://hadoop001:4040/jobs/

举例示范：
执行代码：

scala> sc.parallelize(List("a","wc","b","a","word","wc")).map((_,1)).collect
res1: Array[(String, Int)] = Array((a,1), (wc,1), (b,1), (a,1), (word,1), (wc,1))

看页面：

遇到一个action就会触发一个job，上面的collect是一个action。
可以看出它总共有一个stage，有一个parallelize和map两个算子。总共有两个task任务。

再具体点：


再来执行一下代码：

scala> sc.parallelize(List("a","wc","b","a","word","wc")).map((_,1)).reduceByKey(_+_).collect
res2: Array[(String, Int)] = Array((b,1), (word,1), (wc,2), (a,2))

看页面：

遇到一个action就会触发一个job，上面的collect是一个action。
遇到shuffle就会触发一个stage，上面reduceByKey存在shuffle过程，会触发stage。
可以看出，它总共有2个stage，2个stage分别有两个任务，总共4个任务。





总结，由上面可以看出，一个job可以有多个stage，一个stage是有一堆task组成的，一个task是被发送盗一个executor去执行的最小的工作单元。

RDD之cache(persist)

官网解释

看官网：http://spark.apache.org/docs/latest/rdd-programming-guide.html#rdd-persistence

One of the most important capabilities in Spark is persisting (or caching) a dataset in memory across operations. When you persist an RDD, each node stores any partitions of it that it computes in memory and reuses them in other actions on that dataset (or datasets derived from it). This allows future actions to be much faster (often by more than 10x). Caching is a key tool for iterative algorithms and fast interactive use.
Spark最为重要的特性之一就是可以在多个操作之间，把数据persisting（或caching）到内存中（在这里是executor里）。当你persist or cache一个RDD的时候，(RDD是由partition构成的)，每个节点会存储这些分区，每个节点在内存中计算这些分区的数据并在该数据集上的其他操作（或从中派生的数据集）中重用这些分区数据。这样后面的action操作会更快（通常超过10倍）。缓存是迭代算法和快速交互式使用的关键工具。

You can mark an RDD to be persisted using the persist() or cache() methods on it. The first time it is computed in an action, it will be kept in memory on the nodes. Spark’s cache is fault-tolerant – if any partition of an RDD is lost, it will automatically be recomputed using the transformations that originally created it.
你可以用persist()或cache()方法将RDD标记为持久化。某个操作（Action）触发该RDD的数据第一次被计算时，它将保存在节点的内存中，计算的结果数据（也就是该RDD的数据）就会以分区的形式被缓存于计算节点的内存中。Spark的缓存是容错的 - 如果RDD的任何分区丢失，它将自动使用之前创建它的transformations重新计算，通过血缘信息（Lineage）。

In addition, each persisted RDD can be stored using a different storage level, allowing you, for example, to persist the dataset on disk, persist it in memory but as serialized Java objects (to save space), replicate it across nodes. These levels are set by passing a StorageLevel object (Scala, Java, Python) to persist(). The cache() method is a shorthand for using the default storage level, which is StorageLevel.MEMORY_ONLY (store deserialized objects in memory).
此外，每个持久RDD可以使用不同的StorageLevel 进行存储，例如，允许你将数据集保存在磁盘上，保存在内存中，但作为序列化的Java对象（以节省空间），将其复制到节点上。这些级别通过传递一个 StorageLevel对象（Scala， Java， Python）来设置persist()。该cache()方法是使用默认存储级别的简写，它是StorageLevel.MEMORY_ONLY（将反序列化对象存储在内存中）。

RDD的存储形式或存储介质是可以通过存储级别（Storage Level）被定义的。例如，将数据持久化到磁盘、将Java对象序列化之后（有利于节省空间）缓存至内存、开启复制（RDD的分区数据可以被备份到多个节点防止丢失）或者使用堆外内存（Tachyon）。persist()可以接收一个StorageLevel对象（Scala、Java、Python）用以定义存储级别，如果使用的是默认的存储级别（StorageLevel.MEMORY_ONLY)，Spark提供了一个便利方法：cache()。

举例介绍

执行一下代码：

scala> val lines = sc.textFile("hdfs://hadoop001:9000/data/wordcount.txt")
lines: org.apache.spark.rdd.RDD[String] = hdfs://hadoop001:9000/data/wordcount.txt MapPartitionsRDD[6] at textFile at <console>:24

scala> lines.collect
res3: Array[String] = Array(world       world   hello, China    hello, people   person, love)

看一下页面上是没有的被cache的：

再执行一下：

scala> lines.cache
res4: lines.type = hdfs://hadoop001:9000/data/wordcount.txt MapPartitionsRDD[6] at textFile at <console>:24

再看页面上，还是没有被cache，因为cache或persist是lazy的，和transformtion是一样的，它需要通过action才能被触发。
在上面基础上，再执行一下，含有action的，去触发它，再看页面，就有了：

scala> lines.collect
res5: Array[String] = Array(world       world   hello, China    hello, people   person, love)

从上面可以看到，输入的是74.0B,缓存了之后变成了312.0 B,变大了。

cache和persist区别（面试经常问的）

可以看相应的源码

   // 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()

可以看到cache底层调用的是persist，persist底层调用的是persist(StorageLevel.MEMORY_ONLY)(方法的重载)。

persist之StorageLevel

存储级别选项如下：

源码：
https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/storage/StorageLevel.scala

看一下源码

class StorageLevel private(
    private var _useDisk: Boolean,   //是否使用磁盘
    private var _useMemory: Boolean,   //是否使用内存
    private var _useOffHeap: Boolean,   //是否使用Heap
    private var _deserialized: Boolean,   //是否使用 反序列化
    private var _replication: Int = 1)   //是否使用副本（不写就默认1）
  extends Externalizable {
  // TODO: Also add fields for caching priority, dataset ID, and flushing.
  private def this(flags: Int, replication: Int) {
    this((flags & 8) != 0, (flags & 4) != 0, (flags & 2) != 0, (flags & 1) != 0, replication)
  }

再看

看页面：

如何使用缓存？
定义了一个RDD lines，
可以这样使用：lines.persist(StorageLevel .MEMORY_ONLY_SER_2 )
如何去掉缓存？
lines.unpersist(true)
这样直接执行后，页面上就看不到它的缓存了，这也说明unpersist并不是lazy的，它是和action一样。

如何选择Storage Level？

Spark also automatically persists some intermediate data in shuffle operations (e.g. reduceByKey), even without users calling persist. This is done to avoid recomputing the entire input if a node fails during the shuffle. We still recommend users call persist on the resulting RDD if they plan to reuse it.
即使用户没有主动使用persist，在含有shuffle的操作(比如reduceByKey)中， Spark也会自动persist缓存一些中介数据。这样做可以避免：在shuffle过程中，某个节点挂了，重新计算整个输入的数据。如果RDD的结果后面会重复使用，Spark仍然推荐用户去使用persist 或者cache去把数据缓存下来。
那么如何去选择存储级别？
实际上存储级别的选取就是Memory与CPU效率之间的双重权衡，可以参考下述内容：

• 1）如果你的RDD适合默认的存储级别（MEMORY_ONLY），那么就让它们保持这种状态。这是CPU工作最为高效的一种方式，允许RDD上的操作以尽可能快的速度运行。
• 2）如果 1）不能满足，就是说走不了 1）的话，再用 2），则尝试使用MEMORY_ONLY_SER，且选择一种快速的序列化库selecting a fast serialization library，以使对象更加节省空间，但访问速度仍然相当快。（Java和Scala）
• 3）不要把数据persist或cache到磁盘，除非计算是非常“昂贵”的或者计算过程会过滤掉大量数据，因为重新计算一个分区数据的速度可能要高于从磁盘读取一个分区数据的速度；（仅仅针对于Spark core里面，只要带有存磁盘的一律不要）
• 4）如果需要快速的失败恢复机制，则使用备份的存储级别，如MEMORY_ONLY_2、MEMORY_AND_DISK_2；虽然所有的存储级别都可以通过重新计算丢失的数据实现容错，但是备份机制使得大部分情况下应用无需中断，即数据丢失情况下，直接使用备份数据，而不需要重新计算数据的过程。（实际上生产上，是没有那么多资源给你用的，你如果副本搞个几份，得不偿失）（在这里都不需要副本，直接默认1就可以了）
• 5）如果处于大内存或多应用的场景下，OFF_HEAP可以带来以下的好处：
  a. 它允许Spark Executors可以共享Tachyon的内存数据；
  b. 它很大程序上减少JVM垃圾回收带来的性能开销；
  c. Spark Executors故障不会导致数据丢失。

综合上面，总结下来可以看出，对于Spark core里面，只要带有存磁盘的一律不要，然后都不需要副本，直接默认1就可以了，所以我们只需要考虑MEMORY_ONLY、MEMORY_ONLY_SER 这两种就可以了。

去除缓存数据Removing Data

Spark automatically monitors cache usage on each node and drops out old data partitions in a least-recently-used (LRU) fashion. If you would like to manually remove an RDD instead of waiting for it to fall out of the cache, use the RDD.unpersist() method.
Spark会自动监视每个节点上的高速缓存使用情况，并以最近最少使用（LRU算法）方式删除旧数据分区数据。如果你想要手动删除RDD，而不是等待它退出缓存，请使用该RDD.unpersist()方法。
cache的数据在executor上面，当你sc.stop()后，理论上这些缓存都会消失，但是还是最好在最后加上xxx.unpersist(true)。

recompute重新计算

看上图，当RDD1里面的某个分区r3挂掉了，后面的计算肯定不能继续往下走，那么就需要找到r3的父RDD的相应的分区，在这里它找到了p2，会由p2再重新计算一下得到r3，但是其他的分区不会参与计算的。再比如，如果p2也挂掉了，继续往前找，直到找到为止。
上面r3挂掉只是简单的重新计算，血缘关系比较简单。
下面介绍一下窄依赖和宽依赖。

窄依赖和宽依赖（面试经常）

RDD和它依赖的父RDD的关系有两种不同的类型，即窄依赖（narrow dependency）和宽依赖（wide dependency）。
窄依赖：一个父RDD的Partition只能被子RDD的某个partition使用一次。
宽依赖：一个父RDD的Partition会被子RDD的partition使用多次。
宽依赖存在shuffle的。
如果问你join是窄依赖还是宽依赖，要分情况来说。
Lineage（血统）：RDD只支持粗粒度转换，即只记录单个块上执行的单个操作。将创建RDD的一系列Lineage（即血统）记录下来，以便恢复丢失的分区。RDD的Lineage会记录RDD的元数据信息和转换行为，当该RDD的部分分区数据丢失时，它可以根据这些信息来重新运算和恢复丢失的数据分区。
见下图，左边是窄依赖，右边是宽依赖：

下图中有3个stage，A到B遇到groupBy有shuffle，拆了个stage1；
C到D，D、E到F，map、union都是窄依赖，没有shuffle，不会拆成stage。
B、F到G拆分stage3.

遇到action之前，如果遇到shuffle算子，就会变成两个stage。遇到一个action会触发一个job。
一个action触发一个job，一个job由n个stage构成，一个stage由n个task构成。
并不是说一个算子就是一个task。
像上图，窄依赖中，它是以pipeline的方式运行的，一条水管子直接走到底。C中的一个partition到D中的对应的partition，再到F中的对应的那个partition，直接到底，这是一个task。一个partition是一个task。并行度 = partition数量 = task数量。

Key-Value Pairs键值对

Spark可以对RDD做很多操作，这些RDD可以包含各种各样的类型，但是只有少部分特殊的操作只在键值对结构的RDD上可用。最常用的是含有shuffle的操作，如通过元素的 key 来进行 grouping 或 aggregating 操作。
在scala中，这些操作都是自动可用的scala包含了tuple2 RDDS（内置对象的元组的语言，由简单的写作（A，B）），键值对的操作都定义在了PairRDDFunctions类中，该类将对元组RDD的功能进行增强。
例如，下面的代码使用的 Key-Value 对的 reduceByKey 操作统计文本文件中每一行出现了多少次:

val lines = sc.textFile("data.txt")
val pairs = lines.map(s => (s, 1))
val counts = pairs.reduceByKey((a, b) => a + b)

//reduceByKey虽然表面上没有用PairRDDFunctions，但是底层还是调用PairRDDFunctions。  它是PairRDDFunctions.scala类里面的。有隐式转换的。可以跟踪一下源码。
//面试题：reduceByKey是哪个类里面的？？？并不是RDD.scala类里面的。

我们也可以使用counts.sortByKey()，例如，对其进行按字母排序，最后使用counts.collect()方法收集结果数据返回到驱动程序。
注意：当你使用键值对RDD操作一个自定义的对象时，如果你重写了equals()方法也必须重写hashCode()方法。有关详情, 请参阅 Object.hashCode() documentation 中列出的约定。