一種基于Spark大數(shù)據(jù)處理平臺的查詢方法

張海峰，魏可欣

(1.中通服咨詢設(shè)計研究院有限公司，江蘇南京 210019 2.南京大學 商學院，江蘇南京 210093 3.蘇州大學政治與公共管理學院，江蘇蘇州 215123)

隨著互聯(lián)網(wǎng)的發(fā)展，數(shù)以萬計的網(wǎng)頁不斷涌現(xiàn)，而要搜索這些網(wǎng)頁首先要抓取存儲，然后進行分析計算。但是日益龐大的數(shù)據(jù)使得存儲面臨著單臺機器容量不夠和查詢面臨著耗時太多的問題［1－3］。針對這兩個問題，Google提出了分布式存儲和分布式并行計算的解決方案，而后來的Hadoop產(chǎn)品是這種解決方案的源碼實現(xiàn)。Hadoop提供了分布式文件系統(tǒng)HDFS和分布式并行計算框架MapReduce，隨著Hadoop的發(fā)展，其生態(tài)系統(tǒng)又不斷涌現(xiàn)新的項目產(chǎn)品，如Hbase、Hive、Pig等，但它們都是基于 HDFS存儲層和MapReduce計算框架，MapReduce通過在集群的多個節(jié)點上并行執(zhí)行計算，因此大大加快了查詢的速度，但隨著數(shù)據(jù)量的日益增大，MapReduce漸漸顯得力不從心，于是基于內(nèi)存計算的Spark計算框架應運而生，Spark的查詢速度相比Hadoop提升了100倍，因此是目前最先進的分布式并行計算框架［4－5］。 隨著 Spark生態(tài)系統(tǒng)的發(fā)展，在其上又涌現(xiàn)出 Spark SQL、Spark Streaming、MLlib、GraphX 等，其中Spark SQL是針對SQL用戶開發(fā)的可以用SQL語言對結(jié)構(gòu)化數(shù)據(jù)進行分析查詢的工具［6－8］。

目前主流的大數(shù)據(jù)分析系統(tǒng)主要是Hadoop和Spark，更準確地說，Spark只是一個并行計算框架，而hadoop中包含計算框架MapReduce和分布式文件系統(tǒng)HDFS。Hadoop更廣泛地說還包括在其生態(tài)系統(tǒng)上的其他系統(tǒng)，如Hbase、Hive等；而Spark生態(tài)系統(tǒng)已經(jīng)發(fā)展成為一個包含多個子項目的集合，其中包含 Spark SQL、Spark Streaming、GraphX、MLlib等子項目，Spark是MapReduce的替代方案，性能提升近百倍，Spark SQL是Spark之上提供結(jié)構(gòu)化數(shù)據(jù)SQL查詢與分析的查詢引擎，是MapReduce之上的Hive 的替代方案［9－14］。 但無論是 MapReduce 還是Spark，針對大規(guī)模數(shù)據(jù)查詢時，從開始執(zhí)行查詢到輸出結(jié)果有較長的延遲，超出了用戶的容忍程度，且對查詢結(jié)果的數(shù)據(jù)量有限制，不支持較大規(guī)模的查詢結(jié)果輸出［15－16］。 而 Spark SQL 作為 Spark 之上廣泛使用的SQL查詢引擎，其自身也存在內(nèi)存資源浪費和影響性能的問題［17－19］。

1 基于多任務并行計算的快速查詢方法

Spark執(zhí)行大數(shù)據(jù)分析時是通過分配多個任務并行計算來達到快速查詢的目的，每個任務的計算結(jié)果都是最終結(jié)果的一個子集，但Spark需要等到所有任務均執(zhí)行結(jié)束后才輸出結(jié)果，極大地延遲了客戶響應時間，而Spark SQL作為Spark的一個用SQL語言執(zhí)行結(jié)構(gòu)化數(shù)據(jù)查詢的模塊，自然也存在客戶響應延遲的問題［20－22］。 一方面，Spark 核心在內(nèi)存中存儲了整個查詢的計算結(jié)果，剩余的堆?？臻g和內(nèi)存完全限制了其所能存儲的結(jié)果數(shù)據(jù)，因此完全不支持計算結(jié)果較龐大的查詢場景［23］。另一方面，若結(jié)果數(shù)據(jù)限制值配置不當，一旦查詢結(jié)果超量，將造成 Spark應用內(nèi)存溢出崩潰［24］。同時，Spark SQL獲取Spark核心的計算結(jié)果后，要進行一系列格式轉(zhuǎn)化，如逗號替換為制表符，還要將結(jié)果復制到另一個容器中才能輸出，無論是轉(zhuǎn)化還是復制，都將造成數(shù)據(jù)在內(nèi)存中有多個備份，對于查詢結(jié)果較大的情形，極大地浪費了資源，另外在一定程度上影響了性能［25－29］。 具體分析如下。

Spark系統(tǒng)從接收用戶SQL語句到輸出結(jié)果給用戶的整個過程大致可以分為5個步驟［30－33］：

（1）Spark SQL模塊接收用戶的SQL語句后，進行語法解析、執(zhí)行策略優(yōu)化、執(zhí)行Job生成，最后通過調(diào)用Spark核心中的SparkContext的接口進行Job的提交；

（2）SparkContext收到Job后，定義Task執(zhí)行成功后如何存儲計算結(jié)果，然后提交 Job給eventProcessActor，接著等待 eventProcessActor告知Job執(zhí)行結(jié)束，結(jié)束后將計算結(jié)果返回給Spark SQL；

（3）eventProcessActor收到提交job的事件后，在各個節(jié)點分配多個Task開始并行計算；

（4）每個Task執(zhí)行完畢后向eventProcessActor報告狀態(tài)和結(jié)果，eventProcessActor統(tǒng)計Job的所有Task是否全部完成，若完成，則通知SparkContext提交的 Job已結(jié)束，SparkContext返回計算結(jié)果給Spark SQL；

（5）Spark SQL得到計算結(jié)果后，先進行格式轉(zhuǎn)化，然后拷貝一份給輸出程序，最后由輸出程序輸出結(jié)果，如圖1所示。

圖1 Spark SQL執(zhí)行查詢的架構(gòu)圖

步驟1主要是解析SQL語句的語法并生成一組代表一個Job的RDD，RDD是一種分布式數(shù)據(jù)結(jié)構(gòu)，它描述了要處理的分布式存儲的數(shù)據(jù)以及怎樣處理的算法，因此一個RDD就代表對數(shù)據(jù)的一個操作，一組RDD就是一個操作序列，按序完成了這一系列操作后即代表完成了一次查詢計算；Spark采用了延遲執(zhí)行策略［16］，即每個操作不先執(zhí)行，而是先生成操作的序列，然后把這個序列發(fā)送給執(zhí)行器執(zhí)行；這一組RDD所代表的操作因為有序且不循環(huán)，因此其組成的邏輯依賴圖又稱有向無環(huán)圖（DAG）；在DAG中，下游的RDD是上游的RDD執(zhí)行某個操作后生成的，如圖2所示。

圖2 Spark SQL生成DAG框架圖

步驟2主要是將DAG提交給處于另一個線程環(huán)境的eventProcessActor，提交前，分配一塊內(nèi)存，并告知eventProcessActor當Task執(zhí)行成功后往這塊內(nèi)存中存儲結(jié)果，提交后，當前線程掛起，等待eventProcessActor在Job完成后喚醒它，被喚醒后，此時所有Task已經(jīng)執(zhí)行結(jié)束，并且計算結(jié)果全部已經(jīng)存儲到事先分配的內(nèi)存中［17］；因此直接將這塊存儲了結(jié)果的內(nèi)存地址返回給Spark SQL模塊。由于要等到所有Task執(zhí)行結(jié)束才能返回結(jié)果，因此客戶響應時間過長，事實上，每個Task的結(jié)果都是最終結(jié)果的一個子集，沒有必要一起輸出所有的結(jié)果子集；另外，由于輸出前要存儲整個查詢結(jié)果，結(jié)果的大小將直接受限于程序的堆棧大小，如圖3所示。

圖3 SparkContext提交Job流程圖

步驟3主要是實現(xiàn)DAG階段的劃分以及每個階段Task集合的生成。每個階段的所有Task執(zhí)行的都是相同的操作，只不過它們作用的數(shù)據(jù)不同罷了，因此它們可以完全并行執(zhí)行；但是不同階段的Task就不一定能并行了。如圖4所示，每個深灰色填充矩形代表一個數(shù)據(jù)塊，并且每個數(shù)據(jù)塊都對應一個Task對其進行計算，由于RDD2的數(shù)據(jù)塊是根據(jù)RDD1的多個數(shù)據(jù)塊計算得來的，因此就需要等待執(zhí)行RDD1的所有Task結(jié)束才能開始計算RDD2，所以RDD1和RDD2需要分屬兩個不同的階段，而RDD2計算出RDD5時，每個數(shù)據(jù)塊都是獨立進行的，互不依賴，RDD2中計算其中一個數(shù)據(jù)塊的Task不需要等待其他數(shù)據(jù)塊的Task結(jié)束即可開始向RDD5生成的計算（此處是 join操作），因此RDD2和RDD5可以同屬一個階段；同理，RDD3和RDD4可以同屬一個階段，但RDD4不能和RDD5同屬一個階段；圖中，Stage1和Stage2互不依賴，可以并且執(zhí)行，Stage3同時依賴Stage1和Stage2，因此必須等待Stage1和Stage2均完成后才能執(zhí)行；DAG的整體執(zhí)行過程如圖5所示。

圖4 RDD階段劃分示意圖

圖5 DAG按階段執(zhí)行過程圖

步驟4主要是最后一個階段的Task執(zhí)行成功后，將計算結(jié)果存儲到SparkContext指定的內(nèi)存中；在圖5中，Stage1和Stage2的Task執(zhí)行結(jié)束后只生成中間結(jié)果，Stage3的每個Task才是最終結(jié)果，而最終輸出的結(jié)果是由Stage3的每個Task的結(jié)果拼接而成，在拼接的過程中可能會有排序。如圖6所示，如果查詢語句要求對結(jié)果進行排序，則Task的結(jié)果按序存放，如果不對結(jié)果排序，則結(jié)果按照Task完成的先后順序排序，每次查詢的結(jié)果排列順序?qū)⑹请S機的。對于結(jié)果排序的情況，既然每個Task知道它的結(jié)果應該排在什么位置，那么應該排在首位的Task就已經(jīng)計算出了最終結(jié)果的頭部，可以立即通知客戶了；對于結(jié)果不排序的情況，由于客戶不關(guān)心結(jié)果的排列順序，因此不管哪個Task先計算完成，其結(jié)果就可以告知客戶，沒有必要去等到其他Task，而且即使等待，最終的結(jié)果也是按照它們執(zhí)行完成的先后順序排序的。

圖6 排序查詢Task存儲計算結(jié)果

步驟5主要是對記錄行數(shù)組形式的結(jié)果格式化為字符串序列，每一行記錄轉(zhuǎn)換成字符串格式，并且將列分隔符替換為制表符，最后輸出程序在提取格式化后的結(jié)果時，復制一份到輸出程序，然后輸出。事實上，對結(jié)果的格式化不是必須的，格式化可能看起來更美觀，但是卻消耗了大量內(nèi)存和性能，在某些情況下，數(shù)據(jù)本身已經(jīng)很工整了，此時就沒必要去格式化。

2 基于Spark大數(shù)據(jù)處理平臺的查詢方法創(chuàng)新

本研究要解決的技術(shù)問題是提供一種基于Spark大數(shù)據(jù)處理平臺的查詢方法，該查詢方法在執(zhí)行常規(guī)的簡單查詢時（DAG階段比較少），即使要處理的數(shù)據(jù)非常龐大，也能快速返回結(jié)果；執(zhí)行復雜的查詢時，能在原有基礎(chǔ)上大大縮短用戶響應時間，無論是執(zhí)行哪種查詢，都試圖實現(xiàn)只要有結(jié)果滿足輸出條件就立即輸出，沒有任何延遲。

2.1 實驗設(shè)計

基于Spark大數(shù)據(jù)處理平臺的查詢方法是，當Spark應用程序向Spark大數(shù)據(jù)處理平臺提交Job時，同時傳遞結(jié)果格式化規(guī)則、結(jié)果輸出規(guī)則以及結(jié)果是否要排序的通知，并且在Spark平臺內(nèi)部根據(jù)傳遞的這些信息設(shè)定Task執(zhí)行成功后的處理策略：

若是排序查詢時，判斷當前Task結(jié)果的排名序號是否是上一次輸出序號的下一位，若是，則根據(jù)Spark應用程序傳遞的結(jié)果格式化規(guī)則和輸出結(jié)果，然后按照排名序號判斷緊挨其后是否有排名連續(xù)的已經(jīng)存儲的其他Task結(jié)果，有則一并輸出這些結(jié)果，已經(jīng)輸出的結(jié)果其占用的內(nèi)存立即釋放；若不是，存儲當前Task結(jié)果到相應的排名序號索引位置上。這樣，只要結(jié)果按照排名序號滿足輸出條件就立即輸出，無任何延遲，最快的情況下，計算首結(jié)果的Task第一個完成，最慢的情況下，計算首結(jié)果的Task最后一個完成，因此平均下來至少縮短一半的響應時間；

若是非排序查詢時，每一個Task成功后立即根據(jù)Spark應用程序傳遞的結(jié)果格式化規(guī)則和輸出結(jié)果，結(jié)果不存儲。這樣，只要有Task執(zhí)行成功，就立即輸出它的結(jié)果，隨著Task集合的不斷完成，結(jié)果也是連續(xù)地輸出，直到輸出最后一個完成的Task，在這種情況下，整個計算過程沒有任何輸出延遲，只要有新的計算結(jié)果就立即輸出；對于大規(guī)模數(shù)據(jù)集的分析來說，任務數(shù)增加了，但每個任務處理的數(shù)據(jù)塊大小沒變，而無論處理的數(shù)據(jù)集多大，都是第一個完成的Task立即輸出結(jié)果，因此成功實現(xiàn)了只要是簡單查詢，哪怕是超大規(guī)模數(shù)據(jù)集，也能快速輸出首結(jié)果。

如果是非排序查詢，Spark大數(shù)據(jù)處理平臺不再申請存儲計算結(jié)果的內(nèi)存，相應地，DAG最后一個階段的Task執(zhí)行成功后直接輸出結(jié)果；如果是排序查詢且Task結(jié)果需要暫時存儲，判斷內(nèi)存是否足夠容納該Task結(jié)果，若內(nèi)存不夠容納，則立即終止當前Job，并通知Spark應用程序查詢結(jié)果超出系統(tǒng)容量，提示客戶增加篩選條件。因此，非排序查詢的情況下，Spark大數(shù)據(jù)處理平臺能夠源源不斷輸出海量的查詢結(jié)果，支持大數(shù)據(jù)量查詢返回的場景；排序查詢的情況下，不會出現(xiàn)查詢結(jié)果過大導致內(nèi)存溢出異常的問題。

Spark SQL應用程序得到計算結(jié)果后，先判斷結(jié)果是否為空，如果為空，不再走輸出流程，如果不為空，根據(jù)配置可以選擇是否格式化，然后走輸出流程。Spark SQL應用程序輸出結(jié)果時，直接引用結(jié)果，不再重新拷貝一份到輸出模塊。Spark SQL應用程序在向Spark大數(shù)據(jù)處理平臺提交Job前，需要預先定義結(jié)果格式化規(guī)則、結(jié)果輸出規(guī)則、結(jié)果是否要排序的通知，并在提交Job時傳遞這些信息，其中結(jié)果格式化規(guī)則根據(jù)配置可以是空。

Spark大數(shù)據(jù)處理平臺所有跟提交Job相關(guān)的接口均重載一份，重載的接口新增結(jié)果格式化規(guī)則、結(jié)果輸出規(guī)則以及結(jié)果是否要排序的通知這3個參數(shù)，最后在正式提交Job前，根據(jù)這3個參數(shù)設(shè)定Task成功后的處理策略；同時Spark SQL應用程序在提交Job時，使用重載的接口。

2.2 實例驗證及實驗分析

（1）實驗1 本分項實驗基于Spark大數(shù)據(jù)處理平臺的查詢方法中結(jié)果無延遲輸出的具體實施方式，如圖7所示。

圖7 查詢結(jié)果無延遲輸出的實現(xiàn)流程圖

Spark大數(shù)據(jù)處理平臺的應用程序編程接口SparkContext提供新的Job提交接口，新接口要求傳遞結(jié)果輸出規(guī)則、結(jié)果格式化規(guī)則、結(jié)果是否要排序的通知。新接口重新定義Task成功時的結(jié)果處理策略，供Task成功事件發(fā)生時執(zhí)行，在該處理策略中，不再是單純地只存儲Task的結(jié)果，而是根據(jù)結(jié)果要排序與否，判斷結(jié)果是否滿足立即輸出條件，若滿足，則直接根據(jù)Spark應用程序定義并傳遞的結(jié)果格式化規(guī)則和輸出規(guī)則對結(jié)果先進行格式化再輸出，結(jié)果輸出后不再存儲；若暫時不滿足輸出條件，則臨時存儲，等到下一個Task成功時再判斷輸出條件是否滿足，若滿足立即輸出并釋放存儲內(nèi)存。

正在開發(fā)的Spark應用程序可以使用該新接口實現(xiàn)結(jié)果無延遲輸出，已有的Spark應用程序仍可以使用原接口正常工作。例如，對于Spark SQL應用程序，由于其處理的是結(jié)構(gòu)化數(shù)據(jù)，因此結(jié)果要格式化為記錄行的數(shù)組形式，又因為要將列分隔符替換為制表符，因此還要將記錄行數(shù)組格式化為字符串數(shù)組，以便利用字符串的字符替換功能進行替換處理，然后格式化為列分隔符為制表符的結(jié)果形式，最后要輸出時，由于Spark SQL應用程序是命令行程序，結(jié)果是直接打印到控制臺上的，因此對于Spark SQL應用程序來說，輸出結(jié)果即是打印結(jié)果到指定的控制臺上；而Spark SQL應用程序在處理SQL語句時，根據(jù)語句中是否包含Order by字句（排序字句）可判斷結(jié)果是否要排序。

以上結(jié)果格式化規(guī)則、結(jié)果輸出規(guī)則、結(jié)果是否要排序三者信息是Spark應用程序特有的，因此需要在應用提交Job時傳遞給Spark大數(shù)據(jù)處理平臺，并最終在Task成功時使用這些信息以便準確及時地輸出結(jié)果。當所有Task均成功結(jié)束后，此時整個Job即成功結(jié)束，而由于所有結(jié)果已在Task成功時全部輸出，因此Job結(jié)束后，Spark應用程序無須再執(zhí)行輸出流程，可以立即進入下次查詢Job的提交。當某個Task執(zhí)行失敗時，此時整個Job宣告失敗結(jié)束，Spark應用程序在得到Job失敗結(jié)束通知后，輸出錯誤信息并等待下一次查詢Job的提交。

（2）實驗2 本分項實驗基于Spark大數(shù)據(jù)處理平臺的查詢方法中Task成功時處理策略的排序查詢處理的具體實施方式，如圖8所示。

圖8 排序查詢處理的實現(xiàn)流程圖

在Task成功時處理策略中，判別當前查詢是否要對結(jié)果排序，若要排序，則開始應用排序查詢處理過程，詳述如下：

判斷當前Task的結(jié)果的排名序號是否是上一次輸出序號的下一位，若是，則按照Spark應用程序傳遞的結(jié)果格式化和輸出規(guī)則先格式化再輸出結(jié)果，然后按照排名序號判斷緊挨其后是否有排名連續(xù)的已經(jīng)存儲的其他Task結(jié)果，有則一并輸出這些結(jié)果，輸出后釋放這些結(jié)果占用的內(nèi)存；若不是，存儲當前Task結(jié)果到相應的排名序號索引位置上。需要說明的是，Task結(jié)果的排名序號是Spark大數(shù)據(jù)處理平臺在執(zhí)行Job的最后一個階段前已由其他階段事先計算好，Job最后一個階段的所有Task分別知道各自結(jié)果的排名序號，因此每個獨立的Task執(zhí)行完畢后，其排名順序已確定了，無須等到所有Task執(zhí)行完畢才能確定。本實驗中提到的Task均是指Job最后一個階段的Task。例如，對于返回結(jié)果的查詢?nèi)绫?所示。

表1 返回結(jié)果查詢

Task1第1個完成，它計算的結(jié)果排第3位，而當前已輸出到第0位（即還沒有輸出），因此不能輸出給客戶，只能先存儲起來，并且存儲到第3個索引位置上；

Task2第2個完成，它計算的結(jié)果排第1位，因此立即輸出，不存儲；接著判斷緊挨其后的索引位置（第2位開始）上是否有排名連續(xù)的結(jié)果，由于第2位索引位置上沒有結(jié)果，不處理；最后更新當前已輸出的排名序號為1；

Task3第3個完成，它計算的結(jié)果排第5位，而當前已輸出到第1位，因此不能輸出給客戶，只能先存儲起來，并且存儲到第5個索引位置上；

Task4第4個完成，它計算的結(jié)果排第4位，而當前已輸出到第1位，因此不能輸出給客戶，只能先存儲起來，并且存儲到第4個索引位置上；

Task5第5個完成，它計算的結(jié)果排第7位，而當前已輸出到第1位，因此不能輸出給客戶，只能先存儲起來，并且存儲到第7個索引位置上；

Task6第6個完成，它計算的結(jié)果排第2位，而當前已輸出到第1位，因此可以立即輸出，然后判斷緊挨其后的索引位置上（第3位開始）是否有排名連續(xù)的結(jié)果，由于第3、4、5位索引位置上均有結(jié)果，因此繼續(xù)輸出這3個排名連續(xù)的結(jié)果，而第7位雖然有結(jié)果，但第6位的結(jié)果還未返回，因此不能輸出；最后釋放第3、4、5位的結(jié)果占用的內(nèi)存并更新當前已輸出的排名序號為5；

Task7最后一個完成，它計算的結(jié)果排第6位，而當前已輸出到第5位，因此可以立即輸出，然后判斷緊挨其后的索引位置上（第7位開始）是否有排名連續(xù)的結(jié)果，由于第7位索引位置上有結(jié)果，因此繼續(xù)輸出第7位的結(jié)果；最后釋放第7位的結(jié)果占用的內(nèi)存并更新當前已輸出的排名序號為7。

至此，所有任務已全部結(jié)束，而結(jié)果也已經(jīng)全部按序輸出。

（3）實驗3 本分項實驗基于Spark大數(shù)據(jù)處理平臺的查詢方法中Task成功時處理策略的非排序查詢處理的具體實施方式，如圖9所示。

圖9 非排序查詢處理的實現(xiàn)流程圖

在Task成功時處理策略中，判別當前查詢是否要對結(jié)果排序，若無須排序，則開始應用非排序查詢處理過程，詳述如下：

先按照Spark應用程序傳遞的結(jié)果格式化規(guī)則對結(jié)果進行格式化，然后按照Spark應用程序傳遞的結(jié)果輸出規(guī)則輸出結(jié)果，至此Task成功時的處理結(jié)束。

所有Task的結(jié)果均不存儲，誰先結(jié)束誰就先輸出，結(jié)束一個輸出一個，直到所有Task結(jié)束，例如，對于如表2所示的查詢。

表2 結(jié)果的查詢

Task1第1個完成，因為對結(jié)果的順序不要求，因此可以立即輸出；其他Task處理過程相同，當所有Task均執(zhí)行結(jié)束后，結(jié)果就已經(jīng)輸出完畢了。

3 結(jié)束語

本研究創(chuàng)新了一種基于Spark大數(shù)據(jù)處理平臺的查詢方法，首先，執(zhí)行常規(guī)的簡單查詢時（DAG階段比較少），即使要處理的數(shù)據(jù)非常龐大，也能快速返回結(jié)果；執(zhí)行復雜的查詢時，能在原有基礎(chǔ)上大大縮短用戶相應時間，無論是執(zhí)行哪種查詢，都試圖實現(xiàn)只要有結(jié)果就立即輸出，沒有任何延遲；其次，非排序查詢的情況下，Spark SQL能夠源源不斷輸出海量的查詢結(jié)果，支持大數(shù)據(jù)量查詢返回的場景；排序查詢的情況下，不會出現(xiàn)查詢結(jié)果過大導致內(nèi)存溢出異常的問題。最后，優(yōu)化Spark SQL對查詢結(jié)果的格式化處理和輸出處理，減少內(nèi)存和性能消耗。