RDD.collect的操作

collect操作,在最後的ResultTask.runTask中,執行的function的操作為下面程式碼.

由於對ResultTask的runTask這個函式的返回值就是這個runTask函式在執行完成RDD傳入的function後的返回值.這裡要說明下如果task的結果超過了spark.driver.maxResultSize配置的最大值時,預設是1G,直接對task的結果進行丟掉,不處理,

defcollect(): Array[T] = withScope {val results = sc.runJob(this, (iter: Iterator[T]) => iter.toArray

這個操作把所有的task返回的array進行連線,合併到一個array中進行返回.  Array.concat(results: _*)}

用於在driver端處理各個Task的結果返回的resultHandler函式,這個函式就是把對應的task的結果直接放到driver端接收資料的一個數組中.

valresults = new Array[U](partitions.size)

(index, res) => results(index) = res

在這個操作中,是直接把每個partition中Iterator的結果轉換成一個array.上面的紅色部分((iter: Iterator[T]) => iter.toArray

從上面的程式碼中可以看出來,針對一個rdd的collect的操作是把當前的rdd中所有的partition中的資料集的iterator直接轉換成一個array[T],這個array也是對應此partition的返回值,使用collect時要確保每個task的返回的資料的大小,同時要保證所有的task中返回的資料的大小不能超過1GB.

在driver端接收到每一個task返回的資料集時,每個task返回的是這個task中所有的資料集的陣列,通過在driver端定義的一個results陣列,這個陣列的長度就是partition的個數,每個task的返回結果儲存到這個對應的index位置上((index, res) => results(index) = res