Spark的閉包清理機制怎么理解

Spark的閉包清理機制怎么理解，相信很多沒有經驗的人對此束手無策，為此本文總結了問題出現的原因和解決方法，通過這篇文章希望你能解決這個問題。

關于Spark任務運行時發生不可序列話的問題。今天就統一講解一下這塊的內容。

首先，要先讀懂scala的閉包是怎么回事兒。

簡單理解scala的閉包

接著就是要理解Spark 算子閉包生成及我們編寫的閉包執行的原理。接下來我們就拿map和mapPartition兩個算子來開啟本文講解：

def map[U: ClassTag](f: T => U): RDD[U] = withScope {
    val cleanF = sc.clean(f)
    new MapPartitionsRDD[U, T](this, (context, pid, iter) => iter.map(cleanF))
  }

def mapPartitions[U: ClassTag](
      f: Iterator[T] => Iterator[U],
      preservesPartitioning: Boolean = false): RDD[U] = withScope {
    val cleanedF = sc.clean(f)
    new MapPartitionsRDD(
      this,
      (context: TaskContext, index: Int, iter: Iterator[T]) => cleanedF(iter),
      preservesPartitioning)
  }

對于任務劃分，調度，執行，結果返回的原理浪尖就不在這里擴展了，浪尖在知識星球里分享過一套Spark 源碼的視頻，可以參考閱讀。

map和mapPartitions的區別面試常考的，對于兩者的區別從源碼里看很明顯，一個是f被迭代器迭代調用，一個是f的參數是迭代器。浪尖很早以前發過一篇文章，徹底講解過foreach和foreachPartition的區別。可以參考理解

Spark源碼系列之foreach和foreachPartition的區別

回到正題，之所以會發生不可序列化的錯誤，主要原因是傳遞給map的f函數不是在driver端執行的，所以會被序列化傳輸到executor節點，然后在executor節點反序列化然后執行。假如f函數里引用了map外部不可序列化的對象就會報不可序列化的異常。

但是，很多時候我們并沒有直接去在閉包里使用不可序列化的對象，這個時候報異常就有點不合適了。比如下面的例子：

*   class SomethingNotSerializable {
   *     def someValue = 1
   *     def scope(name: String)(body: => Unit) = body
   *     def someMethod(): Unit = scope("one") {
   *       def x = someValue
   *       def y = 2
   *       scope("two") { println(y + 1) }
   *     }
   *   }

此示例中，scope(two) 不可序列化，因為它引用了scope(one)(通過y)，而scope(one)引用了SomethingNotSerializable（通過someValue）。但是，其實scope(two)并不直接依賴于SomethingNotSerializable。假如這種情況下拋出不可序列化異常就不科學了，所以Spark會對閉包進行一些清理操作，也即是本文中所要講的。

主要工具類是ClosureCleaner。該工具的主要作用是遍歷閉包的層次結構，并且將沒有被閉包實際引用的鏈路設置為null，但是仍然包含在已經編譯的匿名類中。請注意直接修改封閉中的閉包是不安全的，因為可能有其他代碼路徑會依賴于他們。所以，我們會克隆封閉中的閉包并且相應地設置父指針。

默認情況下，可以傳遞清除閉包。這就意味著，我們需要檢測封閉對象是否由起始對象實際引用，（要么直接引用要么間接引用），如果沒有被實際使用則從層次結構中切斷這些閉包。換句話說，除了清空無用字段的引用之外，也會將沒有被起始閉包引用的引用封閉對象的父指針清空。傳遞性的確定是通過遍歷閉包所調用的

再回到前面的例子，scope(two) 不可序列化，因為它引用了scope(one)(通過y)，而scope(one)引用了SomethingNotSerializable（通過someValue）。但是，其實scope(two)并不直接依賴于SomethingNotSerializable。這就意味著我們可以安全的將其副本scope(one)的父指針清空，同時將其設置為scope（two）的父級，這樣scope(two)就不再需要間接傳遞引用SomethingNotSerializable了。

解決方法

實現序列化是最直接的，假如不能的話。那就讀下面的話：

那么為了不實現序列化還能盡量避免不可序列化錯誤，就不要在map等算子里引用外部變量，而是直接在算子中實例化，假如每次實例化代價高，那就使用mapPartitions。

看完上述內容，你們掌握Spark的閉包清理機制怎么理解的方法了嗎？如果還想學到更多技能或想了解更多相關內容，歡迎關注億速云行業資訊頻道，感謝各位的閱讀！