Spark DAG、Stage 劃分與 Task 調度底層原理深度剖析

---

核心知識點詳解

1. DAG (Directed Acyclic Graph) 的構建過程回顧

Spark 應用程序的執行始于 RDD 的創建和一系列的轉換操作 (Transformations)。這些轉換操作（如 map(), filter(), reduceByKey() 等）并不會立即執行計算，而是構建一個邏輯執行計劃，即一個有向無環圖 (DAG)。這種機制稱為 惰性求值 (Lazy Evaluation)。

• 轉換操作: 當你在 RDD 上調用轉換操作時，Spark 只是在 DAG 中添加一個新的節點，代表一個新的 RDD 及其與父 RDD 的關系。
• 惰性求值: 只有當遇到行動操作 (Actions)（如 count(), collect(), saveAsTextFile() 等）時，Spark 才會真正觸發計算。此時，Spark 會從 DAG 的末端（行動操作所在的 RDD）向前追溯，構建并執行完整的物理執行計劃。
• 依賴關系: DAG 中的每個節點代表一個 RDD，節點之間的邊表示 RDD 之間的依賴關系（即一個 RDD 是另一個 RDD 的父 RDD）。

2. Stage 的劃分過程與底層原理

Spark 將 DAG 劃分為 Stage 的核心原則是基于 RDD 之間的依賴類型。

關鍵概念：RDD 依賴 (Dependency)

RDD 之間的依賴關系分為兩種：

1. 窄依賴 (Narrow Dependency):

   • 定義: 父 RDD 的一個分區只對應子 RDD 的一個或有限個分區。這種關系意味著每個父分區最多只被一個子分區消費。

   • 特點: 1對1 或 1對少數（如 Union）。

   • 例子: map(), filter(), union(), coalesce() (如果減少分區且不觸發 Shuffle)。

   • 優勢

     :

     • 無 Shuffle: 數據流是管道式的，可以在父 RDD 的分區計算完成后直接傳遞給子 RDD 的對應分區，無需跨網絡或進程傳輸數據。
     • 故障恢復高效: 如果某個分區丟失，Spark 只需要重新計算其上游的少數相關分區即可恢復。

   • Stage 內部: 窄依賴操作可以在同一個 Stage 內進行，因為它們是流水線式的，無需等待所有父分區完成即可開始子分區計算。

2. 寬依賴 (Wide Dependency) / Shuffle Dependency:

   • 定義: 父 RDD 的一個分區可能對應子 RDD 的所有分區，數據在處理過程中需要進行重新洗牌 (shuffle) 才能完成計算。這意味著數據需要跨機器或跨進程進行網絡傳輸和重新聚合。

   • 特點: 1對多的關系。

   • 例子: groupByKey(), reduceByKey(), sortByKey(), join(), repartition()。

   • 劣勢

     :

     • 需要 Shuffle: 涉及大量的磁盤 I/O、網絡傳輸、數據序列化/反序列化，開銷較大，是 Spark 應用程序的主要性能瓶頸之一。
     • 故障恢復復雜: 某個分區丟失可能需要重新計算整個 Shuffle 的上游部分。

   • Stage 邊界: 寬依賴是劃分 Stage 的主要依據。 每當 Spark 的 DAGScheduler 遇到一個寬依賴操作時，它就會在其前面切分出一個新的 Stage。

Stage 劃分的底層原理 (如何分解 Stage)

1. 從行動操作 (Action) 開始倒推: 當用戶觸發一個行動操作（如 collect()）時，Spark 的核心調度組件 DAGScheduler 被激活。它會從這個最終 RDD 開始，沿著 RDD 之間的依賴關系圖反向遍歷。

2. 遇到寬依賴就切分 Stage

   :

   • DAGScheduler 會將一連串的窄依賴操作放入同一個 Stage。這些操作可以高效地以管道方式串行執行，無需中間寫入磁盤。
   • 每當 DAGScheduler 在反向遍歷時遇到一個寬依賴 (Shuffle Dependency)，就意味著數據必須進行 Shuffle。為了執行 Shuffle，Spark 需要等待上一個 Stage 的所有 Task 都完成，并將它們的輸出數據寫入磁盤（或內存），作為 Shuffle 的“Map”階段輸出。然后，下一個 Stage 的 Task 才能開始讀取這些數據，作為 Shuffle 的“Reduce”階段輸入。
   • 因此，寬依賴成為了 Stage 的邊界。一個寬依賴操作會形成一個新的 Stage 的開始，而其所有上游的窄依賴操作則構成了一個或多個完整的 Stage。

3. 生成物理執行計劃: DAGScheduler 負責將邏輯的 RDD 轉換圖轉換為物理的 Stage 圖。每個 Stage 對應著一組可以通過管道化執行的 Task。

Stage 劃分示例:

考慮一個 RDD 轉換鏈：RDD1.map().filter().reduceByKey().sortByKey().collect()

                                          (Stage 1: MapPartitions)
[RDD1] --map--> [RDD2] --filter--> [RDD3] (窄依賴操作管道化執行)|(寬依賴 - Shuffle Dependency)|(Stage 2: ShuffleMapStage)[RDD4_ShuffleRead] --reduceByKey--> [RDD5] (窄依賴操作管道化執行)|(寬依賴 - Shuffle Dependency)|(Stage 3: ResultStage)[RDD6_ShuffleRead] --sortByKey--> [RDD7] (窄依賴操作管道化執行)|(行動操作: collect)

在這個例子中，reduceByKey 和 sortByKey 都引入了 Shuffle。因此，整個 DAG 會被劃分為 3 個 Stage：

• Stage 1: 包含 map() 和 filter() 操作。這些操作可以在同一組 Task 中直接處理 RDD1 的分區數據。
• Stage 2: 始于 reduceByKey()。它依賴于 Stage 1 的 Shuffle 輸出。
• Stage 3: 始于 sortByKey()。它依賴于 Stage 2 的 Shuffle 輸出。

3. Task 數量的決定因素

每個 Stage 被分解為多個 Task 并行執行。Task 的數量直接決定了該 Stage 的并行度，進而影響 Spark 應用程序的整體性能。

Task 的數量主要由以下因素決定：

1. RDD 的分區數 (Number of Partitions):

   • 這是最主要且最直接的因素。在大多數情況下，一個 Stage 中的 Task 數量等于該 Stage 的最后一個 RDD 的分區數。

   • Spark 調度器會為 RDD 的每個分區分配一個 Task，由該 Task 負責處理對應分區的數據。

   • 舉例

     :

     • sc.parallelize(numbers)：默認情況下，parallelize 創建的 RDD 的分區數通常等于你的 Spark 應用程序可用的 CPU 核心數（在 local 模式下）或 spark.default.parallelism 配置的值。
     • sc.parallelize(numbers, 3)：如果你明確指定了 3 個分區，那么這個 RDD 及其所有通過窄依賴衍生的 RDD 都將有 3 個分區，從而導致后續的 Task 數量為 3（直到遇到寬依賴）。

2. 寬依賴 (Shuffle) 的影響:

   • 當發生 Shuffle 時，新的 RDD 的分區數可以通過以下方式控制：
     • spark.sql.shuffle.partitions: 對于 Spark SQL (DataFrame/Dataset API) 的 Shuffle 操作，這個配置參數默認是 200，它決定了 Shuffle 輸出的分區數，從而影響下一個 Stage 的 Task 數量。
     • 通過轉換操作的參數顯式指定，例如 reduceByKey(numPartitions)、join(otherRDD, numPartitions)。
     • repartition() 操作總是會觸發 Shuffle，并允許你指定新的分區數。

3. spark.default.parallelism 配置:

   • 這是 Spark 默認的并行度設置，影響著 parallelize 等操作創建 RDD 的初始分區數，以及在集群模式下未明確指定分區數時的默認行為。

4. 輸入數據源的特性 (Input Splits):

   • 對于從 HDFS 或其他分布式文件系統讀取數據，RDD 的初始分區數通常由輸入文件的塊大小或切片 (split) 數量決定。一個 HDFS 塊或一個輸入切片通常對應一個 RDD 分區，進而對應一個 Task。

5. repartition() 或 coalesce() 操作:

   • 這些轉換操作允許你顯式地改變 RDD 的分區數，從而直接控制后續 Stage 的 Task 數量。
   • repartition() 總是會觸發 Shuffle，因為它可能增加或減少分區，并且通常需要重新分配數據。
   • coalesce() 旨在優化分區，如果只是減少分區數且不觸發 Shuffle (shuffle=false)，它可能是窄依賴，通過合并現有分區來減少 Task；但如果是增加分區數或強制 Shuffle (shuffle=true)，它也會觸發 Shuffle。

6. 集群資源:

   • 雖然 RDD 的分區數決定了 Task 的邏輯數量，但 Task 的實際并行執行數量最終受限于集群中可用的 CPU 核心、內存等資源。例如，如果你有 1000 個 Task，但集群只有 100 個核心，那么最多只能同時運行 100 個 Task。

4. 底層執行流程串聯

整個 Spark 應用程序的執行流程可以串聯如下：

1. 用戶代碼 (RDD 轉換): 用戶編寫 RDD 轉換代碼，定義了數據處理的邏輯轉換步驟，構建了一個抽象的 DAG。這些操作是惰性求值的，不會立即觸發計算。

2. 行動操作觸發: 當遇到 collect()、count()、saveAsTextFile() 等行動操作時，Spark 的 DAGScheduler 被激活，它負責將邏輯 DAG 轉化為物理執行計劃。

3. DAGScheduler 構建 Stage

   :

   • DAGScheduler 從行動操作對應的最終 RDD 開始，沿著 RDD 之間的依賴關系圖反向遍歷。
   • 它識別出所有的寬依賴 (Shuffle Dependency)，并將這些寬依賴作為 Stage 的邊界。
   • 每一個連續的窄依賴操作鏈條，都會被歸入同一個 Stage。每個 Stage 內部的 Task 可以管道化執行，無需等待中間結果寫入磁盤。
   • 每個 Stage 都會生成一個 TaskSet，其中包含針對該 Stage 所有分區的一組 Task。

4. TaskScheduler 提交 Task

   :

   • DAGScheduler 將這些構建好的 TaskSet 提交給 TaskScheduler。
   • TaskScheduler 負責將 Task 發送到集群的 Executor 上執行。它管理 Task 的生命周期，處理 Task 失敗和重試。TaskScheduler 會根據集群中可用的資源來調度 Task。

5. Executor 執行 Task

   :

   • Executor 進程接收到 Task 后，在其 JVM 進程中啟動一個線程來執行該 Task。
   • Task 在 Executor 的一個 CPU 核心上運行，處理其分配到的 RDD 分區數據。
   • 如果 Task 屬于一個 Shuffle Stage 的上游 (Map 階段)，它會在處理完數據后，將 Shuffle 輸出（通常是中間數據）寫入 Executor 所在機器的本地磁盤。
   • 如果 Task 屬于一個 Shuffle Stage 的下游 (Reduce 階段)，它會從上游 Task 的 Shuffle 輸出中拉取 (fetch) 數據，并進行聚合計算。

6. 結果返回: 當所有 Stage 的所有 Task 都成功完成后，最終結果會返回給驅動程序 (SparkContext)，或者根據行動操作的類型進行存儲。

---

追問與拓展 (Follow-up Questions & Extensions) 追問與拓展（Follow-up Questions & Extensions）

作為面試官，在聽到這樣的回答后，我可能會進一步追問，以評估你更深層次的理解和實踐經驗：

1. Stage 失敗與重試：

   “如果一個 Stage 中的某個 Task 失敗了，Spark 是如何處理的？會整個 Stage 重試嗎？還是只會重試失敗的 Task？這將如何影響效率？” (考察 Spark 的故障恢復機制和容錯性)

2. repartition 與 coalesce 的關鍵區別：

   “你提到了 repartition 和 coalesce 都可以改變 RDD 的分區數。它們之間有什么關鍵區別？在什么情況下你會選擇 coalesce(numPartitions, false) 而不是 repartition(numPartitions)？” (考察對 Shuffle 優化的理解和實際應用場景)

3. Shuffle 調優：

   “你認為 Shuffle 是 Spark 性能瓶頸的主要原因之一。在實際工作中，你會采取哪些策略來優化 Shuffle 的性能？請舉例說明，比如數據傾斜 (Data Skew) 的處理。” (考察實際的性能調優經驗和問題解決能力)

4. YARN/Mesos/Kubernetes 資源管理：

   “在集群管理器（如 YARN 或 Kubernetes）中，Task 和 Executor 是如何映射到物理資源的？spark.executor.cores 和 spark.executor.memory 這些參數是如何影響 Task 調度和資源利用的？” (考察 Spark 與集群資源管理器的集成和資源配置的理解)

5. DataFrame/Dataset API 的 Stage 劃分：

   “你主要以 RDD 解釋了 Stage 劃分。那么在使用 DataFrame/Dataset API 時，Stage 劃分的原理有何異同？Catalyst 優化器在其中扮演什么角色？” (考察對 Spark SQL 優化器工作原理的理解)

6. Spark UI 的作用與性能分析：

   “現在你的程序已經成功運行并輸出了結果。Spark UI 對你理解上述 DAG、Stage 和 Task 的執行過程有什么幫助？你會在 Spark UI 中關注哪些指標來分析程序的性能，以及如何從這些指標中發現潛在問題？” (考察實際工具使用和性能分析能力)