01 sparkAPI-閱讀總結-TuningSpark

TuningSpark

調整Spark

• 數據序列化
• 內存調整
  • 內存管理概述
  • 肯定內存消耗
  • 調整數據結構
  • 序列化RDD存儲
  • 垃圾收集調整
• 其餘考慮因素
  • 並行程度
  • 減小任務的內存使用狀況
  • 廣播大變量
  • 數據位置
• 摘要

因爲大多數Spark計算的內存特性，Spark程序可能會受到羣集中任何資源的瓶頸：CPU，網絡帶寬或內存。大多數狀況下，若是數據適合內存，瓶頸就是網絡帶寬，但有時候，您還須要進行一些調整，例如 以序列化形式存儲RDD，以減小內存使用。本指南將介紹兩個主要主題：數據序列化，這對於良好的網絡性能相當重要，還能夠減小內存使用和內存調整。咱們還草擬了幾個較小的主題。

數據序列化

序列化在任何分佈式應用程序的性能中起着重要做用。將對象序列化或消耗大量字節的速度慢的格式將大大減慢計算速度。一般，這將是您應該優化Spark應用程序的第一件事。Spark旨在在便利性（容許您使用操做中的任何Java類型）和性能之間取得平衡。它提供了兩個序列化庫：

• Java序列化：默認狀況下，Spark使用Java ObjectOutputStream框架序列化對象，而且能夠與您建立的任何實現的類一塊兒使用java.io.Serializable。您還能夠經過擴展來更緊密地控制序列化的性能 java.io.Externalizable。Java序列化是靈活的，但一般很慢，並致使許多類的大型序列化格式。
• Kryo序列化：Spark還可使用Kryo庫（版本4）更快地序列化對象。Kryo比Java序列化（一般高達10倍）明顯更快，更緊湊，但不支持全部Serializable類型，而且須要您提早註冊您將在程序中使用的類以得到最佳性能。

您能夠經過使用SparkConf初始化做業 並調用來切換到使用Kryo conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")。此設置配置序列化程序，不只用於在工做節點之間混洗數據，還用於將RDD序列化到磁盤。Kryo不是默認值的惟一緣由是由於自定義註冊要求，但咱們建議在任何網絡密集型應用程序中嘗試它。從Spark 2.0.0開始，咱們在使用簡單類型，簡單類型數組或字符串類型對RDD進行混洗時，內部使用Kryo序列化程序。

Spark自動包含Kryo序列化程序，用於來自Twitter chill庫的AllScalaRegistrar中涵蓋的許多經常使用核心Scala類。

要使用Kryo註冊本身的自定義類，請使用該registerKryoClasses方法。

val conf = new SparkConf().setMaster(...).setAppName(...)
conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))
val sc = new SparkContext(conf)

所述KRYO文檔描述了更先進的註冊選項，如添加自定義序列的代碼。

若是對象很大，則可能還須要增長spark.kryoserializer.buffer 配置。此值必須足夠大才能容納要序列化的最大對象。

最後，若是你沒有註冊你的自定義類，Kryo仍然會工做，但它必須存儲每一個對象的完整類名，這是浪費。

內存調整

有三個方面的考慮在調整內存使用：該量的存儲你的對象所使用的（你可能但願你的整個數據集，以適應在內存中），則成本訪問這些對象，而且開銷垃圾收集（若是你有高成交對象的條款）。

默認狀況下，Java對象訪問速度很快，但與其字段中的「原始」數據相比，能夠輕鬆佔用2-5倍的空間。這是因爲如下幾個緣由：

• 每一個不一樣的Java對象都有一個「對象頭」，大約16個字節，幷包含諸如指向其類的指針之類的信息。對於包含很是少數據的對象（好比一個Int字段），這可能比數據大。
• Java String在原始字符串數據上有大約40字節的開銷（由於它們將它存儲在Chars 的數組中並保留額外的數據，例如長度），而且因爲UTF-16的內部使用而將每一個字符存儲爲兩個字節String編碼。所以，10個字符的字符串很容易消耗60個字節。
• 公共集合類，例如HashMap和LinkedList，使用連接數據結構，其中每一個條目都有一個「包裝」對象（例如Map.Entry）。此對象不只具備標題，還具備指向列表中下一個對象的指針（一般爲8個字節）。
• 原始類型的集合一般將它們存儲爲「盒裝」對象，例如java.lang.Integer。

本節將首先概述Spark中的內存管理，而後討論用戶能夠採起的具體策略，以便在他/她的應用程序中更有效地使用內存。特別是，咱們將描述如何肯定對象的內存使用狀況，以及如何經過更改數據結構或以序列化格式存儲數據來改進它。而後咱們將介紹調整Spark的緩存大小和Java垃圾收集器。

內存管理概述

Spark中的內存使用大體屬於如下兩種類別之一：執行和存儲。執行內存是指用於在隨機，鏈接，排序和聚合中進行計算的內存，而存儲內存是指用於在集羣中緩存和傳播內部數據的內存。在Spark中，執行和存儲共享一個統一的區域（M）。當沒有使用執行內存時，存儲能夠獲取全部可用內存，反之亦然。若有必要，執行能夠驅逐存儲，但僅限於總存儲內存使用量低於某個閾值（R）。換句話說，R描述了M高速緩存塊從未被驅逐的子區域。因爲實施的複雜性，存儲可能不會驅逐執行。

該設計確保了幾種理想的特性。首先，不使用緩存的應用程序可使用整個空間執行，從而避免沒必要要的磁盤溢出。其次，使用緩存的應用程序能夠保留最小的存儲空間（R），其中數據塊不受驅逐。最後，這種方法爲各類工做負載提供了合理的開箱即用性能，而無需用戶內部劃份內存的專業知識。

雖然有兩種相關配置，但典型用戶不須要調整它們，由於默認值適用於大多數工做負載：

• spark.memory.fraction表示大小M爲（JVM堆空間 - 300MB）的一小部分（默認值爲0.6）。其他的空間（40％）保留用於用戶數據結構，Spark中的內部元數據，以及在稀疏和異常大的記錄的狀況下防止OOM錯誤。
• spark.memory.storageFraction將大小表示R爲M（默認值0.5）的一小部分。 R是M緩存塊不受執行驅逐的存儲空間。

spark.memory.fraction應該設置值，以便在JVM的舊版或「終身」代中溫馨地適應這個堆空間量。有關詳細信息，請參閱下面的高級GC調整討論。

肯定內存消耗

肯定數據集所需內存消耗量的最佳方法是建立RDD，將其放入緩存中，而後查看Web UI中的「存儲」頁面。該頁面將告訴您RDD佔用多少內存。

爲了估計特定對象的內存消耗，使用SizeEstimator的estimate方法。這對於嘗試使用不一樣的數據佈局來調整內存使用狀況以及肯定廣播變量在每一個執行程序堆上佔用的空間量很是有用。

調整數據結構

減小內存消耗的第一種方法是避免增長開銷的Java功能，例如基於指針的數據結構和包裝器對象。作這件事有不少種方法：

1. 設計您的數據結構以優先選擇對象數組和基本類型，而不是標準的Java或Scala集合類（例如HashMap）。該fastutil 庫提供方便的集合類基本類型是與Java標準庫兼容。
2. 儘量避免使用包含大量小對象和指針的嵌套結構。
3. 考慮使用數字ID或枚舉對象而不是鍵的字符串。
4. 若是RAM少於32 GB，請設置JVM標誌-XX:+UseCompressedOops以使指針爲四個字節而不是八個字節。您能夠添加這些選項 spark-env.sh。

序列化RDD存儲

儘管進行了這種調整，可是當對象仍然太大而沒法有效存儲時，減小內存使用的一種更簡單的方法是使用RDD持久性API中的序列化StorageLevels以序列化形式存儲它們，例如。而後，Spark將每一個RDD分區存儲爲一個大字節數組。因爲必須動態地反序列化每一個對象，所以以序列化形式存儲數據的惟一缺點是訪問時間較慢。若是您但願以序列化形式緩存數據，咱們強烈建議使用Kryo，由於它致使比Java序列化（固然比原始Java對象）小得多的尺寸。MEMORY_ONLY_SER

垃圾收集調整

當您根據程序存儲的RDD進行大量「流失」時，JVM垃圾回收可能會出現問題。（在讀取RDD一次而後在其上運行許多操做的程序中一般不會出現問題。）當Java須要逐出舊對象以便爲新對象騰出空間時，它須要遍歷全部Java對象並查找未使用的。這裏要記住的要點是垃圾收集的成本與Java對象的數量成正比，所以使用具備較少對象的數據結構（例如，Ints而不是a 的數組LinkedList）大大下降了這種成本。更好的方法是以序列化形式保存對象，如上所述：如今只有一個每一個RDD分區的對象（一個字節數組）。在嘗試其餘技術以前，首先要嘗試GC是一個問題是使用序列化緩存。

因爲任務的工做內存（運行任務所需的空間量）與節點上緩存的RDD之間的干擾，GC也多是一個問題。咱們將討論如何控制分配給RDD緩存的空間以緩解這種狀況。

測量GC的影響

GC調優的第一步是收集有關垃圾收集發生頻率和GC使用時間的統計信息。這能夠經過添加-verbose:gc -XX:+PrintGCDetails -XX:+PrintGCTimeStampsJava選項來完成。（有關將Java選項傳遞給Spark做業的信息，請參閱配置指南。）下次運行Spark做業時，每次發生垃圾收集時，您都會看到工做日誌中打印的消息。請注意，這些日誌將位於羣集的工做節點上（stdout位於其工做目錄中的文件中），而不是位於驅動程序上。

高級GC調整

爲了進一步調整垃圾收集，咱們首先須要瞭解JVM中有關內存管理的一些基本信息：

• Java堆空間分爲Young和Old兩個區域。Young表明意味着持有短命的物體，而老一代則用於生命週期較長的物體。

• Young代進一步分爲三個區域[Eden，Survivor1，Survivor2]。

• 垃圾收集過程的簡化描述：當Eden已滿時，在Eden上運行次要GC，並將從Eden和Survivor1中存活的對象複製到Survivor2。倖存者地區被交換。若是對象足夠大或Survivor2已滿，則將其移至Old。最後，當Old接近完整時，將調用完整的GC。

Spark中GC調整的目標是確保只有長壽命的RDD存儲在Old代中，而且Young代的大小足以存儲短時間對象。這將有助於避免完整的GC收集在任務執行期間建立的臨時對象。可能有用的一些步驟是：

• 經過收集GC統計數據來檢查是否有太多垃圾收集。若是在任務完成以前屢次調用完整的GC，則意味着沒有足夠的內存可用於執行任務。

• 若是有太屢次要集合但沒有不少主要的GC，那麼爲Eden分配更多內存將會有所幫助。您能夠將Eden的大小設置爲高估每一個任務所需的內存量。若是肯定Eden的大小E，則可使用該選項設置Young代的大小-Xmn=4/3*E。（按比例增長4/3也是爲了解釋倖存者地區使用的空間。）

• 在打印的GC統計信息中，若是OldGen接近滿，則經過下降來減小用於緩存的內存量spark.memory.fraction; 緩存更少的對象比減慢任務執行速度更好。或者，考慮減少Young代的尺寸。-Xmn若是您按上述設置，這意味着下降。若是沒有，請嘗試更改JVM NewRatio參數的值。許多JVM將此默認爲2，這意味着舊一代佔據堆的2/3。它應該足夠大，使得該分數超過spark.memory.fraction。

• 嘗試使用G1GC垃圾收集器-XX:+UseG1GC。在垃圾收集成爲瓶頸的某些狀況下，它能夠提升性能。請注意，大執行人堆大小，可能重要的是增長了G1區域大小 與-XX:G1HeapRegionSize

• 例如，若是您的任務是從HDFS讀取數據，則可使用從HDFS讀取的數據塊的大小來估計任務使用的內存量。請注意，解壓縮塊的大小一般是塊大小的2或3倍。所以，若是咱們但願有3或4個任務的工做空間，而且HDFS塊大小爲128 MB，咱們能夠估計Eden的大小4*3*128MB。

• 監視垃圾收集所用頻率和時間如何隨新設置而變化。

咱們的經驗代表，GC調整的效果取決於您的應用程序和可用內存量。有更多的微調選項在線描述，但在較高的水平，管理完整的GC如何常常發生能夠減小開銷幫助。

能夠經過設置spark.executor.extraJavaOptions做業的配置來指定執行程序的GC調整標誌。

其餘考慮因素

並行程度

除非您爲每一個操做設置足夠高的並行度，不然將沒法充分利用羣集。Spark根據其大小自動設置要在每一個文件上運行的「map」任務的數量（儘管能夠經過可選參數來控制它SparkContext.textFile等），而且對於分佈式「reduce」操做，例如groupByKey和reduceByKey，它使用最大的父級RDD的分區數量。您能夠將並行級別做爲第二個參數傳遞（請參閱spark.PairRDDFunctions文檔），或者設置config屬性spark.default.parallelism以更改默認值。一般，咱們建議羣集中每一個CPU核心有2-3個任務。

減小任務的內存使用狀況

有時候，你會獲得一個OutOfMemoryError，不是由於你的RDD不適合內存，而是由於你的一個任務的工做集，好比其中一個reduce任務groupByKey，太大了。斯巴克的整理操做（sortByKey，groupByKey，reduceByKey，join，等）創建每一個任務中的哈希表來進行分組，而這每每是大的。這裏最簡單的解決方法是 增長並行度，以便每一個任務的輸入集更小。Spark能夠有效地支持短至200毫秒的任務，由於它能夠在多個任務中重用一個執行程序JVM，而且它具備較低的任務啓動成本，所以您能夠安全地將並行度提升到超過羣集中的核心數。

廣播大變量

使用 可用的廣播功能SparkContext能夠大大減小每一個序列化任務的大小，以及經過羣集啓動做業的成本。若是您的任務使用其中的驅動程序中的任何大對象（例如靜態查找表），請考慮將其轉換爲廣播變量。Spark打印主服務器上每一個任務的序列化大小，所以您能夠查看它以肯定您的任務是否過大; 通常來講，大於約20 KB的任務可能值得優化。

數據位置

數據位置可能會對Spark做業的性能產生重大影響。若是數據和在其上運行的代碼在一塊兒，那麼計算每每很快。可是若是代碼和數據是分開的，那麼必須移動到另外一個。一般，將序列化代碼從一個地方運送到另外一個地方比一塊數據更快，由於代碼大小比數據小得多。Spark圍繞數據局部性的通常原則構建其調度。

數據位置是數據與處理它的代碼的接近程度。根據數據的當前位置，有多個級別的位置。從最近到最遠的順序：

• PROCESS_LOCAL數據與正在運行的代碼位於同一JVM中。這是最好的地方
• NODE_LOCAL數據在同一節點上。示例可能位於同一節點上的HDFS中，也可能位於同一節點上的另外一個執行程序中。這比PROCESS_LOCAL由於數據必須在進程之間傳輸要慢一些
• NO_PREF 能夠從任何地方快速訪問數據，而且沒有位置偏好
• RACK_LOCAL數據位於同一機架的服務器上。數據位於同一機架上的不一樣服務器上，所以須要經過網絡發送，一般經過單個交換機
• ANY 數據在網絡上的其餘位置，而不在同一個機架中

Spark喜歡將全部任務安排在最好的局部性級別，但這不老是可能的。在任何空閒執行器上沒有未處理數據的狀況下，Spark切換到較低的局域級別。有兩種選擇:a)等待繁忙的CPU釋放出來，在同一服務器上的數據上啓動任務;b)當即在須要將數據移動到那裏的較遠的地方啓動新任務。.

Spark一般會作的是等待繁忙的CPU釋放的但願。一旦超時到期，它就開始將數據從遠處移動到空閒CPU。每一個級別之間的回退等待超時能夠單獨配置，也能夠在一個參數中一塊兒配置; 有關詳細信息，請參閱配置頁面spark.locality上的 參數。若是您的任務很長而且看不到位置，則應該增長這些設置，但默認狀況一般頗有效。

摘要

這是一個簡短的指南，指出在調優Spark應用程序時應該瞭解的主要問題 - 最重要的是，數據序列化和內存調整。對於大多數程序，切換到Kryo序列化並以序列化形式保存數據將解決最多見的性能問題。請隨時在 Spark郵件列表中詢問有關其餘調優最佳作法的信息。