RDD作為彈性分布式數據集，它的彈性具體體現在以下七個方面。
1．自動進行內存和磁盤數據存儲的切換
Spark會優先把數據放到內存中，如果內存實在放不下，會放到磁盤里面，不但能計算內存放下的數據，也能計算內存放不下的數據。如果實際數據大於內存，則要考慮數據放置策略和優化算法。當應用程序內存不足時，Spark應用程序將數據自動從內存存儲切換到磁盤存儲，以保障其高效運行。
2．基於Lineage（血統）的高效容錯機制
Lineage是基於Spark RDD的依賴關系來完成的（依賴分為窄依賴和寬依賴兩種形態），每個操作只關聯其父操作，各個分片的數據之間互不影響，出現錯誤時只要恢復單個Split的特定部分即可。常規容錯有兩種方式：一個是數據檢查點；另一個是記錄數據的更新。數據檢查點的基本工作方式，就是通過數據中心的網絡鏈接不同的機器，然后每次操作的時候都要復制數據集，就相當於每次都有一個復制，復制是要通過網絡傳輸的，網絡帶寬就是分布式的瓶頸，對存儲資源也是很大的消耗。記錄數據更新就是每次數據變化了就記錄一下，這種方式不需要重新復制一份數據，但是比較復雜，消耗性能。Spark的RDD通過記錄數據更新的方式為何很高效？因為① RDD是不可變的且Lazy；② RDD的寫操作是粗粒度的。但是，RDD讀操作既可以是粗粒度的，也可以是細粒度的。
3．Task如果失敗，會自動進行特定次數的重試
默認重試次數為4次。TaskSchedulerImpl的源碼如下所示：
private[spark] class TaskSchedulerImpl(
val sc: SparkContext,
val maxTaskFailures: Int,
isLocal: Boolean = false)
extends TaskScheduler with Logging
{
def this(sc: SparkContext) = this(sc, sc.conf.getInt("spark.task.maxFailures", 4))
val conf = sc.conf
TaskSchedulerImpl是底層的任務調度接口TaskScheduler的實現，這些Schedulers從每一個Stage中的DAGScheduler中獲取TaskSet，運行它們，嘗試是否有故障。DAGScheduler是高層調度，它計算每個Job的Stage的DAG，然后提交Stage，用TaskSets的形式啟動底層TaskScheduler調度在集群中運行。
4．Stage如果失敗，會自動進行特定次數的重試
這樣，Stage對象可以跟蹤多個StageInfo（存儲SparkListeners監聽到的Stage的信息，將Stage信息傳遞給Listeners或web UI）。默認重試次數為4次，且可以直接運行計算失敗的階段，只計算失敗的數據分片，Stage的源碼如下所示：
private[scheduler] abstract class Stage(
val id: Int,
val rdd: RDD[_],
val numTasks: Int,
val parents: List[Stage],
val firstJobId: Int,
val callSite: CallSite)
extends Logging {
val numPartitions = rdd.partitions.length
/** Set of jobs that this stage belongs to.屬於這個工作集的Stage */
val jobIds = new HashSet[Int]
val pendingPartitions = new HashSet[Int]
/** The ID to use for the next new attempt for this stage.用於此Stage的下一個新attempt的ID */
private var nextAttemptId: Int = 0
val name: String = callSite.shortForm
val details: String = callSite.longForm
/**
* Pointer to the [StageInfo] object for the most recent attempt. This needs to be initialized
* here, before any attempts have actually been created, because the DAGScheduler uses this
* StageInfo to tell SparkListeners when a job starts (which happens before any stage attempts
* have been created).
* 最新的[StageInfo] Object指針，需要被初始化，任何attempts都是被創造出來的，因為DAGScheduler使用
* StageInfo告訴SparkListeners工作何時開始（即發生前的任何階段已經創建）
*/
private var _latestInfo: StageInfo = StageInfo.fromStage(this, nextAttemptId)
/**
* Set of stage attempt IDs that have failed with a FetchFailure. We keep track of these
* failures in order to avoid endless retries if a stage keeps failing with a FetchFailure.
* We keep track of each attempt ID that has failed to avoid recording duplicate failures if
* multiple tasks from the same stage attempt fail (SPARK-5945).
* 設置Stage attempt IDs當失敗時可以讀取失敗信息，跟蹤這些失敗，為了避免無休止的重復失敗
* 跟蹤每一次attempt，以便避免記錄重復故障，如果從同一stage創建多任務失敗(SPARK-5945)
*/
private val fetchFailedAttemptIds = new HashSet[Int]
private[scheduler] def clearFailures() : Unit = {
fetchFailedAttemptIds.clear()
}
/**
* Check whether we should abort the failedStage due to multiple consecutive fetch failures.
* 檢查是否應該中止由於連續多次讀取失敗的stage
* This method updates the running set of failed stage attempts and returns
* true if the number of failures exceeds the allowable number of failures.
* 如果失敗的次數超過允許的次數，此方法更新失敗stage attempts和返回的運行集
*/
private[scheduler] def failedOnFetchAndShouldAbort(stageAttemptId: Int): Boolean = {
fetchFailedAttemptIds.add(stageAttemptId)
fetchFailedAttemptIds.size >= Stage.MAX_CONSECUTIVE_FETCH_FAILURES
}
/** Creates a new attempt for this stage by creating a new StageInfo with a new attempt ID. */
// 在stage中創建一個新的attempt
def makeNewStageAttempt(
numPartitionsToCompute: Int,
taskLocalityPreferences: Seq[Seq[TaskLocation]] = Seq.empty): Unit = {
val metrics = new TaskMetrics
metrics.register(rdd.sparkContext)
_latestInfo = StageInfo.fromStage(
this, nextAttemptId, Some(numPartitionsToCompute), metrics, taskLocalityPreferences)
nextAttemptId += 1
}
/** Returns the StageInfo for the most recent attempt for this stage. */
// 返回當前stage中最新的stageinfo
def latestInfo: StageInfo = _latestInfo
override final def hashCode(): Int = id
override final def equals(other: Any): Boolean = other match {
case stage: Stage => stage != null && stage.id == id
case _ => false
}
/** Returns the sequence of partition ids that are missing (i.e. needs to be computed). */
// 返回需要重新計算的分區標識的序列
def findMissingPartitions(): Seq[Int]
}
private[scheduler] object Stage {
// The number of consecutive failures allowed before a stage is aborted
// 允許一個stage中止的連續故障數
val MAX_CONSECUTIVE_FETCH_FAILURES = 4
}
Stage是Spark Job運行時具有相同邏輯功能和並行計算任務的一個基本單元。Stage中所有的任務都依賴同樣的Shuffle，每個DAG任務通過DAGScheduler在Stage的邊界處發生Shuffle形成Stage，然后DAGScheduler運行這些階段的拓撲順序。每個Stage都可能是ShuffleMapStage，如果是ShuffleMapStage，則跟蹤每個輸出節點（nodes）上的輸出文件分區，它的任務結果是輸入其他的Stage(s)，或者輸入一個ResultStage，若輸入一個ResultStage，這個ResultStage的任務直接在這個RDD上運行計算這個Spark Action的函數（如count()、 save()等），並生成shuffleDep等字段描述Stage和生成變量，如outputLocs和numAvailableOutputs，為跟蹤map輸出做准備。每個Stage會有firstjobid，確定第一個提交Stage的Job，使用FIFO調度時，會使得其前面的Job先行計算或快速恢復（失敗時）。
ShuffleMapStage是DAG產生數據進行Shuffle的中間階段，它發生在每次Shuffle操作之前，可能包含多個Pipelined操作，ResultStage階段捕獲函數在RDD的分區上運行Action算子計算結果，有些Stage不是運行在RDD的所有的分區上，例如，first()、lookup()等。SparkListener是Spark調度器的事件監聽接口。注意，這個接口隨着Spark版本的不同會發生變化。
5．checkpoint和persist（檢查點和持久化），可主動或被動觸發
checkpoint是對RDD進行的標記，會產生一系列的文件，且所有父依賴都會被刪除，是整個依賴（Lineage）的終點。checkpoint也是Lazy級別的。persist后RDD工作時每個工作節點都會把計算的分片結果保存在內存或磁盤中，下一次如果對相同的RDD進行其他的Action計算，就可以重用。
因為用戶只與Driver Program交互，因此只能用RDD中的cache()方法去cache用戶能看到的RDD。所謂能看到，是指經過Transformation算子處理后生成的RDD，而某些在Transformation算子中Spark自己生成的RDD是不能被用戶直接cache的。例如，reduceByKey()中會生成的ShuffleRDD、MapPartitionsRDD是不能被用戶直接cache的。在Driver Program中設定RDD.cache()后，系統怎樣進行cache?首先，在計算RDD的Partition之前就去判斷Partition要不要被cache，如果要被cache，先將Partition計算出來，然后cache到內存。cache可使用memory，如果寫到HDFS磁盤的話，就要檢查checkpoint。調用RDD.cache()后，RDD就變成persistRDD了，其StorageLevel為MEMORY_ONLY，persistRDD會告知Driver說自己是需要被persist的。此時會調用RDD.iterator()。 RDD.scala的iterator()的源碼如下：
/**
* Internal method to this RDD; will read from cache if applicable, or otherwise compute it.
* This should ''not'' be called by users directly, but is available for implementors of custom
* subclasses of RDD.
* RDD的內部方法，將從合適的緩存中讀取，否則計算它。這不應該被用戶直接使用，但可用於實現自定義的子RDD
*/
final def iterator(split: Partition, context: TaskContext): Iterator[T] = {
if (storageLevel != StorageLevel.NONE) {
getOrCompute(split, context)
} else {
computeOrReadCheckpoint(split, context)
}
}
當RDD.iterator()被調用的時候，也就是要計算該RDD中某個Partition的時候，會先去cacheManager那里獲取一個blockId，然后去BlockManager里匹配該Partition是否被checkpoint了，如果是，那就不用計算該Partition了，直接從checkpoint中讀取該Partition的所有records放入ArrayBuffer里面。如果沒有被checkpoint過，先將Partition計算出來，然后將其所有records放到cache中。總體來說，當RDD會被重復使用（不能太大）時，RDD需要cache。Spark自動監控每個節點緩存的使用情況，利用最近最少使用原則刪除老舊的數據。如果想手動刪除RDD，可以使用RDD.unpersist()方法。
此外，可以利用不同的存儲級別存儲每一個被持久化的RDD。例如，它允許持久化集合到磁盤上，將集合作為序列化的Java對象持久化到內存中、在節點間復制集合或者存儲集合到Alluxio中。可以通過傳遞一個StorageLevel對象給persist()方法設置這些存儲級別。cache()方法使用默認的存儲級別-StorageLevel.MEMORY_ONLY。RDD根據useDisk、useMemory、 useOffHeap、deserialized、replication 5個參數的組合提供了常用的12種基本存儲，完整的存儲級別介紹如下。StorageLevel.scala的源碼如下：
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)
StorageLevel是控制存儲RDD的標志，每個StorageLevel記錄RDD是否使用memory，或使用ExternalBlockStore存儲，如果RDD脫離了memory或ExternalBlockStore，是否扔掉RDD，是否保留數據在內存中的序列化格式，以及是否復制多個節點的RDD分區。另外，org.apache.spark.storage.StorageLevel是單實例（singleton）對象，包含了一些靜態常量和常用的存儲級別，且可用singleton對象工廠方法StorageLevel(...)創建定制化的存儲級別。
Spark的多個存儲級別意味着在內存利用率和CPU利用率間的不同權衡。推薦通過下面的過程選擇一個合適的存儲級別：①如果RDD適合默認的存儲級別（MEMORY_ONLY），就選擇默認的存儲級別。因為這是CPU利用率最高的選項，會使RDD上的操作盡可能地快。②如果不適合用默認級別，就選擇MEMORY_ONLY_SER。選擇一個更快的序列化庫提高對象的空間使用率，但是仍能夠相當快地訪問。③除非算子計算RDD花費較大或者需要過濾大量的數據，不要將RDD存儲到磁盤上，否則重復計算一個分區，就會和從磁盤上讀取數據一樣慢。④如果希望更快地恢復錯誤，可以利用replicated存儲機制，所有的存儲級別都可以通過replicated計算丟失的數據來支持完整的容錯。另外，replicated的數據能在RDD上繼續運行任務，而不需要重復計算丟失的數據。在擁有大量內存的環境中或者多應用程序的環境中，Off_Heap（將對象從堆中脫離出來序列化，然后存儲在一大塊內存中，這就像它存儲到磁盤上一樣，但它仍然在RAM內存中。Off_Heap對象在這種狀態下不能直接使用，須進行序列化及反序列化。序列化和反序列化可能會影響性能，Off_Heap堆外內存不需要進行GC）。Off_Heap具有如下優勢：Off_Heap運行多個執行者共享的Alluxio中相同的內存池，顯著地減少GC。如果單個的Executor崩潰，緩存的數據也不會丟失。
6．數據調度彈性，DAGScheduler、TASKScheduler和資源管理無關
Spark將執行模型抽象為通用的有向無環圖計划（DAG），這可以將多Stage的任務串聯或並行執行，從而不需要將Stage中間結果輸出到HDFS中，當發生節點運行故障時，可有其他可用節點代替該故障節點運行。
7．數據分片的高度彈性（coalesce）
Spark進行數據分片時，默認將數據放在內存中，如果內存放不下，一部分會放在磁盤上進行保存。
RDD.scala的coalesce算子代碼如下：
/**
* Return a new RDD that is reduced into `numPartitions` partitions.
*
* This results in a narrow dependency, e.g. if you go from 1000 partitions
* to 100 partitions, there will not be a shuffle, instead each of the 100
* new partitions will claim 10 of the current partitions.
*
* However, if you're doing a drastic coalesce, e.g. to numPartitions = 1,
* this may result in your computation taking place on fewer nodes than
* you like (e.g. one node in the case of numPartitions = 1). To avoid this,
* you can pass shuffle = true. This will add a shuffle step, but means the
* current upstream partitions will be executed in parallel (per whatever
* the current partitioning is).
*
* Note: With shuffle = true, you can actually coalesce to a larger number
* of partitions. This is useful if you have a small number of partitions,
* say 100, potentially with a few partitions being abnormally large. Calling
* coalesce(1000, shuffle = true) will result in 1000 partitions with the
* data distributed using a hash partitioner.
*/
def coalesce(numPartitions: Int, shuffle: Boolean = false,
partitionCoalescer: Option[PartitionCoalescer] = Option.empty)
(implicit ord: Ordering[T] = null)
: RDD[T] = withScope {
require(numPartitions > 0, s"Number of partitions ($numPartitions) must be positive.")
if (shuffle) {
/** Distributes elements evenly across output partitions, starting from a random partition. */
// 從隨機分區開始，將元素均勻分布在輸出分區上
val distributePartition = (index: Int, items: Iterator[T]) => {
var position = (new Random(index)).nextInt(numPartitions)
items.map { t =>
// Note that the hash code of the key will just be the key itself. The HashPartitioner
// will mod it with the number of total partitions.
// key的哈希碼是key本身，HashPartitioner將它與總分區數進行取模運算
position = position + 1
(position, t)
}
} : Iterator[(Int, T)]
// include a shuffle step so that our upstream tasks are still distributed
// 包括一個shuffle步驟，使我們的上游任務仍然是分布式的
new CoalescedRDD(
new ShuffledRDD[Int, T, T](mapPartitionsWithIndex(distributePartition),
new HashPartitioner(numPartitions)),
numPartitions,
partitionCoalescer).values
} else {
new CoalescedRDD(this, numPartitions, partitionCoalescer)
}
}
例如，在計算的過程中，會產生很多的數據碎片，這時產生一個Partition可能會非常小，如果一個Partition非常小，每次都會消耗一個線程去處理，這時可能會降低它的處理效率，需要考慮把許多小的Partition合並成一個較大的Partition去處理，這樣會提高效率。另外，有可能內存不是那么多，而每個Partition的數據Block比較大，這時需要考慮把Partition變成更小的數據分片，這樣讓Spark處理更多的批次，但是不會出現OOM。