潘鋒烽 熊 勁

(計算機體系結(jié)構(gòu)國家重點實驗室(中國科學(xué)院計算技術(shù)研究所) 北京 100190)(中國科學(xué)院大學(xué) 北京 100049)(panfengfeng@ict.ac.cn)

非易失性內(nèi)存(non-volatile memory, NVM)的出現(xiàn)為解決磁盤性能瓶頸問題提供了新的機會，本文中的NVM指的是基于內(nèi)存總線接口、字節(jié)尋址的非易失內(nèi)存.在內(nèi)存計算的場景下，非易失性內(nèi)存有著非常廣泛的應(yīng)用場景.與磁盤和DRAM相比，NVM的主要優(yōu)勢有：1)NVM有著與DRAM相接近的讀寫延遲和吞吐率，有望用來消除磁盤IO開銷，提升系統(tǒng)的IO性能；2)NVM的存儲密度比DRAM更大，與NAND Flash SSD相似，它能存放更多的數(shù)據(jù)；3)相比于內(nèi)存，NVM具備非易失、可持久化的優(yōu)勢.隨著工業(yè)界大力發(fā)展非易失內(nèi)存，其進展相當(dāng)迅速，2013年Micro公司推出了搭載Flash與DRAM的Hybrid DIMM[1].2014年，AgigA Tech公司推出了DDR接口的 NVDIMM[2-4].Intel與Micro聯(lián)合推出的DDR接口的3D XPoint[5]產(chǎn)品預(yù)計2018年面市，3D XPoint閃存速度堪比內(nèi)存，因此解決磁盤性能問題的一個較為直接的方式是替換存儲介質(zhì)，將傳統(tǒng)的磁盤替換成NVM，這樣使得內(nèi)存計算中的數(shù)據(jù)讀寫能夠獲得NVM的性能，與此同時也保證了數(shù)據(jù)的持久化.

眾所周知，Shuffle的性能是決定大數(shù)據(jù)處理性能的關(guān)鍵因素之一[6-9].由于傳統(tǒng)Shuffle階段的數(shù)據(jù)一般是通過磁盤文件系統(tǒng)進行持久化，所以影響Shuffle性能的一個重要因素是IO開銷[7,10-11]，NVM有助于解決內(nèi)存計算在Shuffle階段由于持久化所帶來的IO開銷.因此本文將NVM引入到Shuffle階段中，但是由于NVM的性能接近DRAM，有研究工作表明，對于NVM現(xiàn)有的系統(tǒng)軟件(NVM文件系統(tǒng))開銷過高，不能充分發(fā)揮NVM的性能[12-15].表1展示了在壓縮[16]過程中文件系統(tǒng)各個部分的開銷比例[17].

從表1可以看出，大約60.5%的時間花費在了文件系統(tǒng)上，其中關(guān)于元數(shù)據(jù)的開銷依然占據(jù)著較大的比例.上述結(jié)果表明，即使使用最新的NVM文件系統(tǒng)，其開銷也是非常大的.如何高效使用NVM來提升Shuffle階段的IO性能是當(dāng)前內(nèi)存計算使用NVM所面臨的一個重要問題與挑戰(zhàn).

Table 1 Percentage of Time Spent in File System表1 文件系統(tǒng)中各個部分的開銷比例[17]

在本文中，我們提出了一種基于NVM的Shuffle優(yōu)化策略——NV-Shuffle.為了最大化發(fā)揮NVM的性能優(yōu)勢，NV-Shuffle摒棄了以文件系統(tǒng)進行Shuffle數(shù)據(jù)存取的方式，而是采用持久化內(nèi)存的方式，直接在用戶態(tài)訪問持久化內(nèi)存，避免了傳統(tǒng)存儲系統(tǒng)中的冗長的IO路徑，例如文件系統(tǒng)、設(shè)備驅(qū)動等.

本文的主要貢獻如下：

1) 利用pVM構(gòu)建Java的持久化內(nèi)存訪問接口——NV-Shuffle接口，使大數(shù)據(jù)平臺能夠直接使用與訪問NVM；

2) 針對NVM提出了一種Shuffle數(shù)據(jù)的組織方式——基于Hash的私有持久化緩沖區(qū)，從而能夠高效處理并發(fā)、故障、網(wǎng)絡(luò)傳輸?shù)确矫娴膯栴}；

3) 針對傳統(tǒng)Shuffle階段預(yù)先創(chuàng)建文件的問題，提出一種符合NVM空間自身的分配策略——延遲分配，從而能夠提升NVM的空間利用率；

4) 對于Shuffle數(shù)據(jù)的讀取與恢復(fù)，通過使用映射表方式來管理，使其能夠快速定位、讀取數(shù)據(jù)以及在出現(xiàn)失效之后進行快速的數(shù)據(jù)恢復(fù).

1 相關(guān)工作

本節(jié)主要分為2個部分：傳統(tǒng)計算框架關(guān)于Shuffle的優(yōu)化以及NVM的相關(guān)工作.

1.1 Shuffle優(yōu)化

傳統(tǒng)大數(shù)據(jù)處理平臺中的Shuffle階段都是使用單一類型的存儲設(shè)備(Disk或者SSD)，以文件系統(tǒng)的方式進行數(shù)據(jù)的存取，并沒有涉及到異構(gòu)存儲的使用.因此關(guān)于該部分的優(yōu)化主要圍繞的是在使用磁盤的場景下，如何減少磁盤的IO開銷，主要涉及2個問題：

1) 數(shù)據(jù)重復(fù)讀寫引起的磁盤IO開銷；

2) 數(shù)據(jù)拉取過程中隨機、小粒度的IO模式引起的磁盤IO開銷.

根據(jù)不同的優(yōu)化方法，分為內(nèi)存加速、磁盤IO聚合和高速網(wǎng)絡(luò)加速.

1.1.1 內(nèi)存加速

在Shuffle階段，都會存在數(shù)據(jù)重復(fù)讀寫的問題，由于節(jié)點分配給每個Task的內(nèi)存是有限的，使得數(shù)據(jù)多次從磁盤中讀取出來之后寫回，例如Map端可能會使數(shù)據(jù)重復(fù)2次讀寫，Reduce端則可能會更多，這無疑引起較多的磁盤IO開銷，影響作業(yè)的執(zhí)行效率.針對該問題，Themis系統(tǒng)[18]提出了“2-IO property”，即作業(yè)在處理數(shù)據(jù)的過程中，數(shù)據(jù)從磁盤的讀寫次數(shù)只有2次(HDFS和中間結(jié)果各一次)，其余過程都不會與磁盤交互.為了達到該目標(biāo)，Themis系統(tǒng)在Shuffle階段使用了動態(tài)內(nèi)存分配策略對該過程中的數(shù)據(jù)進行存儲，從而大幅度減少磁盤IO開銷，降低了作業(yè)時間.類似地，SpongeFiles[19]發(fā)現(xiàn)由于某些作業(yè)存在數(shù)據(jù)傾斜的問題，引發(fā)的一個現(xiàn)象是某些Task的內(nèi)存空間已經(jīng)用完且觸發(fā)了Spill，而某些Task的內(nèi)存空間還有剩余，如圖1(a).基于該現(xiàn)象，SpongeFiles通過共享Task中未使用的內(nèi)存空間，如圖1(b)，減少Spill的次數(shù)，從而降低了磁盤IO開銷.

Fig. 1 The way how task organizes memory圖1 Task內(nèi)存的使用方式

1.1.2 聚合磁盤IO

在Shuffle階段過程中，Reduce端在讀取過程中會存在大量的小粒度、隨機讀，從而引起磁盤的大量尋道，降低IO性能，延長作業(yè)執(zhí)行時間，對于磁盤而言，適用的IO模式是大粒度、順序的.因此當(dāng)中間結(jié)果的數(shù)據(jù)量較大時，磁盤會出現(xiàn)性能瓶頸.為了解決該問題，Sailfish系統(tǒng)[20]采用了“Batching Data IO”的技術(shù)來避免Shuffle讀取時產(chǎn)生的大量尋道問題，即寫Shuffle數(shù)據(jù)時，聚集每個Map Task相對應(yīng)的分區(qū)的數(shù)據(jù)，利用分布式文件系統(tǒng)來存儲相應(yīng)的數(shù)據(jù)，從而在讀取中間結(jié)果時，原先的隨機、小粒度的讀轉(zhuǎn)換成了順序、大粒度的讀，加速了中間結(jié)果的數(shù)據(jù)讀取，如圖2所示，但是存在的問題是將中間結(jié)果寫入到分布式文件系統(tǒng)中，需要涉及到2次網(wǎng)絡(luò)IO：前期寫入和后期讀取，而原始Shuffle將中間結(jié)果寫入到本地磁盤時，網(wǎng)絡(luò)IO只涉及到了“后期讀取”，因此在考慮使用分布式文件系統(tǒng)存儲中間結(jié)果時，還需要考慮當(dāng)前集群的網(wǎng)絡(luò)環(huán)境.

1.1.3 高速網(wǎng)絡(luò)加速

Shuffle過程中，Reduce端拉取數(shù)據(jù)會引起重復(fù)數(shù)據(jù)的讀寫，重復(fù)數(shù)據(jù)的讀寫意味著大量的磁盤操作，并且以小粒度、隨機為主，因此當(dāng)中間結(jié)果的數(shù)據(jù)量較大時，該階段可能會存在磁盤性能瓶頸.

Fig. 2 Batching data IO圖2 聚合磁盤IO

Hadoop-A[21]和Hadoop-IB[22]利用高速網(wǎng)絡(luò)(RDMA[23-24])的特性，通過高效的算法，避免了重復(fù)數(shù)據(jù)的讀寫.大致流程:首先建立關(guān)于key的優(yōu)先隊列；然后Reduce根據(jù)優(yōu)先隊列中key的順序，進行數(shù)據(jù)的Shuffle，Merge，Reduce操作.優(yōu)先隊列的主要作用是使得Shuffle,Merge,Reduce三階段進行pipeline：由于key是有序排列的，因此根據(jù)key的順序從Map端進行讀取，從而避免了Reduce端的重復(fù)數(shù)據(jù)讀寫.

1.2 非易失內(nèi)存相關(guān)工作

隨著非易失內(nèi)存NVM(主要指的是持久化內(nèi)存)的發(fā)展，一部分研究工作集中于如何對該存儲介質(zhì)上的數(shù)據(jù)進行訪問和管理，大致可以分為2類：1)提供文件系統(tǒng)抽象，即基于NVM的文件系統(tǒng)(NVM file system)；2)提供持久化內(nèi)存抽象，上層應(yīng)用可通過新的接口或者編程模型來直接使用NVM(persistent regionsheaps).

Fig. 3 The stack of Linux file system圖3 Linux IO軟件棧

1.2.1 基于NVM的文件系統(tǒng)

傳統(tǒng)文件系統(tǒng)主要是針對慢速塊設(shè)備(例如磁盤)而設(shè)計的，如圖3所示，應(yīng)用程序關(guān)于文件的讀寫請求發(fā)出之后，通過虛擬文件系統(tǒng)到達文件所在的文件系統(tǒng)，例如EXT4，并且需要經(jīng)過通用塊層、IO調(diào)度層和塊設(shè)備驅(qū)動等軟件層次才能最終到達存儲設(shè)備，而對于內(nèi)存文件系統(tǒng)而言，其訪問層次就明顯減少，它主要利用高度優(yōu)化的內(nèi)存管理來訪問內(nèi)存.

典型的工作，例如BPFS[25]，SCMFS[26]，PMFS[27]等.BPFS是由Condit等人在2009年提出的一種關(guān)于字節(jié)尋址的持久化內(nèi)存文件系統(tǒng)，它主要利用了NVM的兩大特性，即字節(jié)尋址和原地更新.BPFS以內(nèi)存的方式來管理持久化內(nèi)存，并在其上構(gòu)建文件系統(tǒng)的數(shù)據(jù)結(jié)構(gòu)，避免了數(shù)據(jù)拷貝，通過short-circuit shadow paging技術(shù)來支持?jǐn)?shù)據(jù)的原子性、細粒度、一致的更新，從而使得機器重啟之后仍能正確訪問其中的數(shù)據(jù)[25].Wu等人在2011年提出了一個在虛擬地址空間中實現(xiàn)的文件系統(tǒng)SCMFS，該系統(tǒng)利用內(nèi)存管理單元實現(xiàn)虛擬地址到物理地址的轉(zhuǎn)換，并且利用現(xiàn)有的內(nèi)存管理模塊進行塊管理，使得每個文件的虛擬地址空間都是連續(xù)的，簡化了文件操作，減少了CPU開銷，另外它還采用了空間預(yù)分配機制以及相應(yīng)的垃圾回收機制[26].Intel在2014年提出了基于NVM的文件系統(tǒng)——PMFS，它繞開了傳統(tǒng)文件系統(tǒng)的緩存層，而直接對持久化內(nèi)存進行訪問，通過該種方式能夠有效避免數(shù)據(jù)拷貝，大幅提升文件系統(tǒng)的性能[27].

基于NVM的文件系統(tǒng)主要設(shè)計目標(biāo)是降低傳統(tǒng)文件系統(tǒng)軟件棧的開銷，并且利用NVM的特性來提升文件系統(tǒng)的性能，如字節(jié)尋址、原子更新、高性能的隨機訪問，但是這些文件系統(tǒng)都是基于虛擬文件系統(tǒng)，文件系統(tǒng)本身的開銷還是存在的，例如元數(shù)據(jù)的管理、名字空間的維護，這些對于文件系統(tǒng)而言也是一個較大的開銷，無法最大化發(fā)揮NVM的性能優(yōu)勢，因此有一部分研究工作集中于為上層應(yīng)用程序提供用戶態(tài)的編程接口，程序能夠在用戶態(tài)訪問NVM.

1.2.2 持久化內(nèi)存的編程模型與接口

傳統(tǒng)存儲系統(tǒng)中，數(shù)據(jù)結(jié)構(gòu)存儲2種格式：內(nèi)存格式和磁盤格式.這2種格式之間的轉(zhuǎn)換只有在數(shù)據(jù)持久化的過程中進行.而在持久化內(nèi)存中，數(shù)據(jù)結(jié)構(gòu)可以直接持久化在內(nèi)存中，無需進行格式的轉(zhuǎn)換.一些研究工作[13,15,28-29]利用該特性簡化了NVM的使用，例如，Volos等人提出的輕量級主存系統(tǒng)Mnemosyne[15]在PCM設(shè)備能保證64 b原子寫的假設(shè)下修改系統(tǒng)庫函數(shù)，為程序員設(shè)計了一個直接訪問非易失存儲器的接口，對于開發(fā)過程中想要持久化的數(shù)據(jù), 只需用特定的數(shù)據(jù)類型pstatic聲明即可.Coburn等人提出了一個輕量級高性能持久化對象系統(tǒng)——NV-heaps(non-volatile memory heaps)[13]，它是基于BPFS[25]的硬件設(shè)計結(jié)構(gòu)，為程序員提供了包括對象、指針、主存分配等接口，實現(xiàn)指針安全(防止非易失存儲上的指針指向易失性存儲介質(zhì)的情況發(fā)生)、ACID事務(wù)處理、傳統(tǒng)API服務(wù)、高性能以及可靠的功能，以便在系統(tǒng)崩潰、斷電或其他常見失效發(fā)生時保護非易失存儲設(shè)備上數(shù)據(jù)的正確性.另外，NV-Tree[30]，CDDS[31]等工作提出了一個基于NVM的一致性、可持久化的數(shù)據(jù)結(jié)構(gòu)，使得應(yīng)用可以直接使用相應(yīng)的數(shù)據(jù)結(jié)構(gòu)，而不需要關(guān)心數(shù)據(jù)結(jié)構(gòu)在持久化內(nèi)存上的具體實現(xiàn)細節(jié).上述工作主要還是在基于文件系統(tǒng)之上(粗粒度的空間管理)提供了細粒度的數(shù)據(jù)管理、空間分配與釋放、持久化以及事務(wù)管理，而pVM[17]，它是基于VM(virtual memory)，而不是文件系統(tǒng)，因此在性能上，pVM更勝一籌.

2 NV-Shuffle設(shè)計與實現(xiàn)

NV-Shuffle的主要設(shè)計目標(biāo)是高效利用NVM加速Shuffle階段，其核心思想是基于持久化內(nèi)存的訪問接口進行NV-Buffer的分配與數(shù)據(jù)讀寫，通過Hash-based Private NV-Buffer進行Shuffle數(shù)據(jù)的組織，使用KeyValue方式進行NV-Buffer的管理.

Fig. 4 The architecture of NV-Shuffle圖4 NV-Shuffle架構(gòu)圖

圖4展示了NV-Shuffle的總體框架.從圖4中可以看出，NV-Buffer是Shuffle數(shù)據(jù)存儲的基本單位，在Shuffle過程中，通過NVmalloc和NVfree接口進行NV-Buffer的分配與釋放，而對于NV-Buffer的訪問與傳統(tǒng)的Memory訪問類似.至于Shuffle過程中數(shù)據(jù)如何組織、NV-Buffer如何分配與管理都將在下面幾節(jié)進行詳細描述.

2.1 NV-Shuffle接口

傳統(tǒng)文件系統(tǒng)主要是為了磁盤而設(shè)計的，而對于NVM而言，其性能接近于DRAM，而且CPU可直接通過指令訪問，導(dǎo)致文件系統(tǒng)自身的開銷過大，例如元數(shù)據(jù)的管理、名字空間的維護等，而針對這些問題，現(xiàn)有的一些研究針對NVM重新設(shè)計了新的文件系統(tǒng)，例如BPFS[25]，SCMFS[26]，PMFS[27]等，使其避免或者簡化了文件系統(tǒng)的設(shè)計，從而獲取較高的性能.

NVM內(nèi)存除了用于外存儲(文件系統(tǒng))和易失性內(nèi)存外，利用其非易失特性，還可以實現(xiàn)持久化內(nèi)存，比如，近年來的Mnemosyne[28]，NV-Heaps[29]，pVM[17]等工作，它們提供了用戶態(tài)的編程接口，使程序能夠在用戶態(tài)訪問非易失主存, 避免了傳統(tǒng)存儲系統(tǒng)中的冗長路徑，包括文件系統(tǒng)、設(shè)備驅(qū)動等.在持久化內(nèi)存中實現(xiàn)持久化的數(shù)據(jù)結(jié)構(gòu)，保證系統(tǒng)掉電后數(shù)據(jù)的一致性和可訪問性，從而使得程序可直接在用戶態(tài)進行持久化數(shù)據(jù)結(jié)構(gòu)的高效訪問.而pVM與Mnemosyne,NV-Heap的區(qū)別在于前者是基于VM(virtual memory)實現(xiàn)的，而后者是基于VFS(virtual file system).相比于Mnemosyne,NV-Heap，pVM在擴容、性能等方面存在優(yōu)勢.

對于Shuffle數(shù)據(jù)而言，只需要保證它們在作業(yè)運行期間可靠地、持久化地存儲，持久化內(nèi)存比文件系統(tǒng)有更小的額外開銷，因此，我們采用基于持久化內(nèi)存的Shuffle數(shù)據(jù)存儲.NV-Shuffle接口需要滿足以下3個特征.

1) 基于持久化內(nèi)存的接口訪問NVM.①提供類似于Posix mallocfree的接口；②摒棄傳統(tǒng)文件系統(tǒng)的方式來存儲Shuffle數(shù)據(jù)；③應(yīng)用程序可以通過接口直接使用或者獲取NVM空間.

2) 數(shù)據(jù)可持久化.當(dāng)操作系統(tǒng)或者應(yīng)用由于故障等原因崩潰并重新啟動時，NVM中的數(shù)據(jù)可以進行恢復(fù)，以便后續(xù)進行數(shù)據(jù)的讀取.

3) 接口的適用性.能夠在大數(shù)據(jù)平臺上使用.當(dāng)前大多數(shù)的大數(shù)據(jù)平臺都是基于JavaScala而編寫的，因此能夠提供基于Java的接口以適應(yīng)大數(shù)據(jù)的場景.

NV-Shuffle使用持久化內(nèi)存分配和回收接口：NVmalloc和NVfree，其實現(xiàn)主要借鑒了pVM，由于pVM本身提供的是C++的接口，我們在JVM中添加了NVmalloc和NVfree接口，即用其封裝了pVM相應(yīng)的接口，以供大數(shù)據(jù)平臺上的應(yīng)用使用.但是在性能評測中，我們利用內(nèi)存來模擬 NVM，并且使用Java自帶的分配堆外內(nèi)存接口進行內(nèi)存的分配與使用.

2.2 基于NVM的Shuffle數(shù)據(jù)組織方式

傳統(tǒng)基于文件系統(tǒng)的Shuffle數(shù)據(jù)的組織方式有2種：Sort-based組織方式和Hash-based組織方式.對于Sort-based Shuffle，每個Task最終產(chǎn)生一個文件，且在該文件中的數(shù)據(jù)是根據(jù)partition ID排序的.在最終文件形成的過程中，可能會存在多個文件，而最終文件是通過將這幾個文件合并而成.因此Sort-based Shuffle的主要開銷在于Sort,Spill,Merge，并且在合并的過程中數(shù)據(jù)需要進行序列化和反序列化，增加了開銷；而對于Hash-based Shuffle，每個partition ID都會對應(yīng)一個文件，優(yōu)勢在于不需要進行Sort,Spill，但是劣勢在于可能會存在大量的小文件，并且小文件同時打開，會占用較多的內(nèi)存空間.

對于基于NVM的Shuffle數(shù)據(jù)的組織方式，主要從2個方面考慮.

1) Shuffle數(shù)據(jù)本身的特征：①Shuffle數(shù)據(jù)通過partition ID劃分成多個分區(qū)，每個分區(qū)的數(shù)據(jù)存儲在一起，分區(qū)之間按partition ID的順序進行存儲；②多個Task同時進行各自Shuffle數(shù)據(jù)的讀寫，即多個Map Task之間不共享Shuffle數(shù)據(jù)；

2) NVM與磁盤之間的區(qū)別：NVM與磁盤的最大區(qū)別在于小粒度、隨機IO的性能，其中磁盤性能受小粒度、隨機IO的影響較大，而NVM的小粒度、隨機IO的性能基本上與順序IO性能相同.

綜上所述，對于NVM而言，Shuffle數(shù)據(jù)的組織方式采用Hash-based更為合適：可以減少Sort,Spill,Merge的性能開銷；而且對于NVM而言沒有小文件和隨機IO的問題.

根據(jù)Task,Partition,NV-Buffer三者之間的關(guān)系，又可以將Hash-based NV-Buffer分成2種形式：基于Hash的共享持久化緩沖區(qū)(Hash-based Shared NV-Buffer)和基于Hash的私有持久化緩沖區(qū)(Hash-based Private NV-Buffer).

2.2.1 基于Hash的共享持久化緩沖區(qū)

Shared NV-Buffer是通過partition ID進行區(qū)分的，也就是，不同Task上具有相同partition ID的keyvalue存儲在同一個NV-Buffer上，如圖5所示：

Fig. 5 Hash-based Shared NV-Buffer圖5 基于Hash的共享持久化緩存

使用Hash-based Shared NV-Buffer的優(yōu)勢在于：1)NV-Buffer的空間利用率較高；2)申請NV-Buffer的開銷較小.但是該種方式的劣勢在于：

1) 多個Task可能并發(fā)地往同一個NV-Buffer中寫數(shù)據(jù)，這時會由于并發(fā)寫而帶來鎖開銷；

2) 由于所有Task中相同partition ID的數(shù)據(jù)都聚集在同一個NV-Buffer中，假設(shè)其中一個Task執(zhí)行失敗了，或者某一個Task執(zhí)行較慢，此時系統(tǒng)會在其他節(jié)點上同時啟動一個Task處理相同的數(shù)據(jù)，此時就需要對失敗的或者執(zhí)行較慢的Task所產(chǎn)生的數(shù)據(jù)進行清理與刪除.

2.2.2 基于Hash的私有持久化緩沖區(qū)

Private NV-Buffer是通過partition ID和Task進行區(qū)分，也就是，每個Task的每個partition ID都會對應(yīng)一個NV-Buffer，這種方式與傳統(tǒng)Hash-based的方式是類似的，如圖6所示：

Fig. 6 Hash-based Private NV-Buffer圖6 基于Hash的私有化持久化緩存

而使用Hash-based Private NV-Buffer的優(yōu)勢在于：1)Task并發(fā)寫時沒有鎖競爭開銷；2)Task之間的數(shù)據(jù)是通過各自的NV-Buffer完全隔離，因此失效Task的數(shù)據(jù)直接進行刪除即可，但是這種方式的劣勢在于：

1) 空間利用率低.每個Task上不同partition ID的數(shù)據(jù)量是不相同的，可能差異非常大，統(tǒng)一進行分配NV-Buffer則會造成某些NV-Buffer空間利用率低，導(dǎo)致NVM空間的浪費.

2) 申請NV-Buffer的次數(shù)較多.每個Task的每個partition ID都會申請一個NV-Buffer，申請的次數(shù)是Shared NV-Buffer的M倍，其中M表示Map Task的個數(shù)，因此可能會在一定程度上存在性能開銷.

2.2.3 設(shè)計抉擇

通過對Hash-based Shared NV-Buffer和Hash-based Private NV-Buffer的優(yōu)缺點分析，我們采用Hash-based Private NV-Buffer的方式對Shuffle數(shù)據(jù)進行組織，其原因主要有以下2點：

1) Hash-based Shared NV-Buffer的問題主要來自于“多個Task共享一個NV-Buffer”，共享NV-Buffer帶來2個方面的問題：并發(fā)帶來的開銷和失效處理時帶來的開銷.另外，根據(jù)第2.2節(jié)對于Shuffle階段特征的描述，Hash-based Shared NV-Buffer與Shuffle階段的特征(Task之間互不影響)不匹配；

2) Hash-based Private NV-Buffer基本上與Shuffle階段的特征匹配.不過，它有2個方面的劣勢：空間利用率和性能開銷，但是相比于共享NV-Buffer所帶來的2個問題，Hash-based Private NV-Buffer所面臨的上述2個劣勢更易解決.

2.3 NVM空間(NV-Buffer)的分配策略

在2.2.3小節(jié)中，我們通過Hash-based Private NV-Buffer的方式來組織Shuffle數(shù)據(jù)，但是Hash-based Private NV-Buffer存在2個問題：空間利用率低和性能開銷大.而影響空間利用率和性能開銷的一個重要因素在于空間分配粒度——NV-Buffer粒度.一方面，NV-Buffer的粒度較小，則會引發(fā)高頻率的空間分配，從而影響性能；另一方面，NV-Buffer的粒度較大，則會降低空間利用率，從而造成NVM空間的浪費.因此，NV-Buffer粒度的選取是空間和性能上的一種權(quán)衡.

2.3.1 NV-Buffer粒度的選取

首先，我們通過實驗來觀察不同NV-Buffer粒度下的性能開銷問題.該實驗主要集中在Map Task端，即事先為Map Task中的每個partition分配NV-Buffer，而partition的個數(shù)等于Reduce Task的個數(shù)，分配完成之后進行Map Task的計算.在實驗中，我們假設(shè)負載的數(shù)據(jù)分布是均勻的，即Map Task中每個partition的數(shù)據(jù)量基本上是相同的，因此partition數(shù)據(jù)量的計算公式如下所示：

(1)

其中Map Task所處理的數(shù)據(jù)量一般為Block Size的大小B，而partition個數(shù)為Reduce Task的個數(shù)R.因此在執(zhí)行過程中，一個Map Task需要執(zhí)行R次NV-Buffer的空間分配，且每次分配的空間大小約為BR，對于不同Reduce Task個數(shù)的結(jié)果如圖7所示：

Fig. 7 The time of malloc in Map Task圖7 Map Task中malloc所花費的時間

從圖7中可以看出：1)malloc的開銷隨著Reduce個數(shù)逐漸增加而上升，其主要原因在于Reduce個數(shù)增多，malloc執(zhí)行的次數(shù)也就相應(yīng)的增加，從而增加了總的malloc的開銷；2)malloc所花費的時間較小，以Reduce個數(shù)為215為例，一個map task執(zhí)行了215次malloc所所花費的時間不到1 s，因此從總體上看malloc的開銷較小，因此即使malloc次數(shù)較多(控制在一定范圍之內(nèi))，對于job整體上的性能影響也是有限的，所以在NV-Buffer粒度的選取上，由于malloc的開銷比較小，為了能夠保證NVM空間較高的利用率，NV-Buffer粒度以小粒度為基準(zhǔn)進行分配(默認為8 KB).

2.3.2 延遲分配策略

選取了NV-Buffer粒度之后，需要解決如何進行NV-Buffer的申請與分配.在Hash-based Private NV-Buffer中，按照之前的流程需要進行M×R次NV-Buffer的申請與分配，然而，在一些場景下，可能某些partition并不存在數(shù)據(jù)，例如，

1) 工作負載的數(shù)據(jù)本身存在傾斜，導(dǎo)致某些partition的數(shù)據(jù)多，而某些partition的數(shù)據(jù)少甚至沒有；

Fig. 8 The impact of partition size on data distribution圖8 partition粒度對于數(shù)據(jù)分布的影響

2) partition粒度過細，即Reduce個數(shù)較多.從圖8中可以看出，當(dāng)Reduce個數(shù)較少時，基本上不存在size=0的partition，而當(dāng)Reduce個數(shù)較多是，size=0的partition個數(shù)上升，例如當(dāng)Reduce個數(shù)為2 048時，size=0的個數(shù)占到了約20%，而一般地，Reduce個數(shù)設(shè)置較大的作業(yè)在集群規(guī)模較大(成百上千個節(jié)點)、輸入數(shù)據(jù)量較大(TB級)的場景下是常見的[20].

因此，對于這些沒有數(shù)據(jù)的partition而言，并不需要進行NV-Buffer的分配，而在原始流程中.會根據(jù)partition的個數(shù)首先創(chuàng)建好相應(yīng)個數(shù)的文件，然后進行數(shù)據(jù)的寫入，那么相對應(yīng)的，對于NV-Shuffle而言，首先會分配一定大小的NV-Buffer(8 KB)，然后往NV-Buffer中寫數(shù)據(jù)，當(dāng)NV-Buffer寫滿時，會繼續(xù)申請新的NV-Buffer來存儲Shuffle數(shù)據(jù)，如圖9所示：

Fig. 9 Original procedure圖9 原始流程(預(yù)先分配)

對于文件而言，沒有數(shù)據(jù)的寫入時，只會存在少量的元數(shù)據(jù)；而對于NV-Buffer而言，事先分配了相應(yīng)粒度的NV-Buffer，如果沒有數(shù)據(jù)的寫入，雖然NV-Buffer以小粒度分配，但還是會造成空間上的浪費，因此，我們將原始的預(yù)先分配流程進行優(yōu)化，采用延遲分配的方式，即只有當(dāng)partition有數(shù)據(jù)時，才會申請相應(yīng)的NV-Buffer，如果沒有數(shù)據(jù)則始終不進行申請，具體流程如圖10所示：

Fig. 10 Late allocation policy圖10 延遲分配策略

從圖10可以看出，通過延遲分配能夠避免預(yù)先分配中為沒有數(shù)據(jù)的partition而分配的NV-Buffer.

2.4 NV-Buffer管理

NV-Shuffle摒棄了文件系統(tǒng)，而采用NV-Buffer的方式來使用NVM，并且由于NV-Shuffle采用Hash-based Private NV-Buffer的方式，會生成非常多的NV-Buffer，如何對這些NV-Buffer進行管理是一個非常重要的問題.

首先，我們需要了解Shuffle階段對于NV-Buffer的訪問模式，如圖11所示：

Fig. 11 NV-Buffer access pattern圖11 NV-Buffer訪問模式

從圖11可以看出，NV-Buffer的訪問模式較為簡單，當(dāng)進行讀取時，不同Task需要讀取與之對應(yīng)的數(shù)據(jù)，而這些數(shù)據(jù)是根據(jù)partition ID進行區(qū)分的.在讀取的過程中，一個partition ID會對應(yīng)多個NV-Buffer，因此需要一個數(shù)據(jù)結(jié)構(gòu)來存放partition ID與NV-Buffer之間的關(guān)系，綜上所述，對于NV-Buffer組織與管理需求分為以下3個方面.

1) 簡單.

2) 快速查詢.直接通過partition ID查詢NV-Buffer.

3) 數(shù)據(jù)聚合.相同partition ID的NV-Buffer聚合.

根據(jù)上述特征，我們采用映射表的方式來存儲partition ID與NV-Buffer之間的關(guān)系，并且需要考慮的是一個partition ID會對應(yīng)多個NV-Buffer的場景.

當(dāng)一個Task寫滿一個NV-Buffer，會將相應(yīng)的〈partition ID，NV-Buffer〉插入到映射表中，而當(dāng)一個Task執(zhí)行完成之后，會將該映射表中的內(nèi)容上傳到Driver中，由Driver來維護映射表，主要有3個作用：

1) 信息只有在執(zhí)行完成后才能上傳，失效的Task不上傳，保證了后續(xù)的Task所讀到的內(nèi)容都是正確的；

2) Reduce Task會根據(jù)Driver中記錄的信息，到相應(yīng)的節(jié)點上去拉取數(shù)據(jù)，如圖12所示；

Fig. 12 Shuffle數(shù)據(jù)位置信息的獲取圖12 Access to the address of shuffle data

3) 當(dāng)數(shù)據(jù)所在的Executor由于某些原因崩潰而重新啟動一個新的Executor，Reduce Task還是能夠通過Driver中記錄的NV-Buffer信息進行數(shù)據(jù)的拉取，從而避免了重計算的開銷.原始Spark中，Reduce Task也是按照上述流程進行數(shù)據(jù)的讀取，但是從第3.5節(jié)中關(guān)于失效恢復(fù)的測試中可以看出，原始Spark中Shuffle數(shù)據(jù)是與Executor綁定的(Executor ID)，當(dāng)一個Executor崩潰后，重啟的Executor不復(fù)用原來Executor的ID，所以，即便原來的Executor崩潰前已運行結(jié)束的Task的Shuffle數(shù)據(jù)已完好保存下來，在失效恢復(fù)后也無法利用，需要重新計算.而對于NV-Shuffle而言，我們對于NV-Buffer的管理與組織不依賴于Executor ID，從而使得在失效恢復(fù)時，能夠快速找到崩潰Executor的Shuffle數(shù)據(jù).

3 實驗與結(jié)果

3.1 實驗環(huán)境

3.1.1 測試平臺以及Spark集群配置信息

由于在評測中，我們直接使用內(nèi)存來模擬NVM，所以服務(wù)器上的內(nèi)存主要有2個用途：Spark自身使用以及Shuffle階段的數(shù)據(jù)存儲.因此關(guān)于Spark主要參數(shù)的設(shè)置，如表2所示.

3.1.2 測試負載

根據(jù)Shuffle階段數(shù)據(jù)量的不同，我們將負載大體分為2類：Shuffle-heavy和Shuffle-light.其中Shuffle-heavy的負載指Shuffle階段的數(shù)據(jù)量較大(通常與輸入數(shù)據(jù)量相同，或者遠大于輸入數(shù)據(jù)量)，而Shuffle-light的負載指Shuffle階段的數(shù)據(jù)量較小(通常要遠小于輸入數(shù)據(jù)量).因此在本次實驗中我們使用了3種不同類型的負載：1)Sort(Shuffle-heavy，即中間結(jié)果數(shù)據(jù)量≈輸入數(shù)據(jù)量)；2)Word-count(Shuffle-light，即中間結(jié)果數(shù)據(jù)量?輸入數(shù)據(jù)量)；3)TPC-H[32]的22條查詢語句，每一條語句所執(zhí)行的操作不同，因此為了得到其輸入數(shù)據(jù)量與中間結(jié)果數(shù)據(jù)量之間的關(guān)系，我們使用TPC-H官方的數(shù)據(jù)生成器，生成了160 GB的原始數(shù)據(jù)，并將其導(dǎo)入到HDFS中.TPC-H的數(shù)據(jù)是由8張不同的表構(gòu)成，在160 GB的原始數(shù)據(jù)量中，每張表的數(shù)據(jù)量分為是：lineitem 120 GB, orders 27 GB, partsupp 19 GB, customer 3.7 GB, part 3.7 GB, supplier 219 MB, nation 2.2 KB， region 389 B.通過Spark SQL執(zhí)行22條TPC-H查詢語句，執(zhí)行結(jié)果如表3所示.

Table 2 The Configuration of Spark表2 Spark配置信息

從輸入數(shù)據(jù)量和中間結(jié)果數(shù)據(jù)量來看，22條查詢語句有不同的特征，主要可以分為以下3類：

1) Shuffle數(shù)據(jù)量?輸入數(shù)據(jù)量，例如Q8，Q9，Q21等；

2) Shuffle數(shù)據(jù)量≈輸入數(shù)據(jù)量，例如Q4，Q12，Q20等;

3) Shuffle數(shù)據(jù)量?輸入數(shù)據(jù)量，例如Q1，Q6，Q14等.

而對于NV-Shuffle而言，其有利的場景應(yīng)該包含2個方面：1)輸入數(shù)據(jù)量較大；2)中間結(jié)果數(shù)據(jù)量≈輸入數(shù)據(jù)量或者 Shuffle數(shù)據(jù)量?輸入數(shù)據(jù)量.由于節(jié)點的內(nèi)存是有限，因此我們在選取數(shù)據(jù)集大小時，需要考慮集群內(nèi)存是否能夠存儲中間結(jié)果數(shù)據(jù)，集群內(nèi)存與Shuffle數(shù)據(jù)量之間的關(guān)系需要滿足式(2)：

集群剩余總內(nèi)存>2×Shuffle數(shù)據(jù)量,

(2)

其中，集群剩余總內(nèi)存表示的是集群留給Shuffle數(shù)據(jù)的存儲空間，Shuffle數(shù)據(jù)在job的過程中會存儲2份，即Map端1份，Reduce端1份.因此在條件中Shuffle數(shù)據(jù)所占用的空間需要是原先的2倍，只有當(dāng)集群剩余總內(nèi)存大于2倍Shuffle數(shù)據(jù)量，該job才能順利執(zhí)行，否則失敗.當(dāng)然，該限制條件主要是由于當(dāng)前的測試環(huán)境造成的.所以，對于上述負載所采取的數(shù)據(jù)量：1)Sort：～100 GB；2)Wordcount：～256 GB；3)TPC-H：～160 GB.

Fig. 13 The execution time of Sort and Wordcount圖13 Sort和Wordcount的執(zhí)行時間

Table 3 The Basic Characters of TPC-H Queries表3 TPC-H Queries的基本特征

3.1.3 對比對象

原始的Shuffle是使用文件系統(tǒng)存儲Shuffle數(shù)據(jù)，如果想將NVM應(yīng)用于原始Shuffle，一種簡單的辦法是在NVM上創(chuàng)建一個本地文件系統(tǒng).由于我們是用內(nèi)存模擬NVM，因此可以在NVM上先創(chuàng)建塊設(shè)備(用RAMDisk)，再在RAMDisk上創(chuàng)建Ext4.此外，專門針對內(nèi)存設(shè)計的文件系統(tǒng)(如tmpfs)，也可以在NVM上直接使用，但它沒有持久化保存數(shù)據(jù)的功能，所以其性能優(yōu)勢在于消除了持久化數(shù)據(jù)帶來的開銷.因此，我們選擇了基于RAMDisk的Ext4和tmpfs做對比對象.

Table 4 Shuffle Data Storage表4 Shuffle數(shù)據(jù)的存儲方式

根據(jù)Shuffle模式不同，我們使用了常見的模式Hash-based和sort-based.因此，NV-Shuffle的對比對象如表5所示：

Table 5 Compared System表5 對比對象

3.2 性能測試

圖13展示了Sort和Wordcount在各種對比系統(tǒng)中的執(zhí)行時間以及歸一化執(zhí)行時間(以NV-Shuffle的執(zhí)行時間為基準(zhǔn))，從圖13中可以看出，NV-Shuffle對Shuffle-heavy類型和Shuffle-light類型負載的影響是不同的：

1) Shuffle-heavy類型.相比于其他模式，NV-Shuffle有較為明顯的優(yōu)勢.以Sort為例，圖13(b)顯示了Sort以NV-Shuffle為基準(zhǔn)的執(zhí)行時間比例，分別是1.4,1.52，1.18，1.26.由于Sort負載是Shuffle-heavy類型的負載，輸入的數(shù)據(jù)量約為100 GB，且中間結(jié)果的數(shù)據(jù)量也為100 GB左右，因此在Shuffle階段讀寫數(shù)據(jù)量較大，此時NV-Shuffle的性能優(yōu)勢得以體現(xiàn)；

2) Shuffle-light類型.各種模式的執(zhí)行時間基本上相同.以Wordcount為例，從圖13(b)可以看出，其執(zhí)行時間比例大致都為1.其主要原因在于，Wordcount在Input,Shuffle，Output這3個階段的數(shù)據(jù)量分別是244.5 GB，15 GB，5.8 GB.中間結(jié)果的數(shù)據(jù)量太小，導(dǎo)致發(fā)揮不了NV-Shuffle的性能優(yōu)勢，因此對于Wordcount負載而言，5種對比系統(tǒng)的執(zhí)行時間基本上相同.

為了更加清楚地觀察Sort和Wordcount在5種不同模式下的執(zhí)行時間，我們統(tǒng)計了Sort和Wordcount在Map stage和Reduce stage的執(zhí)行時間，如圖14所示：

Fig. 14 The execution time of different stages圖14 不同負載的Map Stage和Reduce Stage的執(zhí)行時間

圖14展示了Sort和Wordcount在Map Stage和Reduce Stage的執(zhí)行時間.從圖14中也可以看出，Shuffle階段的數(shù)據(jù)量是決定NV-Shuffle性能優(yōu)勢的重要因素.另外，通過比較不同存儲方式和Shuffle模式，我們也可以看出：

1) tmpfs與RAMDisk在性能上的差異.由于tmpfs是內(nèi)存文件系統(tǒng)，而RAMDisk則是使用內(nèi)存來模擬塊設(shè)備，并且在其上格式化一個普通的文件系統(tǒng)，例如在本次實驗中使用的Ext4，因此在性能上，本身tmpfs是要優(yōu)于RAMDisk.由于Shuffle數(shù)據(jù)量上的差異，Sort的差異要比Wordcount的差異更為明顯.

2) Hash-based Shuffle與Sort-based Shuffle在性能上的差異.在數(shù)據(jù)規(guī)模與Reduce Task個數(shù)不是特別多的場景下，Hash-based模式的性能還是要優(yōu)于Sort-based模式，但是當(dāng)Reduce Task個數(shù)較大時，Hash-based模式的局限性就會體現(xiàn)出來：小文件對系統(tǒng)性能的影響.我們使用RAMDisk模式測試了不同Reduce Task個數(shù)對于Sort負載的影響，其結(jié)果如圖15所示.

Fig. 15 The executing time of different configuration圖15 不同配置下的執(zhí)行時間

圖15展示了在不同Reduce Task個數(shù)的場景下，Sort執(zhí)行時間的變化情況.從圖15中可以看出，當(dāng)Reduce Task設(shè)置得較小時，Hash-based模式在性能上還是存在優(yōu)勢的；當(dāng)Reduce Task個數(shù)較大時，由于本身的局限性，Hash-based的性能不如Sort-based.這也是為什么在原始Spark中，會存在一個參數(shù)“spark.shuffle.sort.bypassMergeThreshold”的原因.當(dāng)然，由于NV-Shuffle摒棄了文件系統(tǒng)的存儲方式，因此小文件問題不會影響NV-Shuffle的性能，從而NV-Shuffle仍舊采用類似于Hash-based的數(shù)據(jù)組織方式.

同樣地，我們使用TPC-H的22條查詢語句對比了上述5種對比系統(tǒng)的執(zhí)行時間，為了能夠更好地展現(xiàn)22條查詢語句的執(zhí)行時間，我們根據(jù)RAMDisk上Sort的執(zhí)行時間進行分類：

1) 執(zhí)行時間T?200 s；

2) 執(zhí)行時間T>200 s；

3) 執(zhí)行時間100 s

4) 執(zhí)行時間T<100 s.

圖16展示了TPC-H中查詢語句的執(zhí)行時間和Shuffle數(shù)據(jù)量，圖16中左坐標(biāo)表示的是查詢的執(zhí)行時間，右坐標(biāo)表示的是該查詢語句執(zhí)行過程中Shuffle的數(shù)據(jù)總量.為了能夠更好地展現(xiàn)NV-Shuffle與其他4種對比對象在TPC-H負載中的差異，我們選擇了每條查詢語句在RAMDisk Hash，tmpfs Hash，RAMDisk Sort，tmpfs Sort這4種方式下執(zhí)行時間最少的方式，與NV-Shuffle進行對比，對比的結(jié)果如圖17所示：

Fig. 16 The results of TPC-H圖16 TPC-H的測試結(jié)果

圖17展示了相比于MinTime(RAMDisk Hash，tmpfs Hash，RAMDisk Sort，tmpfs Sort)，NV-Shuffle降低的百分比.從圖17中可以看出：

Fig. 17 Comparison of NV-Shuffle with Min(RAMDisk Hash, tmpfs Hash, RAMDisk Sort, tmpfs Sort)圖17 NV-Shuffle與Min (RAMDisk Hash，tmpfs Hash，RAMDisk Sort，tmpfs Sort)之間的對比

1) Shuffle數(shù)據(jù)量越大，NV-Shuffle的效果越明顯(10%～25%)，例如Q5,Q8,Q9,Q21等；

2) 由于其Shuffle數(shù)據(jù)量少，NV-Shuffle基本上沒有效果，例如Q1,Q6,Q15,Q19等；

3) 其他Query語句，雖然存在一定量的Shuffle數(shù)據(jù)，但是NV-Shuffle能夠降低的執(zhí)行時間非常有限(3%～8%)，例如Q2,Q3,Q15等.

3.3 Shuffle數(shù)據(jù)的組織方式測試

NV-Shuffle使用了Hash-based Private NV-Buffer的方式，而與之對應(yīng)的是Hash-based Shared NV-Buffer方式，在本節(jié)中，我們主要對比這2種方式的性能差異.

Fig. 18 The relationship between parallelism, the number of partitions and performance overheads圖18 并行度P、partition個數(shù)R、性能開銷O三者之間的關(guān)系

在第2.2節(jié)中，我們提到Hash-based Shared NV-Buffer的2個缺點是并發(fā)寫帶來的鎖開銷和失效恢復(fù)(暫不評價).因此，我們來看并發(fā)寫帶來的鎖開銷對于Hash-based Shared NV-Buffer的影響.影響并發(fā)寫的2個因素：節(jié)點上Task的并行度和partition個數(shù).并行度P、partition個數(shù)R、鎖競爭引發(fā)的開銷O這三者之間的關(guān)系，如圖18所示：

圖18中的橫坐標(biāo)表示的是并行度與partition個數(shù)的比例，縱坐標(biāo)表示的是性能開銷，從圖18中可以看出，當(dāng)PR越小，性能開銷就越小，反之，性能開銷則越大.因此，我們進行了一些對比實驗，該實驗我們使用了一個1+1的Spark集群，且使用了Sort負載進行測試.

圖19展示了Shared與Private在極端場景下的性能對比，即partition的個數(shù)為1，并行度分別為4,8,16,32,64，則PR的比例是逐漸增大.在實驗中，我們統(tǒng)計了Map Task的平均執(zhí)行時間，從圖19中可以看出：1)Private的平均時間都要小于Shared的平均時間，一個主要的原因在于Shared存在鎖競爭的開銷；2)當(dāng)并行度逐漸增大時，例如32,64，無論是Private還是Shared，平均時間都有一個明顯的上升，引起該現(xiàn)象的主要原因在于測試節(jié)點的CPU核數(shù)為12，啟用超線程之后變?yōu)?4，因此當(dāng)并行度為32或者64時，存在CPU資源的競爭，從而導(dǎo)致性能的下降.為了更加清晰地看出不同并行度下的CPU利用率，我們列舉了并行度為8,32時節(jié)點在Map Stage時的CPU利用率，如圖20所示.

Fig. 19 The results when R=1圖19 R=1的性能評測

Fig. 20 The CPU utilization under different parallelism圖20 不同并行度下的CPU利用率

Fig. 21 The results when R=128圖21 R=128的性能評測

從圖20可以看出，當(dāng)并行度超過節(jié)點本身的CPU資源限制時CPU利用率較高，也會導(dǎo)致系統(tǒng)性能的下降.但是在實際的生產(chǎn)環(huán)境中，partition個數(shù)為1的場景并不常見，例如Taobao的負載[33]，其job的平均Reduce個數(shù)為12，而Sailfish[20]使用Yahoo的負載，它的Reduce個數(shù)為1 024～65 536之間.因此我們以R=128為測試場景，其測試結(jié)果如圖21所示.

從圖21可以看出，由于增加了partition個數(shù)，能夠在一定程度上緩解競爭所帶來的性能開銷，即在同一時間內(nèi)，并不是所有的Task都往一個partition上寫，如圖22所示:

Fig. 22 The Shared mode under different partitions圖22 不同partition個數(shù)下的Shared模式

雖然通過增加partition的個數(shù)能夠緩解鎖競爭，但是從圖22的測試結(jié)果看，Private模式的性能要優(yōu)于Shared模式，并且在2.2.1節(jié)中描述了Shared模式在失效處理上的劣勢，因此無論從性能還是錯誤處理的角度，Private模式都要優(yōu)于Shared模式.

3.4 NVM空間分配策略測試

在2.3.2節(jié)中，我們了解到NV-Shuffle malloc對于不同粒度NV-Buffer的開銷較小，并且我們修改了原先流程中對于NV-Buffer的分配機制(預(yù)先分配變?yōu)檠舆t分配)，因此在本節(jié)中，一方面，我們需要觀察不同粒度的NV-Buffer對于負載執(zhí)行時間的影響；另一方面，延遲分配對于系統(tǒng)的影響.

1) 不同粒度的NV-Buffer對于執(zhí)行時間的影響.在該實驗中，我們使用不同粒度的NV-Buffer(4 KB，8 KB，16 KB，32 KB，64 KB)對Sort和Wordcount進行測試，其結(jié)果如圖23所示：

Fig. 23 The impact of the NV-Buffer size on job execution圖23 NV-Buffer粒度對于作業(yè)執(zhí)行的影響

圖23展示了不同粒度對于作業(yè)的執(zhí)行時間的影響，從圖23中可以看出，隨著分配粒度逐漸增大，執(zhí)行時間存在小幅度的降低，但是整體上對于執(zhí)行時間的影響不大，但是分配粒度的選取在一定程度上會影響空間利用率，因此我們將分配粒度的默認值設(shè)置為8 KB.

2) 延遲分配的影響.延遲分配可以避免為那些不存在數(shù)據(jù)的partition進行分配NV-Buffer，從而在一定程度上節(jié)省NVM空間資源.在第2.3節(jié)中，我們統(tǒng)計了不同Reduce個數(shù)下size=0的partition個數(shù)，如圖8所示，即當(dāng)Reduce個數(shù)越大時，其size=0的partition個數(shù)就越多，例如當(dāng)Reduce個數(shù)為2 048時，平均每個Map Task中size=0的partition個數(shù)為419，相比于預(yù)先分配，延遲分配不會為這個419個partition分配NV-Buffer，從而在一定程度上節(jié)省了空間資源，那么與預(yù)先分配對比能夠節(jié)省的空間資源大小，如式(3)所示：

節(jié)省的空間=M×P0×NV-Buffer,

(3)

其中，M表示是Map Task的總個數(shù)，P0表示Map Task中size=0的partition的平均個數(shù)，NV-Buffer則是NV-Shuffle的分配粒度.上例中，通過延遲分配可以節(jié)省大約2.6 GB的NVM空間.

3.5 失效恢復(fù)測試

該部分主要評測NV-Shuffle的Shuffle數(shù)據(jù)恢復(fù)時間與傳統(tǒng)Spark的Shuffle數(shù)據(jù)恢復(fù)時間.我們通過人工手動的方式在“一定的時間”kill集群中某一個節(jié)點上的Executor進程，從而觀察作業(yè)最終完成的時間.其中“一定的時間”包含2個部分：1)在Map Stage階段，以完成Map Task個數(shù)為準(zhǔn)；2)在Reduce Stage階段，以完成Reduce Task個數(shù)為準(zhǔn).在該實驗中，我們使用2種較為簡單的負載：Sort和Wordcount，其中Map Task的個數(shù)分別為800和2 048，Reduce Task的個數(shù)都設(shè)置為256.對于Spark，我們使用了tmpfs Hash的模式來存儲Shuffle數(shù)據(jù)，從而與NV-Shuffle進行對比.

圖24展示了在Map階段的不同時機進行kill Executor時，Sort與Wordcount的執(zhí)行時間.從圖24中可以看出：1)NV-Shuffle的執(zhí)行時間不受Executor被kill的影響，其主要原因在于由于NV-Shuffle對于NV-Buffer的管理和組織不依賴于Executor ID，因此不需要重新執(zhí)行之前已經(jīng)完成的Map Task；2)對于Spark而言，其對Map階段產(chǎn)生的數(shù)據(jù)的組織與維護是與Executor ID相關(guān)的，因此當(dāng)Executor被kill時，之前所生成的數(shù)據(jù)便無效，需要進行重計算，因此在Map階段越晚進行kill Executor，重計算的Task越多，執(zhí)行時間越長.圖25展示了不同時機進行kill Executor時需要重計算的Task個數(shù).

Fig. 24 The execution time when executor is killed during Map Stage圖24 在Map階段不同類型負載在Executor被kill時的執(zhí)行時間

Fig. 25 The number of recomputing tasks when executor is killed during Map Stage圖25 在Map階段Executor被kill之后重計算Task的個數(shù)

Fig. 26 The execution time when executor is killed during Reduce Stage圖26 在Reduce階段不同類型負載在Executor被kill時的執(zhí)行時間

圖26展示了在Reduce階段的不同時機進行kill Executor時，Sort與Wordcount的執(zhí)行時間.從圖26可以看出：

1) NV-Shuffle的執(zhí)行時間不受Executor被kill的影響，其主要原因在于由于NV-Shuffle對于NV-Buffer的管理和組織不依賴于Executor ID，因此當(dāng)Executor被kill時，Reduce Task還是能夠通過Driver找回數(shù)據(jù)進行拉??；

2) 當(dāng)Executor被kill后，重啟的Executor的ID已經(jīng)發(fā)生改變，Reduce Task無法通過新的Executor ID去獲取數(shù)據(jù)，Spark需要重新計算以產(chǎn)生相應(yīng)的數(shù)據(jù)，因此比正常Sort和Wordcount的執(zhí)行時間分別超出了約36.8%和27.3%；

3) 不同kill時機對NV-Shuffle和Spark的作業(yè)執(zhí)行時間都沒有產(chǎn)生影響：①NV-Shuffle由于不需要進行重計算就能夠找回數(shù)據(jù)，因此無論何時Executor被kill都不會影響最終的執(zhí)行時間；②對于Spark而言，kill集群中某個節(jié)點上的Executor，則它所生成的數(shù)據(jù)就要通過重計算重新獲取，即所有在該節(jié)點上執(zhí)行完成的Map Task都要重新執(zhí)行一遍，鑒于當(dāng)前數(shù)據(jù)均勻分布在集群上，則需要重新執(zhí)行的Map Task的個數(shù)分別約為200(Sort)和512(Wordcount)(如圖27所示).等Map Task都執(zhí)行完成之后Reduce Stage則繼續(xù)執(zhí)行剩余沒有完成的Reduce Task，因此不同的kill時機不會影響作業(yè)的執(zhí)行時間，增加的時間只是重新執(zhí)行Map Task的時間.

Fig. 27 The number of recomputing tasks when executor is killed during Reduce Stage圖27 在Reduce階段Executor被kill之后重計算Task的個數(shù)

4 總 結(jié)

Shuffle階段由于持久化所帶來的IO開銷可能大大延長數(shù)據(jù)處理的時間，NVM憑借其讀寫速度快、非易失性以及高密度性等諸多優(yōu)點，為改變大數(shù)據(jù)處理過程中對持久化IO的依賴、克服目前基于內(nèi)存計算的大數(shù)據(jù)處理中的IO性能瓶頸提供了新機會.鑒于上述2點，我們提出了NV-Shuffle，一種基于NVM的Shuffle優(yōu)化策略.NV-Shuffle摒棄了以文件系統(tǒng)進行Shuffle數(shù)據(jù)存取的方式，而是采用持久化內(nèi)存的方式，直接在用戶態(tài)訪問持久化內(nèi)存，避免了傳統(tǒng)存儲系統(tǒng)中的冗長的 IO 路徑帶來的額外開銷.為此我們利用pVM構(gòu)建Java的持久化內(nèi)存訪問接口，并針對傳統(tǒng)大數(shù)據(jù)平臺中的Shuffle階段，提出了基于NVM的Shuffle數(shù)據(jù)組織方式、NVM空間的分配策略以及NV-Buffer的組織管理方式.我們在Spark上實現(xiàn)了NV-Shuffle，并進行了詳細的性能評測與分析.實驗結(jié)果顯示，對于Shuffle-heavy類型的負載，在用內(nèi)存模擬的NVM上，相比于原始的Shuffle實現(xiàn)，NV-Shuffle可節(jié)省10%～40%的執(zhí)行時間.而且，在有節(jié)點上的Executor失效的情況下，NV-Shuffle可節(jié)省時間的百分比根據(jù)失效時間點的不同而不同，具體表現(xiàn)為：在Map階段，失效越晚，節(jié)省時間的百分比越大；在Reduce階段，節(jié)省時間的百分比與失效時間點無關(guān).

[1]Micron. Micron sdram[EB/OL]. [2017-09-29]. https://www.micron.com/products/dram/sdram

[2]AgigA Tech. Agigaram?ddr2 nvdimm[EB/OL]. [2017-09-29]. http://www.agigatech.com/ddr2.php

[3]AgigA Tech. Agigaram?ddr3 nvdimm[EB/OL]. [2017-09-29]. http://www.agigatech.com/ddr3.php

[4]AgigA Tech. Agigaram?ddr4 nvdimm[EB/OL]. [2017-09-29]. http://www.agigatech.com/ddr4.php

[5]Intel. Intel-micron memory 3d xpoint[EB/OL]. [2017-09-29]. http://intel.ly/1eicr0a,

[6]Guo Y, Rao J, Zhou X. ishuffle: Improving Hadoop performance with shuffle-on-write[C] //Proc of the 10th Int Conf on Autonomic Computing (ICAC). Berkeley, CA: USENIX, 2013: 107-117

[7]Rao S, Ramakrishnan R, Silberstein A, et al. Sailfish: A framework for large scale data processing[C] //Proc of the 3rd ACM Symp on Cloud Computing (SoCC). New York: ACM, 2012: 4-16

[8]Wang Y, Que X, Yu W, et al. Hadoop acceleration through network levitated merge[C] //Proc of the 2011 Int Conf for High Performance Computing, Networking, Storage and Analysis (SC). New York: ACM, 2011: 57-70

[9]Rahman M, Islam N S, Lu X, et al. High-performance RDMA-based design of Hadoop MapReduce over infiniband[C] //Proc of the 27th IEEE Int Parallel and Distributed Processing Symp Workshops and PhD Forum (IPDPSW). Piscataway, NJ: IEEE, 2013: 1908-1917

[10]Ruan X, Chen H. Improving shuffle I/O performance for big data processing using hybrid storage[C] //Proc of the 2017 Int Conf on Computing, Networking and Communications (ICNC). Piscataway, NJ: IEEE, 2017: 476-480

[11]Hong J, Li L, Han C, et al. Optimizing Hadoop framework for solid state drives[C] //Proc of the 2016 IEEE Int Congress on Big Data (BigData Congress). Piscataway, NJ: IEEE, 2016: 9-17

[12]Arulraj J, Pavlo A, Dulloor S R. Let’s talk about storage and recovery methods for non-volatile memory database systems[C] //Proc of the 2015 ACM SIGMOD Int Conf on Management of Data. New York: ACM, 2015: 707-722

[13]Coburn J, Caulfield A M, Akel A, et al. NV-heaps: Making persistent objects fast and safe with next-generation, non-volatile memories[C] //Proc of the 16th Int Conf on Architectural Support for Programming Languages and Operating Systems (ASPLOS XVI). New York: ACM, 2011: 105-118

[14]Dulloor S R, Kumar S, Keshavamurthy A, et al. System software for persistent memory[C] //Proc of the 9th European Conf on Computer Systems. New York: ACM, 2014: 15-26

[15]Volos H, Tack A J, Swift M M. Mnemosyne: Lightweight persistent memory[C] //Proc of the 16th Int Conf on Architectural Support for Programming Languages and Operating Systems (ASPLOS XVI). New York: ACM, 2011: 91-104

[16]TinyURL. Snappy compession[EB/OL]. [2017-09-29]. http://tinyurl.com/ku899co

[17]Kannan S, Gavrilovska A, Schwan K. pVM: Persistent virtual memory for efficient capacity scaling and object storage[C] //Proc of the 11th European Conf on Computer Systems. New York: ACM, 2016: 1-14

[18]Rasmussen A, Conley M, Porter G, et al. Themis: An I/O-efficient MapReduce[C] //Proc of the 3rd ACM Symp on Cloud Computing (SoCC). New York: ACM, 2012: No.13

[19]Elmeleegy K, Olston C, Reed B. SpongeFiles: Mitigating data skew in MapReduce using distributed memory[C] //Proc of the 2014 ACM SIGMOD Int Conf on Management of Data. New York: ACM, 2014: 551-562

[20]Rao S, Ramakrishnan R, Silberstein A, et al. Sailfish: A framework for large scale data processing[C] //Proc of the 3rd ACM Symp on Cloud Computing (SoCC). New York: ACM, 2012: No.4

[21]Wang Y, Que X, Yu W, et al. Hadoop acceleration through network levitated merge[C] //Proc of the 2011 Int Conf for High Performance Computing, Networking, Storage and Analysis (SC). New York: ACM, 2011: 57-70

[22]Rahman M, Islam N S, Lu X, et al. High-performance RDMA-based design of Hadoop MapReduce over infiniband[C] //Proc of the 27th IEEE Int Parallel and Distributed Processing Symp Workshops and PhD Forum (IPDPSW). Piscataway, NJ: IEEE, 2013: 1908-1917

[23] RDMA. Introduction to Remote Direct Memory Access [EB/OL]. [2017-09-29]. http://www.rdmamojo.com/2014/03/31/remote-direct-memory-access-rdma

[24]IBTA. Infiniband Trade Association[EB/OL]. [2017-09-29]. http://www.infinibandta.org

[25]Condit J, Nightingale E B, Frost C, et al. Better I/O through byte-addressable, persistent memory[C] //Proc of the 22nd ACM SIGOPS Sym on Operating Systems Principles. New York: ACM, 2009: 133-146

[26]Wu X, Reddy A. SCMFS: A file system for storage class memory[C] //Proc of 2011 Int Conf for High Performance Computing, Networking, Storage and Analysis. New York: ACM, 2011: 39-51

[27]Dulloor S R, Kumar S, Keshavamurthy A, et al. System software for persistent memory[C] //Proc of the 9th European Conf on Computer Systems. New York: ACM, 2014: 15-26

[28]GitHub. PMemLibrary[EB/OL]. [2017-09-29]. https://github.com/pmem/linux-examples

[29]Moraru I, Andersen D G, Kaminsky M, et al. Consistent, durable, and safe memory management for byte-addressable non-volatile main memory[C] //Proc of the 1st ACM SIGOPS Conf on Timely Results in Operating Systems. New York: ACM, 2013: 1-13

[30]Yang J, Wei Q, Chen C, et al. NV-tree: Reducing consistency cost for NVM-based single level systems[C] //Proc of the 13th USENIX Conf on File and Storage Technologies (FAST). Berkeley, CA: USENIX, 2015: 167-181

[31]Venkataraman S, Tolia N, Ranganathan P, et al. Consistent and durable data structures for non-volatile byte-addressable memory[C] //Proc of the 9th USENIX Conf on File and Storage Technologies (FAST). Berkeley, CA: USENIX, 2011: 61-75

[32]TPC. TPC BenchmarkTMH: Standard Specification, Revision 2.17.3[S]. San Francisco: TPC, 2017

[33]Ren Z, Xu X, Wan J, et al. Workload characterization on a production Hadoop cluster: A case study on Taobao [C] //Proc of 2012 IEEE Int Symp on Workload Characterization (IISWC). Piscataway, NJ: IEEE, 2012: 1-11