Spark RDD分割槽數和分割槽器

partition

rdd有個partitions方法：

final def partitions: array[partition]，

能夠返回乙個陣列，陣列元素是rdd的partition。

partition是rdd的最小資料處理單元，可以看作是乙個資料塊，每個partition有個編號index。

乙個partition被乙個map task處理。

partitioner

mr任務的map階段的處理結果會進行分片（也可以叫分割槽，這個分割槽不同於上面的分割槽），分片的數量就是reduce task的數量。

具體怎麼分片由分割槽器partitioner決定，spark中預設定義了兩種partitioner：

hash分割槽器會根據key-value的鍵值key的hashcode進行分割槽，速度快，但是可能產生資料偏移，造成每個分割槽中資料量不均衡。

range分割槽器會對現有rdd中的key-value資料進行抽樣，盡量找出均衡分割點，一定程度上解決了資料偏移問題，力求分割槽後的每個分區內資料量均衡，但是速度相對慢。

partitioner分割槽詳情

在對父rdd執行完map階段任務後和在執行reduce階段任務前，會對map階段中間結果進行分割槽。

分割槽由父rdd的partitioner確定，主要包括兩部分工作：

確定分割槽數量（也就是reduce task數量），也是子rdd的partition數量。

決定將map階段中間結果的每個key-value對分到哪個分割槽上。

假設乙個父rdd要執行reducebykey任務，我們可以顯式的指定分割槽器：

val rdd_child = rdd_parent.
reducebykey
(new
hashpartitioner(3
), _+_)

hashpartitioner構造引數3就是分割槽數量，也是啟動的reduce task數量，也是reducebykey結果返回的子rdd的partitions方法返回的陣列的長度。

如果沒有顯式指定分割槽器，則會呼叫org.apache.spark包下伴生物件partitioner的defaultpartitioner靜態方法返回的分割槽器作為預設分割槽器。

defaultpartitioner返回預設分割槽器的過程如下：

嘗試利用父rdd的partitioner，如果父rdd沒有partitioner，則會檢視sparkconf中是否定義了spark.default.parallelism配置引數，如果定義了就返回new hashpartitioner(sc.defaultparallelism)作為預設分割槽器，如果沒定義就返回new hashpartitioner(rdd_parent.partitions.length)作為預設分割槽器——

以下是原始碼：

包下伴生物件object partitioner的方法
def defaultpartitioner
(rdd: rdd[_]
, others: rdd[_]*)
: partitioner =
else
else
}}

更具體的，無論是以本地模式、standalone 模式、yarn 模式或者是 mesos 模式來執行 apache spark，分割槽的預設個數等於對spark.default.parallelism的指定值，若該值未設定，則 apache spark 會根據不同集群模式的特徵，來確定這個值。

Spark RDD分割槽數和分割槽器

SparkRDD的分割槽

Spark RDD 建立和分割槽規則

舉例說明Spark RDD的分割槽依賴

Spark RDD分割槽數和分割槽器

SparkRDD的分割槽

Spark RDD 建立和分割槽規則

舉例說明Spark RDD的分割槽 依賴

相關推薦

舉例說明Spark RDD的分割槽依賴