新世紀(jì)以來，互聯(lián)網(wǎng)及個(gè)人終端的普及，傳統(tǒng)行業(yè)的信息化及物聯(lián)網(wǎng)的發(fā)展等產(chǎn)業(yè)變化產(chǎn)生了大量的數(shù)據(jù)，遠(yuǎn)遠(yuǎn)超出了單臺(tái)機(jī)器能夠處理的范圍，分布式存儲(chǔ)與處理成為唯一的選項(xiàng)。從2005年開始，Hadoop從最初Nutch項(xiàng)目的一部分，逐步發(fā)展成為目前最流行的大數(shù)據(jù)處理平臺(tái)。Hadoop生態(tài)圈的各個(gè)項(xiàng)目，圍繞著大數(shù)據(jù)的存儲(chǔ)，計(jì)算，分析，展示，安全等各個(gè)方面，構(gòu)建了一個(gè)完整的大數(shù)據(jù)生態(tài)系統(tǒng)，并有Cloudera，HortonWorks，MapR等數(shù)十家公司基于開源的Hadoop平臺(tái)構(gòu)建自己的商業(yè)模式，可以認(rèn)為是最近十年來最成功的開源社區(qū)。

Hadoop的成功固然是由于其順應(yīng)了新世紀(jì)以來互聯(lián)網(wǎng)技術(shù)的發(fā)展趨勢(shì)，同時(shí)其基于JVM的平臺(tái)開發(fā)也為Hadoop的快速發(fā)展起到了促進(jìn)作用。Hadoop生態(tài)圈的項(xiàng)目大都基于Java，Scala，Clojure等JVM語言開發(fā)，這些語言良好的語法規(guī)范，豐富的第三方類庫以及完善的工具支持，為Hadoop這樣的超大型項(xiàng)目提供了基礎(chǔ)支撐。同時(shí)，作為在程序員中普及率最高的語言之一，它也降低了更多程序員使用，或是參與開發(fā)Hadoop項(xiàng)目的門檻。同時(shí)，基于Scala開發(fā)的Spark，甚至因?yàn)轫?xiàng)目的火熱反過來極大的促進(jìn)了Scala語言的推廣。但是隨著Hadoop平臺(tái)的逐步發(fā)展，Hadoop生態(tài)圈的項(xiàng)目之間的競爭加劇，越來越多的Hadoop項(xiàng)目注意到了這些JVM語言的一些不足之處，希望通過更有效率的處理方式，提升分布式系統(tǒng)的執(zhí)行效率與健壯性。本文主要以Spark和Flink項(xiàng)目為例，介紹Hadoop社區(qū)觀察到的一些因?yàn)镴VM語言的不足導(dǎo)致的問題，以及相應(yīng)的解決方案與未來可能的發(fā)展方向。

注：本文假設(shè)讀者對(duì)Java和Hadoop系統(tǒng)有基本了解。

背景

目前Hadoop生態(tài)圈共有MapReduce，Tez，Spark及Flink等分布式計(jì)算引擎，分布式計(jì)算引擎項(xiàng)目之間的競爭也相當(dāng)激烈。MapReduce作為Hadoop平臺(tái)的第一個(gè)分布式計(jì)算引擎，具有非常良好的可擴(kuò)展性，Yahoo曾成功的搭建了上萬臺(tái)節(jié)點(diǎn)的MapReduce系統(tǒng)。但是MapReduce只支持Map和Reduce編程范式，使得復(fù)雜數(shù)據(jù)計(jì)算邏輯需要分割為多個(gè)Hadoop Job，而每個(gè)Hadoop Job都需要從HDFS讀取數(shù)據(jù)，并將Job執(zhí)行結(jié)果寫回HDFS，所以會(huì)產(chǎn)生大量額外的IO開銷，目前MapReduce正在逐漸被其他三個(gè)分布式計(jì)算引擎替代。Tez,Spark和Flink都支持圖結(jié)構(gòu)的分布式計(jì)算流，可在同一Job內(nèi)支持任意復(fù)雜邏輯的計(jì)算流。Tez的抽象層次較低，用戶不易直接使用，Spark與Flink都提供了抽象的分布式數(shù)據(jù)集以及可在數(shù)據(jù)集上使用的操作符，用戶可以像操作Scala數(shù)據(jù)集合類似的方式在Spark/FLink中的操作分布式數(shù)據(jù)集，非常的容易上手，同時(shí)，Spark與Flink都在分布式計(jì)算引擎之上，提供了針對(duì)SQL，流處理，機(jī)器學(xué)習(xí)和圖計(jì)算等特定數(shù)據(jù)處理領(lǐng)域的庫。

隨著各個(gè)項(xiàng)目的發(fā)展與日益成熟，通過改進(jìn)分布式計(jì)算框架本身大幅提高性能的機(jī)會(huì)越來越少。同時(shí)，在當(dāng)前數(shù)據(jù)中心的硬件配置中，采用了越來越多更先進(jìn)的IO設(shè)備，例如SSD存儲(chǔ)，10G甚至是40Gbps網(wǎng)絡(luò)，IO帶寬的提升非常明顯，許多計(jì)算密集類型的工作負(fù)載的瓶頸已經(jīng)取決于底層硬件系統(tǒng)的吞吐量，而不是傳統(tǒng)上人們認(rèn)為的IO帶寬，而CPU和內(nèi)存的利用效率，則很大程度上決定了底層硬件系統(tǒng)的吞吐量。所以越來越多的項(xiàng)目將眼光投向了JVM本身，希望通過解決JVM本身帶來的一些問題，提高分布式系統(tǒng)的性能或是健壯性，從而增強(qiáng)自身的競爭力。

JVM本身作為一個(gè)各種類型應(yīng)用執(zhí)行的平臺(tái)，其對(duì)Java對(duì)象的管理也是基于通用的處理策略，其垃圾回收器通過估算Java對(duì)象的生命周期對(duì)Java對(duì)象進(jìn)行有效率的管理。針對(duì)不同類型的應(yīng)用，用戶可能需要針對(duì)該類型應(yīng)用的特點(diǎn)，配置針對(duì)性的JVM參數(shù)更有效率的管理Java對(duì)象，從而提高性能。這種JVM調(diào)優(yōu)的黑魔法需要用戶對(duì)應(yīng)用本身以及JVM的各參數(shù)有深入的了解，極大的提高了分布式計(jì)算平臺(tái)的調(diào)優(yōu)門檻(例如這篇文章中對(duì)Spark的調(diào)優(yōu)Tuning Java Garbage Collection for Spark Applications)。然而類似Spark或是Flink的分布式計(jì)算框架，框架本身了解計(jì)算邏輯每個(gè)步驟的數(shù)據(jù)傳輸，相比于JVM垃圾回收器，其了解更多的Java對(duì)象生命周期，從而為更有效率的管理Java對(duì)象提供了可能。

JVM存在的問題

1. Java對(duì)象開銷

相對(duì)于c/c++等更加接近底層的語言，Java對(duì)象的存儲(chǔ)密度相對(duì)偏低，例如【1】，“abcd”這樣簡單的字符串在UTF-8編碼中需要4個(gè)字節(jié)存儲(chǔ)，但Java采用UTF-16編碼存儲(chǔ)字符串，需要8個(gè)字節(jié)存儲(chǔ)“abcd”，同時(shí)Java對(duì)象還對(duì)象header等其他額外信息，一個(gè)4字節(jié)字符串對(duì)象，在Java中需要48字節(jié)的空間來存儲(chǔ)。對(duì)于大部分的大數(shù)據(jù)應(yīng)用，內(nèi)存都是稀缺資源，更有效率的內(nèi)存存儲(chǔ)，則意味著CPU數(shù)據(jù)訪問吞吐量更高，以及更少的磁盤落地可能。

2. 對(duì)象存儲(chǔ)結(jié)構(gòu)引發(fā)的cache miss

為了緩解CPU處理速度與內(nèi)存訪問速度的差距【2】，現(xiàn)代CPU數(shù)據(jù)訪問一般都會(huì)有多級(jí)緩存。當(dāng)從內(nèi)存加載數(shù)據(jù)到緩存時(shí)，一般是以cache line為單位加載數(shù)據(jù)，所以當(dāng)CPU訪問的數(shù)據(jù)如果是在內(nèi)存中連續(xù)存儲(chǔ)的話，訪問的效率會(huì)非常高。如果CPU要訪問的數(shù)據(jù)不在當(dāng)前緩存所有的cache line中，則需要從內(nèi)存中加載對(duì)應(yīng)的數(shù)據(jù)，這被稱為一次cache miss。當(dāng)cache miss非常高的時(shí)候，CPU大部分的時(shí)間都在等待數(shù)據(jù)加載，而不是真正的處理數(shù)據(jù)。Java對(duì)象并不是連續(xù)的存儲(chǔ)在內(nèi)存上，同時(shí)很多的Java數(shù)據(jù)結(jié)構(gòu)的數(shù)據(jù)聚集性也不好，在Spark的性能調(diào)優(yōu)中，經(jīng)常能夠觀測(cè)到大量的cache miss。Java社區(qū)有個(gè)項(xiàng)目叫做Project Valhalla，可能會(huì)部分的解決這個(gè)問題，有興趣的可以看看這兒OpenJDK: Valhalla。

3. 大數(shù)據(jù)的垃圾回收

Java的垃圾回收機(jī)制，一直讓Java開發(fā)者又愛又恨，一方面它免去了開發(fā)者自己回收資源的步驟，提高了開發(fā)效率，減少了內(nèi)存泄漏的可能，另一方面，垃圾回收也是Java應(yīng)用的一顆不定時(shí)炸彈，有時(shí)秒級(jí)甚至是分鐘級(jí)的垃圾回收極大的影響了Java應(yīng)用的性能和可用性。在當(dāng)前的數(shù)據(jù)中心中，大容量的內(nèi)存得到了廣泛的應(yīng)用，甚至出現(xiàn)了單臺(tái)機(jī)器配置TB內(nèi)存的情況，同時(shí)，大數(shù)據(jù)分析通常會(huì)遍歷整個(gè)源數(shù)據(jù)集，對(duì)數(shù)據(jù)進(jìn)行轉(zhuǎn)換，清洗，處理等步驟。在這個(gè)過程中，會(huì)產(chǎn)生海量的Java對(duì)象，JVM的垃圾回收?qǐng)?zhí)行效率對(duì)性能有很大影響。通過JVM參數(shù)調(diào)優(yōu)提高垃圾回收效率需要用戶對(duì)應(yīng)用和分布式計(jì)算框架以及JVM的各參數(shù)有深入的了解，而且有時(shí)候這也遠(yuǎn)遠(yuǎn)不夠。

4. OOM問題

OutOfMemoryError是分布式計(jì)算框架經(jīng)常會(huì)遇到的問題，當(dāng)JVM中所有對(duì)象大小超過分配給JVM的內(nèi)存大小時(shí)，就會(huì)fOutOfMemoryError錯(cuò)誤，JVM崩潰，分布式框架的健壯性和性能都會(huì)受到影響。通過JVM管理內(nèi)存，同時(shí)試圖解決OOM問題的應(yīng)用，通常都需要檢查Java對(duì)象的大小，并在某些存儲(chǔ)Java對(duì)象特別多的數(shù)據(jù)結(jié)構(gòu)中設(shè)置閾值進(jìn)行控制。但是JVM并沒有提供官方的檢查Java對(duì)象大小的工具，第三方的工具類庫可能無法準(zhǔn)確通用的確定Java對(duì)象的大小【6】。侵入式的閾值檢查也會(huì)為分布式計(jì)算框架的實(shí)現(xiàn)增加很多額外的業(yè)務(wù)邏輯無關(guān)的代碼。

解決方案

為了解決以上提到的問題，高性能分布式計(jì)算框架通常需要以下技術(shù)：

1. 定制的序列化工具。顯式內(nèi)存管理的前提步驟就是序列化，將Java對(duì)象序列化成二進(jìn)制數(shù)據(jù)存儲(chǔ)在內(nèi)存上(on heap或是off-heap)。通用的序列化框架，如Java默認(rèn)的java.io.Serializable將Java對(duì)象以及其成員變量的所有元信息作為其序列化數(shù)據(jù)的一部分，序列化后的數(shù)據(jù)包含了所有反序列化所需的信息。這在某些場景中十分必要，但是對(duì)于Spark或是Flink這樣的分布式計(jì)算框架來說，這些元數(shù)據(jù)信息可能是冗余數(shù)據(jù)。定制的序列化框架，如Hadoop的org.apache.hadoop.io.Writable，需要用戶實(shí)現(xiàn)該接口，并自定義類的序列化和反序列化方法。這種方式效率最高，但需要用戶額外的工作，不夠友好。

2. 顯式的內(nèi)存管理。一般通用的做法是批量申請(qǐng)和釋放內(nèi)存，每個(gè)JVM實(shí)例有一個(gè)統(tǒng)一的內(nèi)存管理器，所有的內(nèi)存的申請(qǐng)和釋放都通過該內(nèi)存管理器進(jìn)行。這可以避免常見的內(nèi)存碎片問題，同時(shí)由于數(shù)據(jù)以二進(jìn)制的方式存儲(chǔ)，可以大大減輕垃圾回收的壓力。

3. 緩存友好的數(shù)據(jù)結(jié)構(gòu)和算法。只將操作相關(guān)的數(shù)據(jù)連續(xù)存儲(chǔ)，可以最大化的利用L1/L2/L3緩存，減少Cache miss的概率，提升CPU計(jì)算的吞吐量。以排序?yàn)槔捎谂判虻闹饕僮魇菍?duì)Key進(jìn)行對(duì)比，如果將所有排序數(shù)據(jù)的Key與Value分開，對(duì)Key連續(xù)存儲(chǔ)，則訪問Key時(shí)的Cache命中率會(huì)大大提高。

定制的序列化工具

分布式計(jì)算框架可以使用定制序列化工具的前提是要處理的數(shù)據(jù)流通常是同一類型，由于數(shù)據(jù)集對(duì)象的類型固定，對(duì)于數(shù)據(jù)集可以只保存一份對(duì)象Schema信息，節(jié)省大量的存儲(chǔ)空間。同時(shí)，對(duì)于固定大小的類型，也可通過固定的偏移位置存取。當(dāng)我們需要訪問某個(gè)對(duì)象成員變量的時(shí)候，通過定制的序列化工具，并不需要反序列化整個(gè)Java對(duì)象，而是可以直接通過偏移量，只是反序列化特定的對(duì)象成員變量。如果對(duì)象的成員變量較多時(shí)，能夠大大減少Java對(duì)象的創(chuàng)建開銷，以及內(nèi)存數(shù)據(jù)的拷貝大小。Spark與Flink數(shù)據(jù)集都支持任意Java或是Scala類型，通過自動(dòng)生成定制序列化工具，Spark與Flink既保證了API接口對(duì)用戶的友好度(不用像Hadoop那樣數(shù)據(jù)類型需要繼承實(shí)現(xiàn)org.apache.hadoop.io.Writable接口)，同時(shí)也達(dá)到了和Hadoop類似的序列化效率。

Spark的序列化框架

Spark支持通用的計(jì)算框架，如Java Serialization和Kryo。其缺點(diǎn)之前也略有論述，總結(jié)如下：

占用較多內(nèi)存。Kryo相對(duì)于Java Serialization更高，它支持一種類型到Integer的映射機(jī)制，序列化時(shí)用Integer代替類型信息，但還不及定制的序列化工具效率。

反序列化時(shí)，必須反序列化整個(gè)Java對(duì)象。

無法直接操作序列化后的二進(jìn)制數(shù)據(jù)。

Project Tungsten 提供了一種更好的解決方式，針對(duì)于DataFrame API(Spark針對(duì)結(jié)構(gòu)化數(shù)據(jù)的類SQL分析API，參考Spark DataFrame Blog)，由于其數(shù)據(jù)集是有固定Schema的Tuple(可大概類比為數(shù)據(jù)庫中的行)，序列化是針對(duì)每個(gè)Tuple存儲(chǔ)其類型信息以及其成員的類型信息是非常浪費(fèi)內(nèi)存的，對(duì)于Spark來說，Tuple類型信息是全局可知的，所以其定制的序列化工具只存儲(chǔ)Tuple的數(shù)據(jù)，如下圖所示

圖1 Spark off-heap object layout

對(duì)于固定大小的成員，如int，long等，其按照偏移量直接內(nèi)聯(lián)存儲(chǔ)。對(duì)于變長的成員，如String，其存儲(chǔ)一個(gè)指針，指向真正的數(shù)據(jù)存儲(chǔ)位置，并在數(shù)據(jù)存儲(chǔ)開始處存儲(chǔ)其長度。通過這種存儲(chǔ)方式，保證了在反序列化時(shí)，當(dāng)只需訪問某一個(gè)成員時(shí)，只需根據(jù)偏移量反序列化這個(gè)成員，并不需要反序列化整個(gè)Tuple。

Project Tungsten的定制序列化工具應(yīng)用在Sort，HashTable，Shuffle等很多對(duì)Spark性能影響最大的地方。比如在Shuffle階段，定制序列化工具不僅提升了序列化的性能，而且減少了網(wǎng)絡(luò)傳輸?shù)臄?shù)據(jù)量，根據(jù)DataBricks的Blog介紹，相對(duì)于Kryo，Shuffle800萬復(fù)雜Tuple數(shù)據(jù)時(shí)，其性能至少提高2倍以上。此外，Project Tungsten也計(jì)劃通過Code generation技術(shù)，自動(dòng)生成序列化代碼，將定制序列化工具推廣到Spark Core層，從而使得更多的Spark應(yīng)用受惠于此優(yōu)化。

Flink的序列化框架

Flink在系統(tǒng)設(shè)計(jì)之初，就借鑒了很多傳統(tǒng)RDBMS的設(shè)計(jì)，其中之一就是對(duì)數(shù)據(jù)集的類型信息進(jìn)行分析，對(duì)于特定Schema的數(shù)據(jù)集的處理過程，進(jìn)行類似RDBMS執(zhí)行計(jì)劃優(yōu)化的優(yōu)化。同時(shí)，數(shù)據(jù)集的類型信息也可以用來設(shè)計(jì)定制的序列化工具。和Spark類似，F(xiàn)link支持任意的Java或是Scala類型，F(xiàn)link通過Java Reflection框架分析基于Java的Flink程序UDF(User Define Function)的返回類型的類型信息，通過Scala Compiler分析基于Scala的Flink程序UDF的返回類型的類型信息。類型信息由TypeInformation類表示，這個(gè)類有諸多具體實(shí)現(xiàn)類，例如(更多詳情參考Flink官方博客Apache Flink: Juggling with Bits and Bytes)：

1. BasicTypeInfo: 任意Java基本類型(裝包或未裝包)和String類型。

2. BasicArrayTypeInfo: 任意Java基本類型數(shù)組(裝包或未裝包)和String數(shù)組。

3. WritableTypeInfo: 任意Hadoop’s Writable接口的實(shí)現(xiàn)類.

4. TupleTypeInfo: 任意的Flink tuple類型(支持Tuple1 to Tuple25). Flink tuples是固定長度固定類型的Java Tuple實(shí)現(xiàn)。

5. CaseClassTypeInfo: 任意的 Scala CaseClass(包括 Scala tuples).

6. PojoTypeInfo: 任意的POJO (Java or Scala)，例如，Java對(duì)象的所有成員變量，要么是public修飾符定義，要么有g(shù)etter/setter方法。

7. GenericTypeInfo: 任意無法匹配之前幾種類型的類。)

前6種類型數(shù)據(jù)集幾乎覆蓋了絕大部分的Flink程序，針對(duì)前6種類型數(shù)據(jù)集，F(xiàn)link皆可以自動(dòng)生成對(duì)應(yīng)的TypeSerializer定制序列化工具，非常有效率的對(duì)數(shù)據(jù)集進(jìn)行序列化和反序列化。對(duì)于第7中類型，F(xiàn)link使用Kryo進(jìn)行序列化和反序列化。此外，對(duì)于可被用作Key的類型，F(xiàn)link還同時(shí)自動(dòng)生成TypeComparator，用來輔助直接對(duì)序列化后的二進(jìn)制數(shù)據(jù)直接進(jìn)行compare，hash等之類的操作。對(duì)于Tuple，CaseClass，Pojo等組合類型，F(xiàn)link自動(dòng)生成的TypeSerializer，TypeComparator同樣是組合的，并把其成員的序列化/反序列化代理給其成員對(duì)應(yīng)的TypeSerializer，TypeComparator，如下圖所示：

圖2 Flink組合類型序列化

此外，如有需要，用戶可通過集成TypeInformation接口，定制實(shí)現(xiàn)自己的序列化工具。

顯式的內(nèi)存管理

垃圾回收的JVM內(nèi)存管理回避不了的問題，JDK8的G1算法改善了JVM垃圾回收的效率和可用范圍，但對(duì)于大數(shù)據(jù)處理的實(shí)際環(huán)境中，還是遠(yuǎn)遠(yuǎn)不夠。這也和現(xiàn)在分布式框架的發(fā)展趨勢(shì)有沖突，越來越多的分布式計(jì)算框架希望盡可能多的將待處理的數(shù)據(jù)集放在內(nèi)存中，而對(duì)于JVM垃圾回收來說，內(nèi)存中Java對(duì)象越少，存活時(shí)間越短，其效率越高。通過JVM進(jìn)行內(nèi)存管理的話，OutOfMemoryError也是一個(gè)很難解決的問題。同時(shí)，在JVM內(nèi)存管理中，Java對(duì)象有潛在的碎片化存儲(chǔ)問題(Java對(duì)象所有信息可能不是在內(nèi)存中連續(xù)存儲(chǔ))，也有可能在所有Java對(duì)象大小沒有超過JVM分配內(nèi)存時(shí)，出現(xiàn)OutOfMemoryError問題。

Flink的內(nèi)存管理

Flink將內(nèi)存分為三個(gè)部分，每個(gè)部分都有不同的用途：

1. Network buffers: 一些以32KB Byte數(shù)組為單位的buffer，主要被網(wǎng)絡(luò)模塊用于數(shù)據(jù)的網(wǎng)絡(luò)傳輸。

2. Memory Manager pool: 大量以32KB Byte數(shù)組為單位的內(nèi)存池，所有的運(yùn)行時(shí)算法(例如Sort/Shuffle/Join)都從這個(gè)內(nèi)存池申請(qǐng)內(nèi)存，并將序列化后的數(shù)據(jù)存儲(chǔ)其中，結(jié)束后釋放回內(nèi)存池。

3. Remaining (Free) Heap: 主要留給UDF中用戶自己創(chuàng)建的Java對(duì)象，由JVM管理。

Network buffers在Flink中主要基于Netty的網(wǎng)絡(luò)傳輸，無需多講。Remaining Heap用于UDF中用戶自己創(chuàng)建的Java對(duì)象，在UDF中，用戶通常是流式的處理數(shù)據(jù)，并不需要很多內(nèi)存，同時(shí)Flink也不鼓勵(lì)用戶在UDF中緩存很多數(shù)據(jù)，因?yàn)檫@會(huì)引起前面提到的諸多問題。Memory Manager pool(以后以內(nèi)存池代指)通常會(huì)配置為最大的一塊內(nèi)存，接下來會(huì)詳細(xì)介紹。

在Flink中，內(nèi)存池由多個(gè)MemorySegment組成，每個(gè)MemorySegment代表一塊連續(xù)的內(nèi)存，底層存儲(chǔ)是byte[]，默認(rèn)32KB大小。MemorySegment提供了根據(jù)偏移量訪問數(shù)據(jù)的各種方法，如get/put int，long，float，double等，MemorySegment之間數(shù)據(jù)拷貝等方法，和java.nio.ByteBuffer類似。對(duì)于Flink的數(shù)據(jù)結(jié)構(gòu)，通常包括多個(gè)向內(nèi)存池申請(qǐng)的MemeorySegment，所有要存入的對(duì)象，通過TypeSerializer序列化之后，將二進(jìn)制數(shù)據(jù)存儲(chǔ)在MemorySegment中，在取出時(shí)，通過TypeSerializer反序列化。數(shù)據(jù)結(jié)構(gòu)通過MemorySegment提供的set/get方法訪問具體的二進(jìn)制數(shù)據(jù)。

Flink這種看起來比較復(fù)雜的內(nèi)存管理方式帶來的好處主要有：

1. 二進(jìn)制的數(shù)據(jù)存儲(chǔ)大大提高了數(shù)據(jù)存儲(chǔ)密度，節(jié)省了存儲(chǔ)空間。

2. 所有的運(yùn)行時(shí)數(shù)據(jù)結(jié)構(gòu)和算法只能通過內(nèi)存池申請(qǐng)內(nèi)存，保證了其使用的內(nèi)存大小是固定的，不會(huì)因?yàn)檫\(yùn)行時(shí)數(shù)據(jù)結(jié)構(gòu)和算法而發(fā)生OOM。而對(duì)于大部分的分布式計(jì)算框架來說，這部分由于要緩存大量數(shù)據(jù)，是最有可能導(dǎo)致OOM的地方。

3. 內(nèi)存池雖然占據(jù)了大部分內(nèi)存，但其中的MemorySegment容量較大(默認(rèn)32KB)，所以內(nèi)存池中的Java對(duì)象其實(shí)很少，而且一直被內(nèi)存池引用，所有在垃圾回收時(shí)很快進(jìn)入持久代，大大減輕了JVM垃圾回收的壓力。

4. Remaining Heap的內(nèi)存雖然由JVM管理，但是由于其主要用來存儲(chǔ)用戶處理的流式數(shù)據(jù)，生命周期非常短，速度很快的Minor GC就會(huì)全部回收掉，一般不會(huì)觸發(fā)Full GC。

Flink當(dāng)前的內(nèi)存管理在最底層是基于byte[]，所以數(shù)據(jù)最終還是on-heap，最近Flink增加了off-heap的內(nèi)存管理支持，將會(huì)在下一個(gè)release中正式出現(xiàn)。Flink off-heap的內(nèi)存管理相對(duì)于on-heap的優(yōu)點(diǎn)主要在于(更多細(xì)節(jié)，請(qǐng)參考Apache Flink: Off-heap Memory in Apache Flink and the curious JIT compiler)：

1. 啟動(dòng)分配了大內(nèi)存(例如100G)的JVM很耗費(fèi)時(shí)間，垃圾回收也很慢。如果采用off-heap，剩下的Network buffer和Remaining heap都會(huì)很小，垃圾回收也不用考慮MemorySegment中的Java對(duì)象了。

2. 更有效率的IO操作。在off-heap下，將MemorySegment寫到磁盤或是網(wǎng)絡(luò)，可以支持zeor-copy技術(shù)，而on-heap的話，則至少需要一次內(nèi)存拷貝。

3. off-heap可用于錯(cuò)誤恢復(fù)，比如JVM崩潰，在on-heap時(shí)，數(shù)據(jù)也隨之丟失，但在off-heap下，off-heap的數(shù)據(jù)可能還在。此外，off-heap上的數(shù)據(jù)還可以和其他程序共享。

Spark的內(nèi)存管理

Spark的off-heap內(nèi)存管理與Flink off-heap模式比較相似，也是通過Java UnSafe API直接訪問off-heap內(nèi)存，通過定制的序列化工具將序列化后的二進(jìn)制數(shù)據(jù)存儲(chǔ)與off-heap上，Spark的數(shù)據(jù)結(jié)構(gòu)和算法直接訪問和操作在off-heap上的二進(jìn)制數(shù)據(jù)。Project Tungsten是一個(gè)正在進(jìn)行中的項(xiàng)目，想了解具體進(jìn)展可以訪問：[SPARK-7075] Project Tungsten (Spark 1.5 Phase 1)，[SPARK-9697] Project Tungsten (Spark 1.6)。

緩存友好的計(jì)算

磁盤IO和網(wǎng)絡(luò)IO之前一直被認(rèn)為是Hadoop系統(tǒng)的瓶頸，但是隨著Spark，F(xiàn)link等新一代的分布式計(jì)算框架的發(fā)展，越來越多的趨勢(shì)使得CPU/Memory逐漸成為瓶頸，這些趨勢(shì)包括：

1. 更先進(jìn)的IO硬件逐漸普及。10GB網(wǎng)絡(luò)和SSD硬盤等已經(jīng)被越來越多的數(shù)據(jù)中心使用。

2. 更高效的存儲(chǔ)格式。Parquet，ORC等列式存儲(chǔ)被越來越多的Hadoop項(xiàng)目支持，其非常高效的壓縮性能大大減少了落地存儲(chǔ)的數(shù)據(jù)量。

3. 更高效的執(zhí)行計(jì)劃。例如Spark DataFrame的執(zhí)行計(jì)劃優(yōu)化器的Fliter-Push-Down優(yōu)化會(huì)將過濾條件盡可能的提前，甚至提前到Parquet的數(shù)據(jù)訪問層，使得在很多實(shí)際的工作負(fù)載中，并不需要很多的磁盤IO。

由于CPU處理速度和內(nèi)存訪問速度的差距，提升CPU的處理效率的關(guān)鍵在于最大化的利用L1/L2/L3/Memory，減少任何不必要的Cache miss。定制的序列化工具給Spark和Flink提供了可能，通過定制的序列化工具，Spark和Flink訪問的二進(jìn)制數(shù)據(jù)本身，因?yàn)檎加脙?nèi)存較小，存儲(chǔ)密度比較大，而且還可以在設(shè)計(jì)數(shù)據(jù)結(jié)構(gòu)和算法時(shí)，盡量連續(xù)存儲(chǔ)，減少內(nèi)存碎片化對(duì)Cache命中率的影響，甚至更進(jìn)一步，Spark與Flink可以將需要操作的部分?jǐn)?shù)據(jù)(如排序時(shí)的Key)連續(xù)存儲(chǔ)，而將其他部分的數(shù)據(jù)存儲(chǔ)在其他地方，從而最大可能的提升Cache命中的概率。

Flink中的數(shù)據(jù)結(jié)構(gòu)

以Flink中的排序?yàn)槔判蛲ǔＪ欠植际接?jì)算框架中一個(gè)非常重的操作，F(xiàn)link通過特殊設(shè)計(jì)的排序算法，獲得了非常好了性能，其排序算法的實(shí)現(xiàn)如下：

1. 將待排序的數(shù)據(jù)經(jīng)過序列化后存儲(chǔ)在兩個(gè)不同的MemorySegment集中。數(shù)據(jù)全部的序列化值存放于其中一個(gè)MemorySegment集中。數(shù)據(jù)序列化后的Key和指向第一個(gè)MemorySegment集中其值的指針存放于第二個(gè)MemorySegment集中。

2. 對(duì)第二個(gè)MemorySegment集中的Key進(jìn)行排序，如需交換Key位置，只需交換對(duì)應(yīng)的Key+Pointer的位置，第一個(gè)MemorySegment集中的數(shù)據(jù)無需改變。 當(dāng)比較兩個(gè)Key大小時(shí)，TypeComparator提供了直接基于二進(jìn)制數(shù)據(jù)的對(duì)比方法，無需反序列化任何數(shù)據(jù)。

3. 排序完成后，訪問數(shù)據(jù)時(shí)，按照第二個(gè)MemorySegment集中Key的順序訪問，并通過Pinter值找到數(shù)據(jù)在第一個(gè)MemorySegment集中的位置，通過TypeSerializer反序列化成Java對(duì)象返回。

圖3 Flink排序算法

這樣實(shí)現(xiàn)的好處有：

1. 通過Key和Full data分離存儲(chǔ)的方式，盡量將被操作的數(shù)據(jù)最小化，提高Cache命中的概率，從而提高CPU的吞吐量。

2. 移動(dòng)數(shù)據(jù)時(shí)，只需移動(dòng)Key+Pointer，而無須移動(dòng)數(shù)據(jù)本身，大大減少了內(nèi)存拷貝的數(shù)據(jù)量。

3. TypeComparator直接基于二進(jìn)制數(shù)據(jù)進(jìn)行操作，節(jié)省了反序列化的時(shí)間。

Spark的數(shù)據(jù)結(jié)構(gòu)

Spark中基于off-heap的排序與Flink幾乎一模一樣，在這里就不多做介紹了，感興趣的話，請(qǐng)參考：https://databricks.com/blog/2015/04/28/project-tungsten-bringing-spark-closer-to-bare-metal.html

總結(jié)

本文主要介紹了Hadoop生態(tài)圈的一些項(xiàng)目遇到的一些因?yàn)镴VM內(nèi)存管理導(dǎo)致的問題，以及社區(qū)是如何應(yīng)對(duì)的?；旧希詢?nèi)存為中心的分布式計(jì)算框架，大都開始了部分脫離JVM，走上了自己管理內(nèi)存的路線，Project Tungsten甚至更進(jìn)一步，提出了通過LLVM，將部分邏輯編譯成本地代碼，從而更加深入的挖掘SIMD等CPU潛力。此外，除了Spark，F(xiàn)link這樣的分布式計(jì)算框架，HBase(HBASE-11425)，HDFS(HDFS-7844)等項(xiàng)目也在部分性能相關(guān)的模塊通過自己管理內(nèi)存來規(guī)避JVM的一些缺陷，同時(shí)提升性能。

來源：知乎《大數(shù)據(jù)應(yīng)用與實(shí)踐》專欄