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
  • 1. 說明 /* Performs operations on System.String instances that contain file or directory path information. These operations are performed in a cross-pla ...
  • 視頻地址:【WebApi+Vue3從0到1搭建《許可權管理系統》系列視頻:搭建JWT系統鑒權-嗶哩嗶哩】 https://b23.tv/R6cOcDO qq群:801913255 一、在appsettings.json中設置鑒權屬性 /*jwt鑒權*/ "JwtSetting": { "Issuer" ...
  • 引言 集成測試可在包含應用支持基礎結構(如資料庫、文件系統和網路)的級別上確保應用組件功能正常。 ASP.NET Core 通過將單元測試框架與測試 Web 主機和記憶體中測試伺服器結合使用來支持集成測試。 簡介 集成測試與單元測試相比,能夠在更廣泛的級別上評估應用的組件,確認多個組件一起工作以生成預 ...
  • 在.NET Emit編程中,我們探討了運算操作指令的重要性和應用。這些指令包括各種數學運算、位操作和比較操作,能夠在動態生成的代碼中實現對數據的處理和操作。通過這些指令,開發人員可以靈活地進行算術運算、邏輯運算和比較操作,從而實現各種複雜的演算法和邏輯......本篇之後,將進入第七部分:實戰項目 ...
  • 前言 多表頭表格是一個常見的業務需求,然而WPF中卻沒有預設實現這個功能,得益於WPF強大的控制項模板設計,我們可以通過修改控制項模板的方式自己實現它。 一、需求分析 下圖為一個典型的統計表格,統計1-12月的數據。 此時我們有一個需求,需要將月份按季度劃分,以便能夠直觀地看到季度統計數據,以下為該需求 ...
  • 如何將 ASP.NET Core MVC 項目的視圖分離到另一個項目 在當下這個年代 SPA 已是主流,人們早已忘記了 MVC 以及 Razor 的故事。但是在某些場景下 SSR 還是有意想不到效果。比如某些靜態頁面,比如追求首屏載入速度的時候。最近在項目中回歸傳統效果還是不錯。 有的時候我們希望將 ...
  • System.AggregateException: 發生一個或多個錯誤。 > Microsoft.WebTools.Shared.Exceptions.WebToolsException: 生成失敗。檢查輸出視窗瞭解更多詳細信息。 內部異常堆棧跟蹤的結尾 > (內部異常 #0) Microsoft ...
  • 引言 在上一章節我們實戰了在Asp.Net Core中的項目實戰,這一章節講解一下如何測試Asp.Net Core的中間件。 TestServer 還記得我們在集成測試中提供的TestServer嗎? TestServer 是由 Microsoft.AspNetCore.TestHost 包提供的。 ...
  • 在發現結果為真的WHEN子句時,CASE表達式的真假值判斷會終止,剩餘的WHEN子句會被忽略: CASE WHEN col_1 IN ('a', 'b') THEN '第一' WHEN col_1 IN ('a') THEN '第二' ELSE '其他' END 註意: 統一各分支返回的數據類型. ...
  • 在C#編程世界中,語法的精妙之處往往體現在那些看似微小卻極具影響力的符號與結構之中。其中,“_ =” 這一組合突然出現還真不知道什麼意思。本文將深入剖析“_ =” 的含義、工作原理及其在實際編程中的廣泛應用,揭示其作為C#語法奇兵的重要角色。 一、下劃線 _:神秘的棄元符號 下劃線 _ 在C#中並非 ...