2. 程式人生 > >Spark原始碼分析之Sort-Based Shuffle讀寫流程

Spark原始碼分析之Sort-Based Shuffle讀寫流程

阿新 • • 發佈：2019-01-22
override def read(): Iterator[Product2[K, C]] = {   // 構造ShuffleBlockFetcherIterator，一個迭代器，它獲取多個塊，對於本地塊，從本地讀取   // 對於遠端塊，通過遠端方法讀取val blockFetcherItr = new ShuffleBlockFetcherIterator(     context,     blockManager.shuffleClient,     blockManager,     //MapOutputTracker在SparkEnv啟動的時候例項化mapOutputTracker.getMapSizesByExecutorId
(handle.shuffleId, startPartition, endPartition),     // Note: we use getSizeAsMb when no suffix is provided for backwards compatibility     SparkEnv.get.conf.getSizeAsMb("spark.reducer.maxSizeInFlight", "48m") * 1024 * 1024,     SparkEnv.get.conf.getInt("spark.reducer.maxReqsInFlight", Int.MaxValue))   //
基於配置檔案對於流進行包裝val wrappedStreams = blockFetcherItr.map { case (blockId, inputStream) =>     serializerManager.wrapStream(blockId, inputStream)   }   // 獲取序列化例項val serializerInstance = dep.serializer.newInstance()   // 對於每一個流建立一個<key,value>迭代器val recordIter = wrappedStreams.flatMap { wrappedStream
=>     serializerInstance.deserializeStream(wrappedStream).asKeyValueIterator   }   // Update the context task metrics for each record read.   val readMetrics = context.taskMetrics.createTempShuffleReadMetrics()   val metricIter = CompletionIterator[(Any, Any), Iterator[(Any, Any)]](     recordIter.map { record =>       readMetrics.incRecordsRead(1)       record     },     context.taskMetrics().mergeShuffleReadMetrics())   // An interruptible iterator must be used here in order to support task cancellation   val interruptibleIter = new InterruptibleIterator[(Any, Any)](context, metricIter)   // 如果reduce端需要聚合val aggregatedIter: Iterator[Product2[K, C]] = if (dep.aggregator.isDefined) {     // 如果map端已經聚合過了if (dep.mapSideCombine) {       //則對讀取到的聚合結果進行聚合val combinedKeyValuesIterator = interruptibleIter.asInstanceOf[Iterator[(K, C)]]       // 針對map端各個partition對key進行聚合後的結果再次聚合dep.aggregator.get.combineCombinersByKey(combinedKeyValuesIterator, context)     } else {       // 如果map端沒有聚合，則針對未合併的<k,v>進行聚合val keyValuesIterator = interruptibleIter.asInstanceOf[Iterator[(K, Nothing)]]       dep.aggregator.get.combineValuesByKey(keyValuesIterator, context)     }   } else {     require(!dep.mapSideCombine, "Map-side combine without Aggregator specified!")     interruptibleIter.asInstanceOf[Iterator[Product2[K, C]]]   }   // 如果需要對key排序，則進行排序。基於sort的shuffle實現過程中，預設只是按照partitionId排序   // 在每一個partition內部並沒有排序，因此添加了keyOrdering變數，提供是否需要對分割槽內部的key排序dep.keyOrdering match {     case Some(keyOrd: Ordering[K]) =>       // 為了減少記憶體壓力和避免GC開銷，引入了外部排序器，當記憶體不足時會根據配置檔案       // spark.shuffle.spill決定是否進行spill操作val sorter =         new ExternalSorter[K, C, C](context, ordering = Some(keyOrd), serializer = dep.serializer)       sorter.insertAll(aggregatedIter)       context.taskMetrics().incMemoryBytesSpilled(sorter.memoryBytesSpilled)       context.taskMetrics().incDiskBytesSpilled(sorter.diskBytesSpilled)       context.taskMetrics().incPeakExecutionMemory(sorter.peakMemoryUsedBytes)       CompletionIterator[Product2[K, C], Iterator[Product2[K, C]]](sorter.iterator, sorter.stop())     case None =>       // 不需要排序直接返回aggregatedIter   } }

相關推薦

Spark原始碼分析之Sort-Based Shuffle讀寫流程

override def read(): Iterator[Product2[K, C]] = {   // 構造ShuffleBlockFetcherIterator，一個迭代器，它獲取多個塊，對於本地塊，從本地讀取   // 對於遠端塊，通過遠端方法讀取val blockFetcherItr = new

Spark原始碼分析之Spark Shell（上）

https://www.cnblogs.com/xing901022/p/6412619.html 文中分析的spark版本為apache的spark-2.1.0-bin-hadoop2.7。 bin目錄結構： -rwxr-xr-x. 1 bigdata bigdata 1089 Dec

Spark 原始碼分析之ShuffleMapTask處理

Spark 原始碼分析之ShuffleMapTask處理 更多資源 SPARK 原始碼分析技術分享(bilibilid視訊彙總套裝視訊): https://www.bilibili.com/video/av37442139/ github: https://github.com

Spark原始碼分析之ResultTask處理

Spark原始碼分析之ResultTask處理 更多資源 SPARK 原始碼分析技術分享(bilibilid視訊彙總套裝視訊): https://www.bilibili.com/video/av37442139/ github: https://github.com/open

Spark原始碼分析之ShuffleMapTask處理

Spark原始碼分析之ShuffleMapTask處理 更多資源 SPARK 原始碼分析技術分享(bilibilid視訊彙總套裝視訊): https://www.bilibili.com/video/av37442139/ github: https://github.com/opensour

Spark 原始碼分析之ShuffleMapTask記憶體資料Spill和合並

前置條件 Hadoop版本: Hadoop 2.6.0-cdh5.15.0 Spark版本: SPARK 1.6.0-cdh5.15.0 JDK.1.8.0_191 scala2.10.7 技能標籤 Spark ShuffleMapTask 記憶體中的資

spark原始碼分析之BypassMergeSortShuffleWriter

概述 spark1.6以後，取消了基於hash的shuffle，只剩下基於sort的shuffle。現在只存在以下三種shuffle writer： BypassMergeSortShuffleWriter UnsafeShuffleWriter SortShuffl

spark原始碼分析之ShuffleExternalSorter

概述 ShuffleExternalSorter是專門用於sort-based shuffle的external sorter。 傳入的record會被追加到data page。當所有的record都已經插入該sorter時，或者當前執行緒的shuffle memory已

spark原始碼分析之TaskMemoryManager

概述 TaskMemoryManager用於管理每個task分配的記憶體。 在off-heap記憶體模式中，可以用64-bit的地址來表示記憶體地址。在on-heap記憶體模式中，通過base object的引用 和該物件中64-bit 的偏移量來表示記憶體地址。 當我

spark原始碼分析之NioBufferedFileInputStream

NioBufferedFileInputStream是spark實現的一種新的位元組流，它既支援內部緩衝區，又支援nio讀取檔案，使用direct buffer避免java堆與native記憶體之間的資料拷貝。在Java jdk中沒有可供直接使用的具備以上2個功能的位元組流。

spark原始碼分析之UnsafeShuffleWriter

概述  SortShuffleManager會判斷在滿足以下條件時呼叫UnsafeShuffleWriter，否則降級為使用SortShuffleWriter： Serializer支援relocation。這是指Serializer可以對已經序列化的物件進行排序，這種排

spark原始碼分析之ReadAheadInputStream

概述 ReadAheadInputStream實現了從當前buffer讀取的data耗盡時，切換到另外一個buffer讀取資料，並啟動任務從底層輸入流非同步預讀data，放入耗盡的buffer中。它通過2個buffer來完成——active buffer和read ah

Spark原始碼分析之三：Stage劃分

        Stage劃分的大體流程如下圖所示：         前面提到，對於JobSubmitted事件，我們通過呼叫DAGScheduler的handleJobSubmitted()方法來處理。那麼我們先來看下程式碼： // 處理Job提交的函式 pri

spark原始碼分析之Master原始碼主備切換機制分析

Master原始碼分析之主備切換機制 1.當選為leader之後的操作 //ElectedLeader 當選leader case ElectedLeader => {

54：Spark中的Tungsten-sort Based Shuffle內幕

本期內容： 1. Tungsten-sort Based Shuffle原理 2. Tungsten-sort Based Shuffle原始碼 ShortShuffleManager Memory

Spark原始碼分析之Master資源排程演算法原理

Master是通過schedule方法進行資源排程，告知worker啟動executor等。 一schedule方法 1判斷master狀態，只有alive狀態的master才可以進行資源排程，sta

spark原始碼分析之worker原理篇

解釋： 1、master要求worker啟動driver和executor 2、worker啟動driver的一個基本的原理，worker會啟動一個執行緒DriverRunner，然後DriverRunner會去負責啟動driver程序，然後在之後對d

Spark原始碼分析之Master註冊機制原理

一 Worker向Master註冊 1.1 Worker啟動，呼叫registerWithMaster，向Master註冊 當worker啟動的時候，會呼叫registerWithMaster方法

spark 原始碼分析之四 -- TaskScheduler的建立和啟動過程

　　在 spark 原始碼分析之二 -- SparkContext 的初始化過程 中，第 14 步 和 16 步分別描述了 TaskScheduler的 初始化 和 啟動過程。 　　話分兩頭，先說 TaskScheduler的初始化過程 　TaskScheduler的例項化 1

spark 原始碼分析之十三 -- SerializerManager剖析

對SerializerManager的說明： 它是為各種Spark元件配置序列化，壓縮和加密的元件，包括自動選擇用於shuffle的Serializer。spark中的資料在network IO 或 local disk IO傳輸過程中。都需要序列化。其預設的 Serializer 是 org.a