**RDD 的五大屬性：** 分區、分區計算函數、依賴、分區器、分區優先位置列表。 * 一系列的分區（分片）信息，每個任務處理一個分區； * 每個分區上都有compute函數，計算該分區中的數據； * RDD之間有一系列的依賴； * 分區器決定數據（key-value）分配至哪個分區； * 優先位置列表，將計算任務分派到其所在處理數據塊的存儲位置； [TOC] # 1. 分區（Partition） **1. RDD有分區構成** 一個RDD 是由多個分區構成的，每個Partition 都有一個唯一索引編號。 ```scala import org.apache.spark.{SparkConf, SparkContext} object SparkDemo1 { def main(args: Array[String]): Unit = { val conf: SparkConf = new SparkConf() .setMaster("local[*]") .setAppName(this.getClass.getName) val sc: SparkContext = SparkContext.getOrCreate(conf) val rdd: RDD[Int] = sc.parallelize(List(1, 2, 3, 4, 5)) println(rdd.getNumPartitions) // 8個分區 rdd.partitions.foreach(x => println(x.index)) // 0 1 2 3 4 5 6 7 } } ``` RDD 分區概念與MapReduce 的輸入切片概念是類似的。對每個分區的運算會被當作一個 Task 執行，換句話說，<mark>分區是 Spark 任務執行的基本單位</mark>。 如果有 100 個分區，RDD 上有 n 個操作，那么將會產生有 `$ n * 100 $` 個任務。 <br/> **2. 使用分區的時候要了解的兩條規則** （1）<ins>只有Key-Value類型的RDD（比如PairRDD）才有分區器</ins>，非Key-Value類型的RDD分區器的值是 None。 （2）每個 RDD 的分區 ID 范圍：<ins>0~numPartitions-1</ins>，決定這個值是屬于那個分區的。 <br/> **3. 分區的影響** （1）一個RDD擁有的分區越多，得到的并行性就越強； （2）每個分區都是被分發到不同 Worker Node 節點上被處理； （3）每個分區對應一個Task；  <br/> # 2. compute函數 RDD 的每個分區上都有一個函數去作用，<mark>Spark 中的 RDD 的計算是以分區為單位的</mark>，<ins>每個 RDD 都會實現 compute 函數</ins>以達到這個目的。不同的 RDD 的compute 函數邏輯各不一樣，比如： 1. MapPartitionsRDD 的 compute 是將用戶的轉換邏輯作用到指定的 Partition上。因為 RDD 的 map 算子產生 MapPartitionsRDD，而 map 算子的參數（具體操作邏輯）是變化的。 2. HadoopRDD 的 compute 是讀取指定 Partition 數據 。 因為`sc.hadoopFile("path")`讀取 HDFS 文件返回的 RDD 具體類型便是 HadoopRDD，所以只需要讀取數據即可。 3. CheckpointRDD 的 compute 是直接讀取檢查點的數據。一旦 RDD 進行checkpoint（在下一章中介紹），將變成 CheckpointRDD。如下圖所示。  <br/> # 3. RDD間的依賴 RDD 有依賴性，通常情況下一個 RDD 是來源于另一個 RDD，這個叫做 **lineage（血統）**。RDD 會記錄下這些依賴，方便容錯，也稱 DAG。如下圖所示，RDD 所依賴的其他RDD 作為構造器參數之一。  <br/> # 4. 分區器（Partitioner） RDD 的分區器是一個可選項，如果 RDD 里面存的數據是 key-value 形式，則可以傳遞一個自定義的 Partitioner 進行重新分區，例如這里自定義的 Partitioner是基于 key 進行分區，那則會將不同 RDD 里面的相同 key 的數據放到同一個Partition 里面。 <br/> Spark 有兩種分區器： **HashPartitioner （ 哈 希 分 區 ）** 和 **RangePartitioner（范圍分區）**。 <br/> HashPartitioner 是默認的分區方式，其算法邏輯是：`$ partition = key.hashCode() % numPartitions $` <br/> RangePartitioner 通過兩個步驟來實現： 第一步：先從整個 RDD 中抽取出樣本數據，將樣本數據排序，計算出每個分區的最大 key 值，形成一個 Array[KEY]類型的數組變量 rangeBounds； 第二步：判斷 key 在 rangeBounds 中所處的范圍，給出該 key 值在下一個 RDD中的分區 id 下標；該分區器要求 RDD 中的 KEY 類型必須是可以排序的。 <br/> # 5.分區優先位置列表 該列表<ins>存儲了存取每個分區的優先位置</ins>。對于一個 HDFS 文件來說，這個列表保存了每個分區所在的數據塊的位置。按照<ins>移動數據不如移動計算的</ins>的理念，Spark 在進行任務調度的時候，會盡可能的將計算任務移動到所要處理的數據塊的存儲位置。<br/> preferredLocations 返回每個分區所在的機器名或者 IP 地址，如果分區數據是多份存儲，那么返回多個機器地址。如下圖所示。