Spark調(diào)優(yōu)

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

數(shù)據(jù)序列化

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

• Java serialization: 默認情況，Spark使用Java自帶的ObjectOutputStream 框架來序列化對象，這樣任何實現(xiàn)了?java.io.Serializable?接口的對象，都能被序列化。同時，你還可以通過擴展?java.io.Externalizable?來控制序列化性能。Java序列化很靈活但性能較差，同時序列化后占用的字節(jié)數(shù)也較多。
• Kryo serialization: Spark還可以使用Kryo 庫（版本2）提供更高效的序列化格式。Kryo的序列化速度和字節(jié)占用都比Java序列化好很多（通常是10倍左右），但Kryo不支持所有實現(xiàn)了Serializable?接口的類型，它需要你在程序中 register 需要序列化的類型，以得到最佳性能。

要切換到使用 Kryo，你可以在?SparkConf?初始化的時候調(diào)用 conf.set(“spark.serializer”, “org.apache.spark.serializer.KryoSerializer”)。這個設置不僅控制各個worker節(jié)點之間的混洗數(shù)據(jù)序列化格式，同時還控制RDD存到磁盤上的序列化格式。目前，Kryo不是默認的序列化格式，因為它需要你在使用前注冊需要序列化的類型，不過我們還是建議在對網(wǎng)絡敏感的應用場景下使用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也能工作，不過每一個對象實例的序列化結(jié)果都會包含一份完整的類名，這有點浪費空間。

內(nèi)存調(diào)優(yōu)

內(nèi)存占用調(diào)優(yōu)主要需要考慮3點：1.數(shù)據(jù)占用的總內(nèi)存（你多半會希望整個數(shù)據(jù)集都能裝進內(nèi)存吧）；2.訪問數(shù)據(jù)集中每個對象的開銷；3.垃圾回收的開銷（如果你的數(shù)據(jù)集中對象周轉(zhuǎn)速度很快的話）。

一般，Java對象的訪問時很快的，但同時Java對象會比原始數(shù)據(jù)（僅包含各個字段值）占用的空間多2~5倍。主要原因有：

• 每個Java對象都有一個對象頭（object header），對象頭大約占用16字節(jié)，其中包含像其對應class的指針這樣的信息。對于一些包含較少數(shù)據(jù)的對象（比如只包含一個Int字段），這個對象頭可能比對象數(shù)據(jù)本身還大。
• Java字符串（String）有大約40子節(jié)點額外開銷（Java String以Char數(shù)據(jù)的形式保存原始數(shù)據(jù)，所以需要一些額外的字段，如數(shù)組長度等），并且每個字符都以兩字節(jié)的UTF-16編碼在內(nèi)部保存。因此，10個字符的String很容易就占了60字節(jié)。
• 一些常見的集合類，如 HashMap、LinkedList，使用的是鏈表類數(shù)據(jù)結(jié)構(gòu)，因此它們對每項數(shù)據(jù)都有一個包裝器。這些包裝器對象不僅其自身就有“對象頭”，同時還有指向下一個包裝器對象的鏈表指針（通常為8字節(jié)）。
• 原始類型的集合通常也是以“裝箱”的形式包裝成對象（如：java.lang.Integer）。

本節(jié)只是Spark內(nèi)存管理的一個概要，下面我們會更詳細地討論各種Spark內(nèi)存調(diào)優(yōu)的具體策略。特別地，我們會討論如何評估數(shù)據(jù)的內(nèi)存使用量，以及如何改進 – 要么改變你的數(shù)據(jù)結(jié)構(gòu)，要么以某種序列化格式存儲數(shù)據(jù)。最后，我們還會討論如何調(diào)整Spark的緩存大小，以及如何調(diào)優(yōu)Java的垃圾回收器。

內(nèi)存管理概覽

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

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

雖然有兩個內(nèi)存劃分相關(guān)的配置參數(shù)，但一般來說，用戶不需要設置，因為默認值已經(jīng)能夠適用于絕大部分的使用場景：

• spark.memory.fraction 表示上面M的大小，其值為相對于JVM堆內(nèi)存的比例（默認0.75）。剩余的25%是為其他用戶數(shù)據(jù)結(jié)構(gòu)、Spark內(nèi)部元數(shù)據(jù)以及避免OOM錯誤的安全預留空間（大量稀疏數(shù)據(jù)和異常大的數(shù)據(jù)記錄）。
• spark.memory.storageFraction 表示上面R的大小，其值為相對于M的一個比例（默認0.5）。R是M中專門用于緩存數(shù)據(jù)塊，且這部分數(shù)據(jù)塊永遠不會因執(zhí)行計算任務而逐出內(nèi)存。

評估內(nèi)存消耗

確定一個數(shù)據(jù)集占用內(nèi)存總量最好的辦法就是，創(chuàng)建一個RDD，并緩存到內(nèi)存中，然后再到web UI上”Storage”頁面查看。頁面上會展示這個RDD總共占用了多少內(nèi)存。

要評估一個特定對象的內(nèi)存占用量，可以用 SizeEstimator.estimate 方法。這個方法對試驗哪種數(shù)據(jù)結(jié)構(gòu)能夠裁剪內(nèi)存占用量比較有用，同時，也可以幫助用戶了解廣播變量在每個執(zhí)行器堆上占用的內(nèi)存量。

數(shù)據(jù)結(jié)構(gòu)調(diào)優(yōu)

減少內(nèi)存消耗的首要方法就是避免過多的Java封裝（減少對象頭和額外輔助字段），比如基于指針的數(shù)據(jù)結(jié)構(gòu)和包裝對象等。以下有幾條建議：

設計數(shù)據(jù)結(jié)構(gòu)的時候，優(yōu)先使用對象數(shù)組和原生類型，減少對復雜集合類型（如：HashMap）的使用。fastutil?提供了一些很方便的原聲類型集合，同時兼容Java標準庫。

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

對應鍵值應盡量使用數(shù)值型或枚舉型，而不是字符串型。

如果內(nèi)存小于32GB，可以設置JVM標志參數(shù) -XX:+UseCompressdOops 將指針設為4字節(jié)而不是8字節(jié)。你可以在??spark-env.sh?中設置這個參數(shù)。

序列化RDD存儲

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

垃圾回收調(diào)優(yōu)

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

如果你的作業(yè)中各個任務需要的工作內(nèi)存和節(jié)點上存儲的RDD緩存占用的內(nèi)存產(chǎn)生沖突，那么GC很可能會出現(xiàn)問題。下面我們將討論一下如何控制好RDD緩存使用的內(nèi)存空間，以減少這種沖突。

衡量GC的影響

GC調(diào)優(yōu)的第一步是統(tǒng)計一下，垃圾回收啟動的頻率以及GC所使用的總時間。給JVM設置一下這幾個參數(shù)（參考Spark配置指南 –??configuration guide，查看Spark作業(yè)中的Java選項參數(shù)）：-verbose:gc -XX:+PrintGCDetails，就可以在后續(xù)Spark作業(yè)的worker日志中看到每次GC花費的時間。注意，這些日志是在集群worker節(jié)點上（在各節(jié)點的工作目錄下stdout文件中），而不是你的驅(qū)動器所在節(jié)點。

高級GC調(diào)優(yōu)

為了進一步調(diào)優(yōu)GC，我們就需要對JVM內(nèi)存管理有一個基本的了解：

• Java堆內(nèi)存可分配的空間有兩個區(qū)域：新生代（Young?generation）和老生代（Old?generation）。新生代用以保存生存周期短的對象，而老生代則是保存生存周期長的對象。
• 新生代區(qū)域被進一步劃分為三個子區(qū)域：Eden，Survivor1，Survivor2。
• 簡要描述一下垃圾回收的過程：如果Eden區(qū)滿了，則啟動一輪minor GC回收Eden中的對象，生存下來（沒有被回收掉）的Eden中的對象和Survivor1區(qū)中的對象一并復制到Survivor2中。兩個Survivor區(qū)域是互相切換使用的（就是說，下次從Eden和Survivor2中復制到Survivor1中）。如果某個對象的年齡（每次GC所有生存下來的對象長一歲）超過某個閾值，或者Survivor2（下次是Survivor1）區(qū)域滿了，則將對象移到老生代（Old區(qū)）。最終如果老生代也滿了，就會啟動full GC。

Spark GC調(diào)優(yōu)的目標就是確保老生代（Old?generation?）只保存長生命周期RDD，而同時新生代（Young?generation?）的空間又能足夠保存短生命周期的對象。這樣就能在任務執(zhí)行期間，避免啟動full GC。以下是GC調(diào)優(yōu)的主要步驟：

• 從GC的統(tǒng)計日志中觀察GC是否啟動太多。如果某個任務結(jié)束前，多次啟動了full GC，則意味著用以執(zhí)行該任務的內(nèi)存不夠。
• 如果GC統(tǒng)計信息中顯示，老生代內(nèi)存空間已經(jīng)接近存滿，可以通過降低 spark.memory.storageFraction 來減少RDD緩存占用的內(nèi)存；減少緩存對象總比任務執(zhí)行緩慢要強！
• 如果major GC比較少，但minor GC很多的話，可以多分配一些Eden內(nèi)存。你可以把Eden的大小設為高于各個任務執(zhí)行所需的工作內(nèi)存。如果要把Eden大小設為E，則可以這樣設置新生代區(qū)域大小：-Xmn=4/3*E。（放大4/3倍，主要是為了給Survivor區(qū)域保留空間）
• 舉例來說，如果你的任務會從HDFS上讀取數(shù)據(jù)，那么單個任務的內(nèi)存需求可以用其所讀取的HDFS數(shù)據(jù)塊的大小來評估。需要特別注意的是，解壓后的HDFS塊是解壓前的2~3倍大。所以如果我們希望保留3~4個任務并行的工作內(nèi)存，并且HDFS塊大小為64MB，那么可以評估Eden的大小應該設為 4*3*64MB。
• 最后，再觀察一下垃圾回收的啟動頻率和總耗時有沒有什么變化。

我們的很多經(jīng)驗表明，GC調(diào)優(yōu)的效果和你的程序代碼以及可用的總內(nèi)存相關(guān)。網(wǎng)上還有不少調(diào)優(yōu)的選項說明（many more tuning options），但總體來說，就是控制好full GC的啟動頻率，就能有效減少垃圾回收開銷。

其他注意事項

并行度

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

Reduce任務的內(nèi)存占用

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

廣播大變量

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

數(shù)據(jù)本地性

數(shù)據(jù)本地性對Spark作業(yè)往往會有較大的影響。如果代碼和其所操作的數(shù)據(jù)在統(tǒng)一節(jié)點上，那么計算速度肯定會更快一些。但如果二者不在一起，那必然需要挪動其中之一。一般來說，挪動序列化好的代碼肯定比挪動一大堆數(shù)據(jù)要快。Spark就是基于這個一般性原則來構(gòu)建數(shù)據(jù)本地性的調(diào)度。

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

• PROCESS_LOCAL 數(shù)據(jù)和運行的代碼處于同一個JVM進程內(nèi)。
• NODE_LOCAL 數(shù)據(jù)和代碼處于同一節(jié)點。例如，數(shù)據(jù)處于HDFS上某個節(jié)點，而對應的執(zhí)行器（executor）也在同一個機器節(jié)點上。這會比PROCESS_LOCAL稍微慢一些，因為數(shù)據(jù)需要跨進程傳遞。
• NO_PREF 數(shù)據(jù)在任何地方處理都一樣，沒有本地性偏好。
• RACK_LOCAL 數(shù)據(jù)和代碼處于同一個機架上的不同機器。這時，數(shù)據(jù)和代碼處于不同機器上，需要通過網(wǎng)絡傳遞，但還是在同一個機架上，一般也就通過一個交換機傳輸即可。
• ANY 數(shù)據(jù)在網(wǎng)絡中其他未知，即數(shù)據(jù)和代碼不在同一個機架上。

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

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

總結(jié)

本文是一個簡短的Spark調(diào)優(yōu)指南，列舉了Spark應用調(diào)優(yōu)一些比較重要的考慮點 – 最重要的就是，數(shù)據(jù)序列化和內(nèi)存調(diào)優(yōu)。對于絕大多數(shù)應用來說，用Kryo格式序列化數(shù)據(jù)能夠解決大多數(shù)的性能問題。如果您有其他關(guān)于性能調(diào)優(yōu)最佳實踐的問題，歡迎郵件咨詢（Spark mailing list?）。

?

該文轉(zhuǎn)自?http://ifeve.com/spark-tuning/

官方英文地址?http://spark.apache.org/docs/latest/tuning.html

本文轉(zhuǎn)自shishanyuan博客園博客，原文鏈接： ??http://www.cnblogs.com/shishanyuan/p/8481854.html?，如需轉(zhuǎn)載請自行聯(lián)系原作者

總結(jié)

以上是生活随笔為你收集整理的Spark官方调优文档翻译（转载）的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。