陳曉　朱志祥　梁小江

摘 要：海量數(shù)據(jù)的實時處理不僅要求計算框架快速高效，同時要求流處理過程中產(chǎn)生的中間數(shù)據(jù)的存儲過程同樣高效，因此，可通過提高Spark Streaming對中間結(jié)果數(shù)據(jù)的處理速度來提升流處理效率。為提高Spark Streaming處理中間結(jié)果的效率，文中選擇HBase作為中間數(shù)據(jù)存儲系統(tǒng)，并通過分析Spark Streaming的架構(gòu)及HBase的存儲原理，給出了Spark Streaming向HBase寫入數(shù)據(jù)的方法并進行優(yōu)化。通過對Spark Streaming存儲過程的優(yōu)化，可以一定程度上提高實時數(shù)據(jù)的流處理效率。

關(guān)鍵詞：Spark Streaming；HBase；大數(shù)據(jù)；內(nèi)存計算；流處理

中圖分類號：TP274.2 文獻標識碼：A 文章編號：2095-1302（2016）04-00-03

0 引 言

隨著移動互聯(lián)網(wǎng)、社交網(wǎng)絡(luò)等領(lǐng)域的快速發(fā)展，數(shù)據(jù)量呈指數(shù)式增長，大數(shù)據(jù)時代全面來臨。在這個高速發(fā)展的時代，數(shù)據(jù)的變化速度也越來越快，對數(shù)據(jù)處理和響應(yīng)時間的要求也更加苛刻，數(shù)據(jù)的實時分析和流式處理變得尤為重要。例如，在移動通信領(lǐng)域，對海量數(shù)據(jù)進行實時的挖掘分析，可以準確識別類似于詐騙的電信請求，從而有效避免電信詐騙的發(fā)生。再比如，通過對移動人口數(shù)據(jù)的實時挖掘分析，快速預(yù)測可能的突發(fā)事件。Spark Streaming是建立在Spark上的實時計算框架，擁有基于內(nèi)存的高速執(zhí)行引擎，并且提供豐富的接口和API，被廣泛用于實時數(shù)據(jù)流的分析處理。

對海量實時數(shù)據(jù)進行處理，必然會產(chǎn)生大量的中間數(shù)據(jù)，如何高效存儲Spark Streaming處理過程中產(chǎn)生的數(shù)據(jù)也是大數(shù)據(jù)處理過程中常見的問題。HBase的LSM樹型存儲結(jié)構(gòu)使其具有實時讀寫數(shù)據(jù)的功能。使用HBase作為Spark Streaming中間數(shù)據(jù)的存儲數(shù)據(jù)庫可大大提高數(shù)據(jù)處理的效率。

本文基于實時數(shù)據(jù)的流式處理過程，給出了Spark Streaming將數(shù)據(jù)寫入HBase的具體方法，并在此基礎(chǔ)上進行了優(yōu)化。

1 Spark Streaming簡介

Spark Streaming是Spark生態(tài)系統(tǒng)的重要組成部分，主要用于實時數(shù)據(jù)流的處理。Spark Streaming的工作原理是將流式計算分解成一系列短小的批處理作業(yè)，本質(zhì)上也是數(shù)據(jù)的批量處理，只是將時間跨度控制在數(shù)十毫秒到數(shù)秒之間。這里批處理的引擎依然為Spark，Spark Streaming將輸入數(shù)據(jù)按照批量大?。ù颂幹笗r間跨度如1秒）分成一段一段的數(shù)據(jù)（Discretized Stream），然后每一段數(shù)據(jù)都會轉(zhuǎn)換成Spark中的彈性數(shù)據(jù)集（Resilient Distributed Dataset，RDD），最后將Spark Streaming中對DStream的具體操作都轉(zhuǎn)換成Spark中對RDD的操作，并將中間結(jié)果暫存在內(nèi)存中。整個流式數(shù)據(jù)處理任務(wù)可以根據(jù)需求對中間數(shù)據(jù)加以利用，比如疊加，或者將結(jié)果存儲到外部設(shè)備，例如文件系統(tǒng)HDFS，或者外部數(shù)據(jù)庫Hive，HBase。

HBase – Hadoop Database是Apache Hadoop的一個子項目，是一個高可靠性、高性能、面向列、可伸縮的分布式存儲系統(tǒng)。HBase采用LSM樹的存儲結(jié)構(gòu)，這種結(jié)構(gòu)的核心在于每一次執(zhí)行插入操作時數(shù)據(jù)都會先進入MemStore（內(nèi)存緩沖區(qū)），當MemStore達到上限時，HBase會將內(nèi)存中的數(shù)據(jù)輸出為有序的StoreFile文件數(shù)據(jù)。而在HBase中數(shù)據(jù)列是由列簇來組織的，所以每一個列簇都會有對應(yīng)的一個數(shù)據(jù)結(jié)構(gòu)，HBase將列簇的存儲數(shù)據(jù)結(jié)構(gòu)抽象為Store，一個Store代表一個列簇。這樣在Store中會形成很多個小的StoreFile，當這些小的File數(shù)量達到閾值時，HBase會用一個線程來把這些小File合并成一個大File。這樣，HBase就把效率低下的文件中的插入、移動操作轉(zhuǎn)變成了單純的文件輸出、合并操作。從而使HBase的讀寫數(shù)據(jù)速度非?？?，能夠支持實時讀寫。所以在對海量實時數(shù)據(jù)進行處理時通常使用HBase作為數(shù)據(jù)存儲系統(tǒng)。

2 Spark Streaming寫入數(shù)據(jù)到HBase

2.1 實現(xiàn)方法

Spark Streaming向HBase寫入數(shù)據(jù)時需要對每一條數(shù)據(jù)執(zhí)行插入操作，通常會采用輸出方法foreachRDD（func），將func（此處指將數(shù)據(jù)插入HBase表格）作用于DStream的每一個RDD。

在上述代碼中，countBase為待處理的DStream，首先對countBase進行foreachRDD操作，然后對每個RDD進行操作。此處依據(jù)項目需求對每個RDD進行非空判斷，然后對每個RDD執(zhí)行foreach操作，進而對RDD的每條數(shù)據(jù)record調(diào)用writeToHBase方法，實現(xiàn)數(shù)據(jù)寫入HBase表格。其中zkQuorum為HBase的zookeeper服務(wù)的主機名配置信息，row為HBase表的行鍵，family為表的列簇，key為表的列，value為列的值。writeToHBase方法為自定義的將數(shù)據(jù)寫入HBase的方法。

由上述代碼可看出，在向外部HBase數(shù)據(jù)庫寫數(shù)據(jù)時，通常要先創(chuàng)建與數(shù)據(jù)庫的連接，并獲取HTable實例，其中HTable為操作HBase表格的接口，通過HTable對象對HBase表格中的數(shù)據(jù)進行增，刪，查詢等操作。對RDD的每條數(shù)據(jù)調(diào)用writeToHBase進行寫入操作之前先對setTable對象進行序列化，即對每條數(shù)據(jù)都創(chuàng)建了連接，獲取HTable實例的confTable方法具體代碼如下，其中tableName為建立連接的表格名稱。

2.2 優(yōu)化方法

方法一成功將DStream數(shù)據(jù)寫入HBase數(shù)據(jù)庫，但是資源開銷較大。Sparking Streaming在向HBase寫入數(shù)據(jù)時，必須給每條數(shù)據(jù)都創(chuàng)建一次連接，獲取一個HTable實例，但是創(chuàng)建連接是一項非常耗時的操作，通常耗時數(shù)秒才能完成。在資源高度緊張的環(huán)境下，每秒都有幾千個請求，為每條數(shù)據(jù)單獨創(chuàng)建HTable實例是非常消耗資源的?；诖颂岢鰞?yōu)化方法，減少建立連接與創(chuàng)建HTable實例的次數(shù)，從而降低資源消耗，提高數(shù)據(jù)寫入HBase表的效率。方法二的代碼如下：

依據(jù)方法二依次對countBase執(zhí)行foreachRDD與foreachPartition操作，為每個分區(qū)創(chuàng)建一個confTable對象。對于RDD一個分區(qū)內(nèi)的所有數(shù)據(jù)，這一個confTable對象是共用的。相比于給RDD中所有數(shù)據(jù)都實例化一個HTable，方法二明顯減少了實例創(chuàng)建次數(shù)，大大提升了Spark Streaming向HBase寫入數(shù)據(jù)的性能。

方法一和方法二的操作流程見圖2和圖3所示。

上圖中方形表示一個RDD，每一個橢圓形代表RDD的一個分區(qū)Partition，分區(qū)里的每個圓形代表RDD的一條數(shù)據(jù)record。上圖展示了DStream經(jīng)過foreachRDD操作后對每個RDD的操作。由圖2和圖3可知，方法一對RDD經(jīng)過foreach操作后對每條數(shù)據(jù)record都要經(jīng)過創(chuàng)建連接然后才能寫入HBase表格。方法二先對RDD進行foreachPartition操作，然后對一個分區(qū)創(chuàng)建一個連接，連接創(chuàng)建后對該分區(qū)foreach每條數(shù)據(jù)record進行寫入操作。對比兩圖可看出優(yōu)化后的方法創(chuàng)建連接的次數(shù)明顯比原始方法少。而對于整個處理任務(wù)來說，建立連接，實例化HTable對象消耗資源過多，所以優(yōu)化后的方法二性能大大提升。

HTable是HBase客戶端，實現(xiàn)對HBase表的增加 （Create）、查詢（Retrieve）、更新（Update）和刪除（Delete）。但是HTable適合對單表操作，對讀或?qū)懖僮鞫疾皇蔷€程安全的。對于寫操作（Put或Delete），如果多線程共享一個HTable實例，寫緩沖區(qū)可能會被破壞。對于讀操作，一些被scan使用的字段同時被多個線程共享，如果此時有Get操作，不能保證數(shù)據(jù)的一致性。而大量數(shù)據(jù)的事實分析通常是多線程運作，為了解決線程安全問題，我們使用HTablePool類創(chuàng)建一個HTable的對象池，讓多個HTable實例共享一個Configuration，使用時通過getTable方法獲取一個HTable對象，然后可以進行各種增加（Create）、刪除（Delete）和更新（Update）等操作，使用完后調(diào)用close（）方法可將HTable對象歸還到池中。方法三代碼如下：

def confTable（zkQuorum：String，tableName：String）： HTable = {

import org.apache.hadoop.fs.Path

val conf = HBaseConfiguration.create（）

conf.addResource（new Path（“/etc/HBase/conf/core-site.xml”））

conf.addResource（new Path（“/etc/HBase/conf/HBase-site.xml”））

conf.set（“HBase.zookeeper.quorum”，zkQuorum）

val pool = new HTablePool（conf， SIZE）；

val hTable = pool.getTable（tableName）；

return hTable

}

由于HTablePool僅僅作為HTable的連接池，里面維護的HTable使用的Configuration是同一個，所以本質(zhì)上所有的HTable共用同一個HConnection。而對于創(chuàng)建一個連接，HConnection的創(chuàng)建所損耗的資源遠遠多于創(chuàng)建一個HTable。既然HTable的創(chuàng)建是輕量級的，那么共享一個HConnection的HTablePool實際價值就不大。只要保證HConnection實例是唯一的，全局共享的，然后在每次操作HBase表時根據(jù)HConnection對象來重新創(chuàng)建，使用完成之后及時關(guān)閉即可。

最簡單的創(chuàng)建HConnection實例的方式是HConnectionManager.createConnection（）。HConnectionManager是一個不可實例化的類，專門用于創(chuàng)建HConnection。該方法創(chuàng)建了一個連接到集群的HConnection實例，該實例被創(chuàng)建的程序管理。通過這個HConnection實例，可以使用HConnection.getTable（byte[]）方法取得HTableInterface implementations的實現(xiàn)。方法四的代碼如下：

def confTable（zkQuorum：String，tableName：String）： HTableInterface = {

import org.apache.hadoop.fs.Path

val conf = HBaseConfiguration.create（）

conf.addResource（new Path（“/etc/HBase/conf/core-site.xml”））

conf.addResource（new Path（“/etc/HBase/conf/HBase-site.xml”））

conf.set（“HBase.zookeeper.quorum”，zkQuorum）

val connection = HConnectionManager.createConnection（conf）；

val table = connection.getTable（“tablename”）；

return table

}

3 結(jié) 語

本文先給出了海量數(shù)據(jù)的實時流處理過程中將數(shù)據(jù)存儲到HBabe中的方法。通過Spark Streaming直接將中間數(shù)據(jù)實時寫入HBase表格，在解決中間數(shù)據(jù)存儲問題的同時，確保了流處理過程的高效率，并進一步優(yōu)化此方法，將數(shù)據(jù)的處理速度大大提高，更大程度的提升流處理的效率。

參考文獻

[1] Spark[EB/OL].http：//spark.apache.org//. 2015

[2] Apache HBase[EB/OL]. http：//HBase.apache.org/. 2015

[3] 卓海藝.基于HBase的海量數(shù)據(jù)實時查詢系統(tǒng)設(shè)計與實現(xiàn)[D].北京：北京郵電大學(xué)， 2013.

[4] 張榆，馬友忠，孟小峰.一種基于HBase的高效空間關(guān)鍵字查詢策略[J].小型微型計算機系統(tǒng)， 2012，33（10）：2141-2146.

[5] 夏俊鸞，邵賽賽.Spark Streaming： 大規(guī)模流式數(shù)據(jù)處理的新貴[EB/OL]. http：//www.csdn.net/article/2014-01-28/2818282-Spark-Streaming-big-data. 2014.