基于Spark的高維數(shù)據(jù)相似性連接

成小海

(天津工業(yè)大學(xué) 計(jì)算機(jī)科學(xué)與軟件學(xué)院，天津 300387)

0 引 言

高維數(shù)據(jù)相似性連接不僅可以用于分類，而且還可以用于預(yù)測(cè)，在文本分類、聚類分析、預(yù)測(cè)分析、模式識(shí)別、圖像處理等領(lǐng)域應(yīng)用廣泛。但高維數(shù)據(jù)相似性連接仍是一個(gè)非常具有挑戰(zhàn)性的工作，主要有以下兩個(gè)原因。首先，數(shù)據(jù)集的規(guī)模非常大(數(shù)百萬(wàn)或數(shù)十億的對(duì)象)；其次，數(shù)據(jù)集的維數(shù)足夠高(數(shù)千或數(shù)萬(wàn))。因此不可能直接對(duì)數(shù)據(jù)集進(jìn)行操作，必須借助有效的方法來(lái)降低數(shù)據(jù)集的維數(shù)，從而進(jìn)行數(shù)據(jù)集間的運(yùn)算[1]，其中用的最多的就是結(jié)合Hadoop框架[2-9]進(jìn)行分布式運(yùn)算。

近年來(lái)，有學(xué)者不斷地對(duì)高維數(shù)據(jù)的相似性連接進(jìn)行了研究和優(yōu)化。例如，戴健等[7]整合MapReduce框架,提出了分布式網(wǎng)格概略化KNN joins(DSGMP-J)和基于MapReduce的voronoi diagram下的KNN joins(VDMP-J)；馬友忠等[8]結(jié)合Hadoop集群的分布式特性和SAX的降維技術(shù)，提出了向量隨機(jī)生成id的方式進(jìn)行SAX轉(zhuǎn)換，雖然可以起到分布式計(jì)算的效果，但在運(yùn)算過(guò)程中由于復(fù)制多份數(shù)據(jù)和相同id向量的重復(fù)計(jì)算導(dǎo)致內(nèi)存消耗較大;劉雪莉等[10]提出了實(shí)體數(shù)據(jù)庫(kù)上相似性連接算法ES-joins，用于解決字符串模糊匹配的相似性連接問(wèn)題；劉艷等[11]利用Δ-tree進(jìn)行高維數(shù)據(jù)相似連接；周健雯等[12]使用R*樹的自相似性連接。

由于Spark具有運(yùn)算速度快、易用性好和通用性強(qiáng)等優(yōu)點(diǎn)，并且可以高效地處理大規(guī)模數(shù)據(jù)集，因此文中采用基于Spark的相似性連接，對(duì)原有基于MapReduce的方法進(jìn)行改進(jìn)，優(yōu)化其中的計(jì)算步驟，同時(shí)結(jié)合Spark強(qiáng)大的RDD算子，以提高計(jì)算速度。

1 相關(guān)工作及背景知識(shí)

1.1 Spark框架

Spark誕生于加州大學(xué)伯克利分校AMP實(shí)驗(yàn)室，是一個(gè)基于大數(shù)據(jù)的分布式并行計(jì)算框架。由于其先進(jìn)的設(shè)計(jì)理念，從2013年成為Apache孵化項(xiàng)目后，先后擴(kuò)展了Spark SQL、Spark Streaming、Mllib、GraphX和SparkR等組件，逐漸成為大數(shù)據(jù)處理一站式解決平臺(tái)。Spark框架提供了多種資源調(diào)度管理，通過(guò)內(nèi)存計(jì)算、有向無(wú)環(huán)圖等機(jī)制保證分布式的快速迭代計(jì)算，同時(shí)引入了RDD的抽象保證數(shù)據(jù)的高容錯(cuò)性。

RDD是一個(gè)分布式的內(nèi)存抽象概念和只讀分區(qū)記錄的集合，這些集合是彈性的，如果數(shù)據(jù)集一部分丟失，則可以根據(jù)父子RDD中的繼承關(guān)系重新進(jìn)行計(jì)算。Spark的創(chuàng)建、轉(zhuǎn)換、控制和行為等操作都是來(lái)自對(duì)RDD的操作。

1.2 與MapReduce的比較

Spark是對(duì)MapReduce分布式大數(shù)據(jù)處理的借鑒，并對(duì)其進(jìn)行了內(nèi)存框架優(yōu)化，使其具有更強(qiáng)的數(shù)據(jù)處理能力。

(1)Spark將中間過(guò)程放進(jìn)內(nèi)存，迭代效率明顯提高。而MapReduce將中間計(jì)算結(jié)果保存在磁盤中，如果多個(gè)Reduce將會(huì)嚴(yán)重影響程序的運(yùn)行速度。Spark支持有向無(wú)環(huán)圖的分布式計(jì)算，中間運(yùn)算結(jié)果保存在內(nèi)存中，使其整體效率明顯提升。

(2)Spark容錯(cuò)性高。Spark使用了RDD操作，可以在計(jì)算時(shí)通過(guò)CheckPoint實(shí)現(xiàn)容錯(cuò)。

(3)Spark在Shuffle時(shí)不是所有情況都進(jìn)行排序，而MapReduce在此階段需要花費(fèi)大量時(shí)間來(lái)排序。

(4)Spark更加通用。Spark提供的數(shù)據(jù)操作類型有很多種，一般較為通用的是轉(zhuǎn)換操作和行動(dòng)操作兩種類型。而Hadoop只提供Map和Reduce兩種操作。

因此，為了讓集群高效工作，可以嘗試使用Spark框架來(lái)處理高維相似性連接問(wèn)題。

結(jié)合SAX技術(shù)來(lái)降低向量的維數(shù)，并通過(guò)計(jì)算SAX表示之間的距離來(lái)提高過(guò)濾效果。這樣可以減少候選對(duì)，節(jié)省計(jì)算成本。

2 問(wèn)題定義

2.1 高維向量的PAA表示

PAA[13]是一種維數(shù)降低技術(shù)，廣泛應(yīng)用于時(shí)間序列處理和軌跡相關(guān)問(wèn)題研究。它就是將原始高維數(shù)據(jù)等間距分割為較低的維數(shù)，利用定義2給出的距離計(jì)算公式計(jì)算出原始向量的近似距離。文中使用的向量是序列無(wú)關(guān)向量，維度的順序不會(huì)影響歐幾里得距離的計(jì)算結(jié)果。在需要的時(shí)候，可以重新排列向量的維度順序，然后用分段聚合近似表示高維向量。

2.2 高維向量的SAX表示

SAX[8,14-16]也是一種維數(shù)降低技術(shù)，其主要是先將原數(shù)據(jù)規(guī)格化，使其服從高斯分布，然后離散化并用PAA表示，最后轉(zhuǎn)化為字符串。該方法具體的變換步驟舉例如下：

(1)將原始時(shí)間序列R={R1,R2,…,Rn}規(guī)格化為均值為0、標(biāo)準(zhǔn)差為1的標(biāo)準(zhǔn)序列；

(2)將其離散化為x個(gè)相等大小的范圍，每個(gè)范圍內(nèi)的數(shù)據(jù)都用定義2中的PAA表示；

(3)通過(guò)查表1確定其所在的區(qū)間，并將每個(gè)區(qū)間用不同符號(hào)編號(hào)，表示形式為R={A1i,A2i,…,Axi}。

表1 從3-8等分區(qū)劃分的高斯分布查找表

圖1為原始向量R用SAX表示后的模型。

圖1 時(shí)間序列R的SAX表示

定義1:高維相似性連接。給定兩個(gè)n維數(shù)據(jù)集R和S，假設(shè)R和S中的所有點(diǎn)都在n維空間中，已知距離閾值為ε,則在R和S上的相似性度量是計(jì)算所有點(diǎn)對(duì)組成的集合(Ri,Si)使得DE(Ri,Si)≤ε，其中Ri∈R,Si∈S和DE代表歐幾里得距離。如果將R和S看作同一數(shù)據(jù)集，則進(jìn)行自相似性連接。文中也主要使用自相似性連接進(jìn)行實(shí)驗(yàn)。

定義2：PAA表示。給定一個(gè)n維向量R，將其維數(shù)進(jìn)行等分，令N為等分后的維度，RN1,RN2,…,RNN是N個(gè)維度表示，其中有關(guān)系Rn=RN1∪RN2∪…∪RNN和Ri∪Rj=?。則向量R用PAA表示為Rn=(RN1,RN2,…,RNN)。

定義3：聚合度λ。假定R的維數(shù)為n,將其用PAA表示后的維度為N，則聚合度λ就定義為λ=n/N。

文中首先將原始高維向量進(jìn)行規(guī)格化。假設(shè)原始向量為R，其中均值為μ，標(biāo)準(zhǔn)差為σ，則規(guī)格化公式為：

Ri'=(Ri-μ)/σ

(1)

給定兩個(gè)向量R和S，它們的歐幾里得距離為:

(2)

它們的PAA表示后的距離可以定義為：

(3)

在文獻(xiàn)[15]中已經(jīng)證明，PAA表示DP的距離是原始?xì)W幾里得距離DE的下限，也就是說(shuō)：

(4)

給定兩個(gè)向量R和S以及它們的SAX表示Rs和Ss，可以定義新距離：

(5)

容易證明，SAX表示(MINDIST)之間的距離是PAA表示DP之間的距離的下限，并且DP是歐幾里得距離DE的下界;根據(jù)傳遞性，MINDIST是歐氏距離的下邊界近似：

(6)

3 高維相似性連接在Spark上的實(shí)現(xiàn)

高維相似性連接是文中討論的主要問(wèn)題。看似簡(jiǎn)單，但由于維數(shù)較大和數(shù)據(jù)量較多，導(dǎo)致計(jì)算成本呈指數(shù)倍增加。如何用較短的時(shí)間和較小的代價(jià)得到正確的結(jié)果是文中進(jìn)行研究的目的。以下是高維相似性連接在Spark集群上實(shí)現(xiàn)的詳細(xì)過(guò)程。

具體流程如圖2所示。

圖2 高維數(shù)據(jù)相似性連接在Spark上的流程

(1)數(shù)據(jù)預(yù)處理。主要通過(guò)map操作對(duì)數(shù)據(jù)進(jìn)行規(guī)格化處理，使每一行數(shù)據(jù)都表示為均值為0、標(biāo)準(zhǔn)差為1的標(biāo)準(zhǔn)序列。首先通過(guò)集群將分布式讀取到的每一行數(shù)據(jù)計(jì)算其均值μ和標(biāo)準(zhǔn)差σ，然后根據(jù)規(guī)格化公式，使其標(biāo)準(zhǔn)化，最后返回標(biāo)準(zhǔn)化后數(shù)據(jù)。

(2)生成SAX鍵值對(duì)。將預(yù)處理中計(jì)算出的標(biāo)準(zhǔn)序列結(jié)合預(yù)先設(shè)定的聚合度λ，表示出相應(yīng)的PAA集合。然后將生成的數(shù)據(jù)集用SAX表示，最后將生成的集合通過(guò)鍵值對(duì)返回。具體過(guò)程見(jiàn)算法1。

(3)將具有相同SAX表示的鍵值對(duì)聚合。該步驟主要是通過(guò)RDD算子的groupByKey函數(shù)實(shí)現(xiàn)。

(4)進(jìn)行笛卡爾積運(yùn)算。將上一步中SAX表示后的RDD數(shù)據(jù)集通過(guò)cartesian函數(shù)實(shí)現(xiàn)自連接，返回通過(guò)Key-value表示后的SAX對(duì)。這個(gè)步驟保證數(shù)據(jù)集中每?jī)蓚€(gè)SAX表示后的數(shù)據(jù)集可以碰面一次。

(5)精化候選向量對(duì)。該步驟是計(jì)算出滿足閾值的相似連接對(duì)。主要通過(guò)map函數(shù)比較上一步中返回的鍵值對(duì)中滿足閾值的key-value對(duì)，然后計(jì)算value中符合條件的原始數(shù)據(jù)對(duì)之間的距離。

算法1:Generate key-value pairs

line=>{

1.numbs<-line.substring(0,line.length-1).split(“ ”)

2.for i<- 0 until numbs.length do

3.numi <- numbs(i).toDouble

4.sum += numi

5.if k

6.if i

7.k=k+1

8.else

9.// Express by piecewise aggregate approximation

10.pp=sum/k

11.sax+=Util.getSax(pp)

12.else

13.pp=sum/k

14.sax+=Util.getSax(pp)

15.sum=0；k=1

16.//return key-value pairs

17.(sax, f.toList)

}

4 實(shí) 驗(yàn)

4.1 實(shí)驗(yàn)配置

該實(shí)驗(yàn)在hadoop2.6.1集群的基礎(chǔ)上部署了Spark1.6.1，程序用Scala編寫，編譯器為IDEA2016。這個(gè)集群包含10個(gè)節(jié)點(diǎn)，每個(gè)節(jié)點(diǎn)的配置如下：核數(shù)為4，內(nèi)存6 GB，磁盤500 GB，操作系統(tǒng)為L(zhǎng)inux CentOS release 6.2(Final)。其中1個(gè)為Master節(jié)點(diǎn)，其余9個(gè)為Worker節(jié)點(diǎn)，實(shí)驗(yàn)中使用的數(shù)據(jù)是從HDFS中讀取并將計(jì)算結(jié)果寫入HDFS中。

使用的數(shù)據(jù)集來(lái)自文獻(xiàn)[13]，選用了其中的部分?jǐn)?shù)據(jù)。其中256維和512維數(shù)據(jù)集通過(guò)960維數(shù)據(jù)生成，各維度向量均采用20萬(wàn)條數(shù)據(jù)進(jìn)行測(cè)試。

分別在Spark和Hadoop平臺(tái)上測(cè)試了相同數(shù)據(jù)集在不同條件約束下的運(yùn)行時(shí)間。

4.2 實(shí)驗(yàn)結(jié)果分析

為了驗(yàn)證改進(jìn)算法的有效性，分別從三個(gè)角度進(jìn)行了測(cè)試，即維數(shù)改變對(duì)實(shí)驗(yàn)結(jié)果的影響、聚合度λ改變對(duì)實(shí)驗(yàn)結(jié)果的影響和閾值ε改變對(duì)實(shí)驗(yàn)結(jié)果的影響。所有實(shí)驗(yàn)都是基于相同硬件配置，對(duì)原有Hadoop集群和優(yōu)化后Spark集群的運(yùn)行時(shí)間的比較。

首先，比較Hadoop平臺(tái)和Spark平臺(tái)在設(shè)置默認(rèn)聚合度λ和閾值ε的前提下，只改變維度大小對(duì)所需時(shí)間的影響情況，其中設(shè)置λ為16，ε為0.1。實(shí)驗(yàn)結(jié)果如圖3所示。

由圖3可以看出，兩種平臺(tái)對(duì)相同數(shù)據(jù)集維度的執(zhí)行時(shí)間是不同的。基于Spark平臺(tái)的運(yùn)行時(shí)間更短，與原有基于Hadoop平臺(tái)的算法相比平均可縮短10%～20%。并且數(shù)據(jù)維度越高，優(yōu)勢(shì)越明顯。

其次，分別改變聚合度λ和閾值ε的設(shè)定參數(shù)，分別測(cè)試對(duì)實(shí)驗(yàn)運(yùn)行時(shí)間的影響情況，其中用到的數(shù)據(jù)集維度默認(rèn)為128維。改變聚合度λ可以改變生成分段聚合近似數(shù)據(jù)集的維數(shù)，從而影響符號(hào)聚合近似表示數(shù)據(jù)集的生成維數(shù)。改變閾值ε可以控制計(jì)算相似度的多少，從而影響所需的計(jì)算時(shí)間。實(shí)驗(yàn)結(jié)果如圖4和圖5所示。

圖3 不同維度兩種框架運(yùn)行時(shí)間對(duì)比

圖4 兩種框架在不同聚合度λ下的運(yùn)行時(shí)間對(duì)比

圖5 兩種框架在不同閾值ε下的運(yùn)行時(shí)間對(duì)比

由圖4可以看出，雖然聚合度越大，運(yùn)行時(shí)間越小，但當(dāng)聚合度增加到一定值的時(shí)候，會(huì)看到運(yùn)行時(shí)間反向增加。這是因?yàn)樵谶M(jìn)行實(shí)際向量間運(yùn)算時(shí)，由于內(nèi)部向量維數(shù)較大，增加了運(yùn)算成本。但在同等條件下，Spark計(jì)算平臺(tái)的時(shí)間減小更多，運(yùn)行效率更高。

由圖5可以看出，改變閾值ε對(duì)實(shí)驗(yàn)結(jié)果的運(yùn)行時(shí)間有一定影響。閾值越大，花費(fèi)的時(shí)間越多，但基于Spark的優(yōu)化速度還是有明顯的優(yōu)勢(shì)。

5 結(jié)束語(yǔ)

隨著數(shù)據(jù)量規(guī)模的增加，相似性計(jì)算成本也將隨之變大。單獨(dú)使用一臺(tái)機(jī)器已經(jīng)無(wú)法滿足相似性計(jì)算的性能、時(shí)間等各方面的要求。提高高維數(shù)據(jù)間的相似性連接效率是必須面對(duì)的課題。文中主要針對(duì)高維數(shù)據(jù)集相似性分類算法進(jìn)行了優(yōu)化和改進(jìn)，使用SAX方法對(duì)數(shù)據(jù)集進(jìn)行裁剪，該方法減少了數(shù)據(jù)之間不必要的距離計(jì)算時(shí)間。結(jié)合現(xiàn)有大數(shù)據(jù)框架分布式并行計(jì)算的特點(diǎn)，可以很好地提高高維數(shù)據(jù)相似性連接效率。通過(guò)實(shí)驗(yàn)發(fā)現(xiàn)，使用Spark框架的高維數(shù)據(jù)集相似性連接算法具有顯著的性能優(yōu)勢(shì)。