《《Spark 官方文檔》Spark調優》要點：
本文介紹了《Spark 官方文檔》Spark調優，希望對您有用。如果有疑問，可以聯系我們。

Spark調優

由于大部分Spark計算都是在內存中完成的,所以Spark程序的瓶頸可能由集群中任意一種資源導致,如：CPU、網絡帶寬、或者內存等.最常見的情況是,數據能裝進內存,而瓶頸是網絡帶寬;當然,有時候我們也需要做一些優化調整來減少內存占用,例如將RDD以序列化格式保存( storing RDDs in serialized form ).本文將主要涵蓋兩個主題：1.數據序列化(這對于優化網絡性能極為重要);2.減少內存占用以及內存調優.同時,我們也會提及其他幾個比較小的主題.

數據序列化

序列化在任何一種分布式應用性能優化時都扮演幾位重要的角色.如果序列化格式序列化過程緩慢,或者需要占用字節很多,都會大大拖慢整體的計算效率.通常,序列化都是Spark應用優化時首先需要關注的地方.Spark著眼于要達到便利性(允許你在計算過程中使用任何Java類型)和性能的一個平衡.Spark主要提供了兩個序列化庫：

Java serialization : 默認情況,Spark使用Java自帶的ObjectOutputStream 框架來序列化對象,這樣任何實現了 java.io.Serializable 接口的對象,都能被序列化.同時,你還可以通過擴展 java.io.Externalizable 來控制序列化性能.Java序列化很靈活但性能較差,同時序列化后占用的字節數也較多.

Kryo serialization : Spark還可以使用Kryo 庫(版本2)提供更高效的序列化格式.Kryo的序列化速度和字節占用都比Java序列化好很多(通常是10倍左右),但Kryo不支持所有實現了 Serializable 接口的類型,它需要你在程序中 register 需要序列化的類型,以得到最佳性能.

要切換到使用 Kryo,你可以在 SparkConf 初始化的時候調用 conf.set(“spark.serializer”, “org.apache.spark.serializer.KryoSerializer”).這個設置不僅控制各個worker節點之間的混洗數據序列化格式,同時還控制RDD存到磁盤上的序列化格式.目前,Kryo不是默認的序列化格式,因為它需要你在使用前注冊需要序列化的類型,不過我們還是建議在對網絡敏感的應用場景下使用Kryo.

Spark對一些常用的Scala核心類型(包括在 Twitter chill 庫的AllScalaRegistrar中)自動使用Kryo序列化格式.

如果你的自定義類型需要使用Kryo序列化,可以用 registerKryoClasses 辦法先注冊：

val conf = new SparkConf.setMaster(...).setAppName(...)

conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))

val sc = new SparkContext(conf)

Kryo的文檔( Kryo documentation )中有詳細描述了更多的高級選項,如：自定義序列化代碼等.

如果你的對象很大,你可能需要增大 spark.kryoserializer.buffer 配置項( config).其值至少需要大于最大對象的序列化長度.

最后,如果你不注冊需要序列化的自定義類型,Kryo也能工作,不過每一個對象實例的序列化結果都會包含一份完整的類名,這有點浪費空間.

內存調優

內存占用調優主要需要考慮3點：1.數據占用的總內存(你多半會希望整個數據集都能裝進內存吧);2.拜訪數據集中每個對象的開銷;3.垃圾回收的開銷(如果你的數據集中對象周轉速度很快的話).

一般,Java對象的拜訪時很快的,但同時Java對象會比原始數據(僅包含各個字段值)占用的空間多2~5倍.主要原因有：

每個Java對象都有一個對象頭(object header),對象頭大約占用16字節,其中包含像其對應class的指針這樣的信息.對于一些包含較少數據的對象(比如只包含一個Int字段),這個對象頭可能比對象數據本身還大.

Java字符串(String)有大約40子節點額外開銷(Java String以Char數據的形式保存原始數據,所以需要一些額外的字段,如數組長度等),并且每個字符都以兩字節的UTF-16編碼在內部保存.因此,10個字符的String很容易就占了60字節.

一些常見的集合類,如 HashMap、LinkedList,使用的是鏈表類數據結構,因此它們對每項數據都有一個包裝器.這些包裝器對象不僅其自身就有“對象頭”,同時還有指向下一個包裝器對象的鏈表指針(通常為8字節).

原始類型的集合通常也是以“裝箱”的形式包裝成對象(如：java.lang.Integer).

本節只是Spark內存管理的一個概要,下面我們會更詳細地討論各種Spark內存調優的具體策略.特別地,我們會討論如何評估數據的內存使用量,以及如何改進 – 要么改變你的數據結構,要么以某種序列化格式存儲數據.最后,我們還會討論如何調整Spark的緩存大小,以及如何調優Java的垃圾回收器.

內存管理概覽

Spark中內存主要用于兩類目的：執行計算和數據存儲.執行計算的內存主要用于混洗(Shuffle)、關聯(join)、排序(sort)以及聚合(aggregation),而數據存儲的內存主要用于緩存和集群內部數據傳播.Spark中執行計算和數據存儲都是共享同一個內存區域(M).如果執行計算沒有占用內存,那么數據存儲可以申請占用所有可用的內存,反之亦然.執行計算可能會搶占數據存儲使用的內存,并將存儲于內存的數據逐出內存,直到數據存儲占用的內存比例降低到一個指定的比例(R).換句話說,R是M基礎上的一個子區域,這個區域的內存數據永遠不會被逐出內存.然而,數據存儲不會搶占執行計算的內存(否則實現太復雜了).

這樣設計主要有這么幾個需要考慮的點.首先,不需要緩存數據的應用可以把整個空間用來執行計算,從而避免頻繁地把數據吐到磁盤上.其次,需要緩存數據的應用能夠有一個數據存儲比例(R)的最低保證,也避免這部分緩存數據被全部逐出內存.最后,這個實現方式能夠在默認情況下,為大多數使用場景提供合理的性能,而不需要專家級用戶來設置內存使用如何劃分.

雖然有兩個內存劃分相關的配置參數,但一般來說,用戶不需要設置,因為默認值已經能夠適用于絕大部分的使用場景：

spark.memory.fraction 表示上面M的大小,其值為相對于JVM堆內存的比例(默認0.75).剩余的25%是為其他用戶數據結構、Spark內部元數據以及避免OOM錯誤的平安預留空間(大量稀疏數據和異常大的數據記錄).

spark.memory.storageFraction 表示上面R的大小,其值為相對于M的一個比例(默認0.5).R是M中專門用于緩存數據塊,且這部分數據塊永遠不會因執行計算任務而逐出內存.

評估內存消耗

確定一個數據集占用內存總量最好的方法就是,創建一個RDD,并緩存到內存中,然后再到web UI上”Storage”頁面查看.頁面上會展示這個RDD總共占用了多少內存.

要評估一個特定對象的內存占用量,可以用 SizeEstimator.estimate 辦法.這個辦法對試驗哪種數據結構能夠裁剪內存占用量比較有用,同時,也可以幫助用戶了解廣播變量在每個執行器堆上占用的內存量.

數據結構調優

減少內存消耗的首要辦法就是避免過多的Java封裝(減少對象頭和額外輔助字段),比如基于指針的數據結構和包裝對象等.以下有幾條建議：

設計數據結構的時候,優先使用對象數組和原生類型,減少對復雜集合類型(如：HashMap)的使用. fastutil 提供了一些很方便的原聲類型集合,同時兼容Java標準庫.

盡可能避免嵌套大量的小對象和指針.

對應鍵值應盡量使用數值型或枚舉型,而不是字符串型.

如果內存小于32GB,可以設置JVM標志參數 -XX:+UseCompressdOops 將指針設為4字節而不是8字節.你可以在 spark-env.sh 中設置這個參數.

序列化RDD存儲

如果經過上面的調整后,存儲的數據對象還是太大,那么你可以試試將這些對象以序列化格式存儲,所需要做的只是通過 RDD persistence API 設置好存儲級別,如：MEMORY_ONLY_SER.Spark會將RDD的每個分區以一個巨大的字節數組形式存儲起來.以序列化格式存儲的唯一缺點就是拜訪數據會變慢一點,因為Spark需要反序列化每個被拜訪的對象.如果你需要序列化緩存數據,我們強烈建議你使用Kryo(using Kryo ),和Java序列化相比,Kryo能大大減少序列化對象占用的空間(當然也比原始Java對象小很多).

垃圾回收調優

JVM的垃圾回收在某些情況下可能會造成瓶頸,比如,你的RDD存儲經常需要“換入換出”(新RDD搶占了老RDD內存,不過如果你的程序沒有這種情況的話那JVM垃圾回收一般不是問題,比如,你的RDD只是載入一次,后續只是在這一個RDD上做操作).當Java需要把老對象逐出內存的時候,JVM需要跟蹤所有的Java對象,并找出那些對象已經沒有用了.概括起來就是,垃圾回收的開銷和對象個數成正比,所以減少對象的個數(比如用 Int數組取代 LinkedList),就能大大減少垃圾回收的開銷.當然,一個更好的辦法就如前面所說的,以序列化形式存儲數據,這時每個RDD分區都只包含有一個對象了(一個巨大的字節數組).在嘗試其他技術方案前,首先可以試試用序列化RDD的方式( serialized caching )評估一下GC是不是一個瓶頸.

如果你的作業中各個任務需要的工作內存和節點上存儲的RDD緩存占用的內存產生沖突,那么GC很可能會出現問題.下面我們將討論一下如何控制好RDD緩存使用的內存空間,以減少這種沖突.

衡量GC的影響

GC調優的第一步是統計一下,垃圾回收啟動的頻率以及GC所使用的總時間.給JVM設置一下這幾個參數(參考Spark配置指南 – configuration guide ,查看Spark作業中的Java選項參數)：-verbose:gc -XX:+PrintGCDetails,就可以在后續Spark作業的worker日志中看到每次GC花費的時間.注意,這些日志是在集群worker節點上(在各節點的工作目錄下stdout文件中),而不是你的驅動器所在節點.

高級GC調優

為了進一步調優GC,我們就需要對JVM內存管理有一個基本的了解：

Java堆內存可分配的空間有兩個區域：新生代(Young generation)和老生代(Old generation).新生代用以保存生存周期短的對象,而老生代則是保存生存周期長的對象.

新生代區域被進一步劃分為三個子區域：Eden,Survivor1,Survivor2.

簡要描述一下垃圾回收的過程：如果Eden區滿了,則啟動一輪minor GC回收Eden中的對象,生存下來(沒有被回收掉)的Eden中的對象和Survivor1區中的對象一并復制到Survivor2中.兩個Survivor區域是互相切換使用的(就是說,下次從Eden和Survivor2中復制到Survivor1中).如果某個對象的年齡(每次GC所有生存下來的對象長一歲)超過某個閾值,或者Survivor2(下次是Survivor1)區域滿了,則將對象移到老生代(Old區).最終如果老生代也滿了,就會啟動full GC.

Spark GC調優的目標就是確保老生代(Old generation )只保存長生命周期RDD,而同時新生代(Young generation )的空間又能足夠保存短生命周期的對象.這樣就能在任務執行期間,避免啟動full GC.以下是GC調優的主要步驟：

從GC的統計日志中觀察GC是否啟動太多.如果某個任務結束前,多次啟動了full GC,則意味著用以執行該任務的內存不夠.

如果GC統計信息中顯示,老生代內存空間已經接近存滿,可以通過降低 spark.memory.storageFraction 來減少RDD緩存占用的內存;減少緩存對象總比任務執行緩慢要強!

如果major GC比較少,但minor GC很多的話,可以多分配一些Eden內存.你可以把Eden的大小設為高于各個任務執行所需的工作內存.如果要把Eden大小設為E,則可以這樣設置新生代區域大小：-Xmn=4/3*E.(放大4/3倍,主要是為了給Survivor區域保留空間)

舉例來說,如果你的任務會從HDFS上讀取數據,那么單個任務的內存需求可以用其所讀取的HDFS數據塊的大小來評估.需要特別注意的是,解壓后的HDFS塊是解壓前的2~3倍大.所以如果我們希望保留3~4個任務并行的工作內存,并且HDFS塊大小為64MB,那么可以評估Eden的大小應該設為 4*3*64MB.

最后,再觀察一下垃圾回收的啟動頻率和總耗時有沒有什么變化.

我們的很多經驗表明,GC調優的效果和你的程序代碼以及可用的總內存相關.網上還有不少調優的選項說明( many more tuning options ),但總體來說,就是控制好full GC的啟動頻率,就能有效減少垃圾回收開銷.

其他注意事項

并行度

一般來說集群并不會滿負荷運轉,除非你吧每個操作的并行度都設得足夠大.Spark會自動根據對應的輸入文件大小來設置“map”類算子的并行度(當然你可以通過一個SparkContext.textFile等函數的可選參數來控制并行度),而對于想 groupByKey 或reduceByKey這類 “reduce” 算子,會使用其各父RDD分區數的最大值.你可以將并行度作為構建RDD第二個參數(參考 spark.PairRDDFunctions ),或者設置 spark.default.parallelism 這個默認值.一般來說,評估并行度的時候,我們建議2~3個任務共享一個CPU.

Reduce任務的內存占用

如果RDD比內存要大,有時候你可能收到一個OutOfMemoryError,但其實這是因為你的任務集中的某個任務太大了,如reduce任務groupByKey.Spark的混洗(Shuffle)算子(sortByKey,groupByKey,reduceByKey,join等)會在每個任務中構建一個哈希表,以便在任務中對數據分組,這個哈希表有時會很大.最簡單的修復方法就是增大并行度,以減小單個任務的輸入集.Spark對于200ms以內的短任務支持非常好,因為Spark可以跨任務復用執行器JVM,任務的啟動開銷很小,因此把并行度增加到比集群中總CPU核數還多是沒有任何問題的.

廣播大變量

使用SparkContext中的廣播變量相關功能( broadcast functionality )能大大減少每個任務本身序列化的大小,以及集群中啟動作業的開銷.如果你的Spark任務正在使用驅動器(driver)程序中定義的巨大對象(比如：靜態查詢表),請考慮使用廣播變量替代之.Spark會在master上將各個任務的序列化后大小打印出來,所以你可以檢查一下各個任務是否過大;通常來說,大于20KB的任務就值得優化一下.

數據本地性

數據本地性對Spark作業往往會有較大的影響.如果代碼和其所操作的數據在統一節點上,那么計算速度肯定會更快一些.但如果二者不在一起,那必然需要挪動其中之一.一般來說,挪動序列化好的代碼肯定比挪動一大堆數據要快.Spark就是基于這個一般性原則來構建數據本地性的調度.

數據本地性是指代碼和其所處理的數據的距離.基于數據當前的位置,數據本地性可以劃分成以下幾個層次(按從近到遠排序)：

PROCESS_LOCAL 數據和運行的代碼處于同一個JVM進程內.

NODE_LOCAL 數據和代碼處于同一節點.例如,數據處于HDFS上某個節點,而對應的執行器(executor)也在同一個機器節點上.這會比PROCESS_LOCAL稍微慢一些,因為數據需要跨進程傳遞.

NO_PREF 數據在任何地方處理都一樣,沒有本地性偏好.

RACK_LOCAL 數據和代碼處于同一個機架上的不同機器.這時,數據和代碼處于不同機器上,需要通過網絡傳遞,但還是在同一個機架上,一般也就通過一個交換機傳輸即可.

ANY 數據在網絡中其他未知,即數據和代碼不在同一個機架上.

Spark傾向于讓所有任務都具有最佳的數據本地性,但這并非總是可行的.某些情況下,可能會出現一些空閑的執行器(executor)沒有待處理的數據,那么Spark可能就會犧牲一些數據本地性.有兩種可能的選項：a)等待已經有任務的CPU,待其釋放后立即在同一臺機器上啟動一個任務;b)立即在其他節點上啟動新任務,并把所需要的數據復制過去.

而通常,Spark會等待一小會,看看是否有CPU會被釋放出來.一旦等待超時,則立即在其他節點上啟動并將所需的數據復制過去.數據本地性各個級別之間的回落超時可以單獨配置,也可以在統一參數內一起設定;詳細請參考 configuration page 中的 spark.locality 相關參數.如果你的任務執行時間比較長并且數據本地性很差,你就應該試試調大這幾個參數,不過默認值一般都能適用于大多數場景了.

總結

本文是一個簡短的Spark調優指南,列舉了Spark應用調優一些比較重要的考慮點 – 最重要的就是,數據序列化和內存調優.對于絕大多數應用來說,用Kryo格式序列化數據能夠辦理大多數的性能問題.如果您有其他關于性能調優最佳實踐的問題,歡迎郵件咨詢( Spark mailing list ).

《《Spark 官方文檔》Spark調優》是否對您有啟發，歡迎查看更多與《《Spark 官方文檔》Spark調優》相關教程，學精學透。維易PHP學院為您提供精彩教程。

轉載請注明本頁網址：
http://www.fzlkiss.com/jiaocheng/7850.html