Windows 编程 php java apache wordpress google Firefox shell Ubuntu nginx linux 开源 mysql Android centos 微软 程序员 云计算 Python

揭秘Spark應用性能調優

引言:在多臺機器上分布數據以及處理數據是Spark的核心能力,即我們所說的大規模的數據集處理。為了充分利用Spark特性,應該考慮一些調優技術。本文每一小節都是關於調優技術的,並給出了如何實現調優的必要步驟。
本文選自《Spark GraphX實戰》。

1 用緩存和持久化來加速 Spark

  我們知道Spark 可以通過 RDD 實現計算鏈的原理 :轉換函數包含在 RDD 鏈中,但僅在調用 action 函數後才會觸發實際的求值過程,執行分布式運算,返回運算結果。要是在 同一 RDD 上重復調用 action 會發生什麽?

RDD 持久化

  一般 RDD 不會保留運算結果,如果再次調用 action 函數,整個 RDD 鏈會重新 運算。有些情況下這不會有問題,但是對於許多機器學習任務和圖處理任務,這就 是很大的問題了。通常需要多次叠代的算法,在同一個 RDD 上執行很多次,反復 地重新加載數據和重新計算會導致時間浪費。更糟糕的是,這些算法通常需要很長 的 RDD 鏈。
  看來我們需要另一種方式來充分利用集群可用內存來保存 RDD 的運算結果。 這就是 Spark 緩存(緩存也是 Spark 支持的一種持久化類型)。
  要在內存中緩存一個 RDD,可以調用 RDD 對象的 cache 函數。以下在 spark- Shell 中執行的代碼,會計算文件的總行數,輸出文件內容 :

val filename = "..."
val rdd1 = sc.textFile(filename).cache
rdd1.count
rdd1.collect
如果不調用 cache 函數,當 count 和 collect 這兩個 action 函數被調用時, 會導致執行從存儲系統中讀文件兩次。調用了 cache 函數,第一個 action 函數(count 函數)會把它的運算結果保留在內存中,在執行第二個 action 函數(collection 函數)時,會直接在使用緩存的數據上繼續運算,而不需要重新計算整個 RDD 鏈。 即使通過轉換緩存的 RDD,生成新的 RDD,緩存的數據仍然可用。下面的代碼會找出所有的註釋行(以 # 開始的行數據)。

val rdd2 =rdd1.filter(_.startsWith("#"))
rdd2.collect
  因為 rdd2 源於已緩存的 rdd1,rdd1 已經把它的運算結果緩存在內存中了, 所以 rdd2 也就不需要重新從存儲系統中讀取數據。

註意:cache 方法作為一個標誌表示 RDD 應當緩存,但並不是立即緩存。 緩存發生在當前 RDD 在下一次要被計算的時候。

持久化等級

  如上所述,緩存是其中一種持久化類型。下表列出了 Spark 支持的所有持久 化等級。
        
  每個持久化等級都定義在單例對象 StorageLevel 中。例如,調用 rdd.persist(StorageLevel.MEMORY_AND_DISK)方法會把 RDD 設置成內存和磁盤緩 存。 cache 方法內部也是調用 rdd.persist(StorageLevel.MEMORY_ONLY)。

註意 :其他的持久化等級,如 MEMORY_ONLY2、MEMORY_AND_ DISK2 等,也是可用的。它們會復制 RDD到集群的其他節點上,以便 提供容錯能力。這些內容超出了本書範圍,感興趣的讀者可以看看 Petar Zec' evic' 和 Marko

Bonac' i(Manning, 2016)的書 Spark in Action,這本書更 深入地介紹了 Spark 容錯方面的內容。

圖的持久化

  無論什麽時候,通過 Graph 對象調用一些函數如 mapVertices 或 aggregateMessages, 這些操作都是基於下層的 RDD 實現的。
  Graph 對象提供了基於頂點 RDD 和邊 RDD 方便的緩存和持久化方法。

在合適的時機反持久化

  雖然看起來緩存是一個應該被到處使用的好東西,但是用得太多也會讓人過度依賴它。
  當緩存越來越多的 RDD 後,可用的內存就會減少。最終 Spark 會把分區數據從 內存中逐出(使用最少最近使用算法,LRU)。同時,緩存過多的 JAVA 對象,JVM 垃圾回收高耗是不可避免的。這就是為什麽當緩存不再被使用時很有必要調用 un- persist 方法。對叠代算法而言,在循環中常用下面的方法調用模式 :

調用 Graph 的 cache 或 persist 方法。
調用 Graph 的 action 函數,觸發 Graph 下面的 RDD 被緩存……
執行算法主體的其余部分。
在循環體的最後部分,反持久化,即釋放緩存。
提示 :用Pregel API的好處是,它已經在內部做了緩存和釋放緩存的操作。

何時不用緩存

  不能盲目地在內存中緩存 RDD。要考慮數據集會被訪問多少次以及每次訪問時 重計算和緩存的代價對比,重計算也可能比增加內存的方式付出的代價小。
  毫無疑問,如果僅僅讀一次數據集,緩存 RDD 就毫無意義,這還會讓作業運 行得更慢,特別是用了有序列化的持久化等級。

2.checkpointing

  圖算法中一個常用的模式是用每個叠代過程中運算後的新數據更新圖。這意味 著,實際構成圖的頂點 RDD 亦或邊 RDD 的鏈會變得越來越長。

定義 :當 RDD 由逐級繼承的祖先 RDD 鏈形成時,我們說從 RDD 到 根 RDD 的路徑是其譜系。

  下面清單所示的示例是一個簡單的算法,可生成一個新頂點集並更新圖。這個 算法叠代的次數由變量 iterations 控制。
        
  上述代碼每一次調用 joinVertices 都會增加一個新 RDD 到頂點 RDD 鏈中。 顯然我們需要使用緩存來確保在每次叠代中避免重新計算 RDD 鏈,但這並不能改變一個事實,那就是有一個不斷增長的子 RDD 到父 RDD 的對象引用列表。
  這樣的後果是,如果運行叠代次數過多,運行的代碼中最終會爆出 Stack- OverflowError 棧溢出錯誤。通常叠代 500次就會出現棧溢出。
  而由 RDD 提供並且被 Graph 繼承的一個特性 :checkpointing,能解決長 RDD 譜系問題。下面清單中的代碼示範了如何使用 checkpointing,這樣就可以持續輸出 頂點,更新結果圖。
              
  一個標記為 checkpointing 的 RDD 會把 RDD 保存到一個 checkpoint 目錄,然 後指向父 RDD 的連接被切斷,即切斷了 lineage 譜系。一個標記為 checkpointing 的 Graph 會導致下面的頂點 RDD 和邊 RDD 做 checkpoint。
  調用 SparkContext.setCheckpointDir 來設置 checkpoint 目錄,指定一個 共享存儲系統的文件路徑,如 HDFS。
  如前面的代碼清單所示,必須在調用 RDD 任何方法之前調用 checkpoint,這 是因為 checkpointing 是一個相當耗時的過程(畢竟需要把圖寫入磁盤文件),通常 需要不斷地 checkpoint 避免棧溢出錯誤,一般可以每 100 次叠代做一次 checkpoint。

註意 :一個加速 checkpointing 的選擇是 checkpoint 到 Tachyon(已 更名為 Alluxio),而不是checkpoint 到標準的文件系 統。Alluxio,來自 AMPLab,是一個“以內存為中心的有容錯能力的分布式文件系統,它能讓Spark 這類集群框架加速訪問共享在內存中的文件”。

3 通過序列化降低內存壓力

  內存壓力(內存不夠用)往往是 Spark 應用性能差和容易出故障的主要原因 之一。這些問題通常表現為頻繁的、耗時的 JVM 垃圾回收和“內存不足”的錯 誤。checkpointing 在這裏也不能緩解內存壓力。遇到這種問題,首先要考慮序列化 Graph 對象。

定義 :數據序列化,Data serialization,是把 JVM 裏表示的對象實 例轉換(序列化)成字節流 ;把字節流通過網絡傳輸到另一個 JVM 進程 中 ;在另一個 JVM 進程中,字節流可以被“反序列化”為一個對象實例。Spark用序列化的方式,可以在網絡間傳輸對象,也可以把序列化後的字節流緩存在內存中。

  要用序列化,可以選用 persist 中下面的 StorageLevels :

StorageLevel.MEMORY_ONLY_SER
StorageLevel.MEMORY_AND_DISK_SER
序列化節省了空間,同時序列化和反序列化也會增加 CPU 的開銷。

使用 Kryo 序列化

  Spark 默認使用 JavaSerializer 來序列化對象,這是一個低效的 Java 序列化框架,一個更好的選擇是選用 Kryo。Kryo 是一個開源的 Java 序列化框架,提供了 快速高效的序列化能力。
  Spark 中使用 Kryo 序列 化,只需要設置 spark.serializer 參數為 org. apache.spark.serializer.KryoSerializer,如這樣設置命令行參數 :

spark-shell --conf "spark.serializer=org.apache.spark.serializer.KryoSerializer"
  要是每次都這樣設置參數,會很煩瑣。可以在 $Spark_HOME/conf/spark-
defaults.conf 這個配置文件中,用標準的屬性文件語法(用 Tab 分隔作為一行),把 spark.serializer 等參數及其對應的值寫入這個配置文件,如下所示 :

spark.serializer org.apache.spark.serializer.KryoSerializer
  為保證性能最佳,Kryo 要求註冊要序列化的類,如果不註冊,類名也會被序列 化在對象字節碼裏,這樣對性能有較大影響。幸運的是,Spark 對其框架裏用到的 類做了自動註冊 ;但是,如果應用程序代碼裏有自定義的類,恰好這些自定義類也 要用 Kryo 序列化,那就需要調用 SparkConf.registerKryoClasses 函數來手 動註冊。下面的清單展示了如何註冊 Person 這個自定義類。
         

檢查 RDD 大小

  在應用程序調優時,常常需要知道 RDD 的大小。這就很棘手,因為文件或數 據庫中對象的大小和 JVM 中對象占用多少內存沒有太大關系。
  一個小技巧是,先將 RDD 緩存到內存中,然後到 Spark UI 中的 Storage 選項卡, 這裏記錄著 RDD 的大小。要衡量配置了序列化的效果,用這個方法也可以。

延伸阅读

    评论