Spark源碼解析(一):RDD之Transfrom運算元

来源:https://www.cnblogs.com/river97/archive/2023/03/31/17277318.html
-Advertisement-
Play Games

一、延遲計算 RDD 代表的是分散式數據形態,因此,RDD 到 RDD 之間的轉換,本質上是數據形態上的轉換(Transformations) 在 RDD 的編程模型中,一共有兩種運算元,Transformations 類運算元和 Actions 類運算元。開發者需要使用 Transformations ...


一、延遲計算

RDD 代表的是分散式數據形態,因此,RDD 到 RDD 之間的轉換,本質上是數據形態上的轉換(Transformations)

在 RDD 的編程模型中,一共有兩種運算元,Transformations 類運算元和 Actions 類運算元。開發者需要使用 Transformations 類運算元,定義並描述數據形態的轉換過程,然後調用 Actions 類運算元,將計算結果收集起來、或是物化到磁碟。

在這樣的編程模型下,Spark 在運行時的計算被劃分為兩個環節。

  1. 基於不同數據形態之間的轉換,構建計算流圖(DAG,Directed Acyclic Graph)
  2. 通過 Actions 類運算元,以回溯的方式去觸發執行這個計算流圖

換句話說,開發者調用的各類 Transformations 運算元,並不立即執行計算,當且僅當開發者調用 Actions 運算元時,之前調用的轉換運算元才會付諸執行。在業內,這樣的計算模式有個專門的術語,叫作“延遲計算”(Lazy Evaluation)。

二、Spark運算元分類

在 RDD 的開發框架下,哪些運算元屬於 Transformations 運算元,哪些運算元是 Actions 運算元呢?

這裡給出一張自己在極客看的課程中的圖

img

三、Transform運算元執行流程(源碼)

Map轉換算是 RDD 的經典轉換操作之一了.就以它開頭.Map的源碼如下:

image-20230224103912741

1. sc.clean(f)

首先掉了一個sc.clean(f) , 我們進到clean函數里看下:

image-20230224104117191

註釋中明確提到了這個函數的功能:clean 整理一個閉包,使其可以序列化併發送到任務.

這裡的代碼有些多,大概知道這個函數的功能是這樣就ok了,閉包的問題會在另一篇文章里仔細介紹

2. MapPartitionsRDD

進入到函數後源碼如下:

image-20230224105719758

這是一個MapPartitionsRDD。我們仔細看它的構成,從而來理解它是如何描述MapPartitionsRDD的.

2.1 var prev:RDD[T]

這裡的 prev 就是父RDD,f 則是Map中傳入的處理函數,除了這兩個就沒有了,也就是說明 RDD中沒有存儲具體的數據本身

這再次印證了轉換不會產生任何數據.它只是單純了記錄父RDD以及如何轉換的過程就完了,不會在轉換階段產生任何數據集

2.2 preservesPartitioning

preservesPartitioning 表示是否保持父RDD的分區信息.
如果為false(預設為false),則會對結果重新分區.也就是Map系預設都會分區
如果為true,保留分區. 則按照 firstParent 保留分區   

image-20230224110557226

可以看到根據 dependencies 找到其第一個父 RDD

image-20230224110711910

2.3 compute 計算邏輯
2.3.1 compute方法

RDD 抽象類要求其所有子類都必須實現 compute 方法,該方法接受的參數之一是一個Partition 對象,目的是計算該分區中的數據。

override def compute(split: Partition, context: TaskContext): Iterator[U] =
  f(context, split.index, firstParent[T].iterator(split, context))

可以看到,compute 方法調用當前 RDD 內的第一個父 RDD 的 iterator 方法,該方的目的是拉取父 RDD 對應分區內的數據。

iterator 方法會返回一個迭代器對象,迭代器內部存儲的每個元素即父 RDD 對應分區內已經計算完畢的數據記錄。得到的迭代器作為 f 方法的一個參數。fRDD 類的 map 方法中指定,即實際的轉換函數。

compute 方法會將迭代器中的記錄一一輸入 f 方法,得到的新迭代器即為所求分區中的數據。

其他 RDD 子類的 compute 方法與之類似,在需要用到父 RDD 的分區數據時候,就會調用 iterator 方法,然後根據需求在得到的數據之上執行粗粒度的操作。換句話說,compute 函數負責的是父 RDD 分區數據到子 RDD 分區數據的變換邏輯。

2.3.2 iterator方法

此方法的實現在 RDD 這個抽象類中

/**
 * 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 implementers of custom
 * subclasses of RDD.
 */
final def iterator(split: Partition, context: TaskContext): Iterator[T] = {
  if (storageLevel != StorageLevel.NONE) {
    getOrCompute(split, context)
  } else {
    computeOrReadCheckpoint(split, context)
  }
}

interator首先檢查 存儲級別 storageLevel:此處可參考RDD持久化

如果存儲級別不是NONE, 說明分區的數據說明分區的數據要麼已經存儲在文件系統當中,要麼當前 RDD 曾經執行過 cachepersise 等持久化操作,此時需要從存儲空間讀取分區數據,調用 getOrCompute 方法

image-20230224114953570

getOrCompute 方法會根據 RDD 編號:id分區編號:partition.index 計算得到當前分區在存儲層對應的塊編號:blockId,通過存儲層提供的數據讀取介面提取出塊的數據。

代碼中的這幾句註釋給的非常到位,大致的判斷順序如下:

  • 塊命中的情況:也就是數據之前已經成功存儲到介質中,這其中可能是數據本身就在存儲介質中(比如通過讀取HDFS創建的RDD),也可能是 RDD 在經過持久化操作並且經歷了一次計算過程,這個時候我們就能成功讀取數據並將其返回
  • 塊未命中的情況:可能是數據已經丟失,或者 RDD 經過持久化操作,但是是當前分區數據是第一次被計算,因此會出現拉取得到數據為 None 的情況。這就意味著我們需要計算分區數據,繼續調用 RDDcomputeOrReadCheckpoint 方法來計算數據,並將計算得到的數據緩存到存儲介質中,下次就無需再重覆計算。

如果當前RDD的存儲級別為 None,說明為未經持久化的 RDD,需要重新計算 RDD 內的數據,這時候調用 RDD 類的 computeOrReadCheckpoint 方法,該方法也在持久化 RDD 的分區獲取數據失敗時被調用。

image-20230224142431572

computeOrReadCheckpoint 方法會檢查當前 RDD 是否已經被標記成檢查點,如果未被標記成檢查點,則執行自身的 compute 方法來計算分區數據,否則就直接拉取父 RDD 分區內的數據。

需要註意的是,對於標記成檢查點的情況,當前 RDD 的父 RDD 不再是原先轉換操作中提供數據的父 RDD,而是被 Apache Spark 替換成一個 CheckpointRDD 對象,該對象中的數據存放在文件系統中,因此最終該對象會從文件系統中讀取數據並返回給 computeOrReadCheckpoint 方法

參考文章:

Cache 和 Checkpoint


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • 前言: https://www.cnblogs.com/LoveBB/p/17277662.html 什麼是範型 JDK 1.5開始引入Java泛型(generics)這個特性,該特性提供了編譯時類型安全檢測機制,允許程式員在編譯時檢測到非法的類型。停,業餘的客套話就不多說了,這些術語,講了N遍,不 ...
  • 還不會 Quartz?如果你還沒有接觸過Quartz,那麼你可能錯過了一個很棒的任務調度框架!Quartz 提供了一種靈活、可靠的方式來管理和執行定時任務,讓咱們的定時任務更加優雅。 ...
  • 1.標識符 程式中對類、變數等的命名,稱為標識符; 標識符命名規則: 由數字、字母、下劃線、美元符組成,不能以數字開頭; 嚴格區分大小寫; 不能與關鍵字或保留字重名; 標識符的命名最好能反應出其作用。 2.關鍵字 程式中對編譯器有特殊意義的詞,例如class被用來定義類,當程式執行遇到class時, ...
  • 模型之間的關係(Relations Between Models) 上一章介紹了為包含基本欄位的模型創建自定義視圖。然而,在任何真實的業務場景中,我們都需要不止一個模型。此外,模型之間的鏈接是必要的。人們可以很容易地想象一個模型包含客戶,另一個模型則包含用戶列表。你可能需要參考任何現有業務模型上的客 ...
  • ChatGPT是一個基於GPT-3.5架構的自然語言處理工具,它具有文本生成、文本分類、對話生成等多種能力。作為一種強大的自然語言處理工具,ChatGPT可以應用於智能客服、智能問答、內容創作等多個領域。如果您對ChatGPT感興趣,可以通過關註本公眾號瞭解更多信息,並體驗基於ChatGPT的小程式... ...
  • Spring Spring為簡化開發而生,讓程式員只關心核心業務的實現,儘可能的不在關註非業務邏輯代碼(事務控制,安全日誌等)。 1,Spring八大模塊 這八大模塊組成了Spring 1.1 Spring Core模塊 這是Spring框架的最基礎的部分,它提供了依賴註入(DependencyIn ...
  • React Native 備忘清單 適合初學者的綜合 React Native 備忘清單,在開始 React Native 之前需要先掌握 react 庫入門,為開發人員分享快速參考備忘單。 React Native (簡稱RN)是Facebook於2015年4月開源的跨平臺移動應用開發框架,是Fa ...
  • React 備忘清單 IT寶庫整理適合初學者入門的React開發速查備忘清單,為開發人員分享快速參考備忘單。 React是用於構建用戶界面的JavaScript庫,起源於Facebook的內部項目,該公司對市場上所有 JavaScript MVC框架都不滿意,決定自行開發一套,用於架設Instagr ...
一周排行
    -Advertisement-
    Play Games
  • 下麵是一個標準的IDistributedCache用例: public class SomeService(IDistributedCache cache) { public async Task<SomeInformation> GetSomeInformationAsync (string na ...
  • 這個庫提供了在啟動期間實例化已註冊的單例,而不是在首次使用它時實例化。 單例通常在首次使用時創建,這可能會導致響應傳入請求的延遲高於平時。在註冊時創建實例有助於防止第一次Request請求的SLA 以往我們要在註冊的時候實例單例可能會這樣寫: //註冊: services.AddSingleton< ...
  • 最近公司的很多項目都要改單點登錄了,不過大部分都還沒敲定,目前立刻要做的就只有一個比較老的項目 先改一個試試手,主要目標就是最短最快實現功能 首先因為要保留原登錄方式,所以頁面上的改動就是在原來登錄頁面下加一個SSO登錄入口 用超鏈接寫的入口,頁面改造後如下圖: 其中超鏈接的 href="Staff ...
  • Like運算符很好用,特別是它所提供的其中*、?這兩種通配符,在Windows文件系統和各類項目中運用非常廣泛。 但Like運算符僅在VB中支持,在C#中,如何實現呢? 以下是關於LikeString的四種實現方式,其中第四種為Regex正則表達式實現,且在.NET Standard 2.0及以上平... ...
  • 一:背景 1. 講故事 前些天有位朋友找到我,說他們的程式記憶體會偶發性暴漲,自己分析了下是非托管記憶體問題,讓我幫忙看下怎麼回事?哈哈,看到這個dump我還是非常有興趣的,居然還有這種游戲幣自助機類型的程式,下次去大玩家看看他們出幣的機器後端是不是C#寫的?由於dump是linux上的程式,剛好win ...
  • 前言 大家好,我是老馬。很高興遇到你。 我們為 java 開發者實現了 java 版本的 nginx https://github.com/houbb/nginx4j 如果你想知道 servlet 如何處理的,可以參考我的另一個項目: 手寫從零實現簡易版 tomcat minicat 手寫 ngin ...
  • 上一次的介紹,主要圍繞如何統一去捕獲異常,以及為每一種異常添加自己的Mapper實現,並且我們知道,當在ExceptionMapper中返回非200的Response,不支持application/json的響應類型,而是寫死的text/plain類型。 Filter為二方包異常手動捕獲 參考:ht ...
  • 大家好,我是R哥。 今天分享一個爽飛了的面試輔導 case: 這個杭州兄弟空窗期 1 個月+,面試了 6 家公司 0 Offer,不知道問題出在哪,難道是杭州的 IT 崩盤了麽? 報名面試輔導後,經過一個多月的輔導打磨,現在成功入職某上市公司,漲薪 30%+,955 工作制,不咋加班,還不捲。 其他 ...
  • 引入依賴 <!--Freemarker wls--> <dependency> <groupId>org.freemarker</groupId> <artifactId>freemarker</artifactId> <version>2.3.30</version> </dependency> ...
  • 你應如何運行程式 互動式命令模式 開始一個互動式會話 一般是在操作系統命令行下輸入python,且不帶任何參數 系統路徑 如果沒有設置系統的PATH環境變數來包括Python的安裝路徑,可能需要機器上Python可執行文件的完整路徑來代替python 運行的位置:代碼位置 不要輸入的內容:提示符和註 ...