MapReduce框架下一種負(fù)載均衡的Top-k連接查詢算法

，，

(1.首都師范大學(xué) 信息工程學(xué)院，北京 100048; 2.北京交通大學(xué) 交通運(yùn)輸學(xué)院，北京 100044)

0 引言

排序查詢處理對于大規(guī)模數(shù)據(jù)分析至關(guān)重要，通常使用的排序查詢方法稱為Top-k連接查詢算法[1]。Top-k查詢中，根據(jù)每個(gè)對象的屬性計(jì)算一個(gè)權(quán)重，再通過給定的評分函數(shù)為對象進(jìn)行評分，返回k個(gè)最重要的結(jié)果[2]。在-+大數(shù)據(jù)時(shí)代，用戶檢查大量未排序的查詢結(jié)果集是不現(xiàn)實(shí)的。并行化執(zhí)行不僅可以實(shí)現(xiàn)高效地運(yùn)行，并且可以返回精準(zhǔn)的結(jié)果。目前MapReduce是一種廣泛應(yīng)用的并行編程環(huán)境[3]。

目前，學(xué)者也提出了一些并行的Top-k連接查詢算法。例如，文獻(xiàn)[4]在MapReduce的背景下，提出了兩種關(guān)于Top-k連接的方法。一種稱為RanKloud的算法，其在掃描記錄期間計(jì)算統(tǒng)計(jì)數(shù)據(jù)，并使用這些統(tǒng)計(jì)數(shù)據(jù)計(jì)算提前終止的閾值(Top-k結(jié)果的最低分?jǐn)?shù))。此外，還提出了一種新的分區(qū)方法，稱為uSplit，旨在以使用敏感方式對數(shù)據(jù)進(jìn)行重新分區(qū)。然而，RanKloud不能保證正確的返回k個(gè)檢索結(jié)果。另外，常用的一種基于 MapReduce 框架的用來計(jì)算Top-k連接結(jié)果的通用二路連接算法為Reduce-side join,簡稱為RSJ[5]，其連接是在Reduce函數(shù)中實(shí)現(xiàn)。

本文在MapReduce編程模型中實(shí)現(xiàn)并行Top-k連接查詢算法(Parallel Top-k Join,P-TKJ)，同時(shí)融入提前終止機(jī)制和負(fù)載均衡機(jī)制來增強(qiáng)Top-k連接處理的性能。主要?jiǎng)?chuàng)新點(diǎn)為：在MapReduce中提出了一個(gè)新的Top-k連接處理框架，盡可能地利用并行性，并避免鏈接MapReduce作業(yè)的初始化開銷；使用直方圖形式的數(shù)據(jù)表示，并融入了提前終止策略、數(shù)據(jù)過濾和負(fù)載平衡策略，以便設(shè)計(jì)出高效的并行Top-k連接算法。

1 MapReduce 編程模型

MapReduce是Hadoop中的一個(gè)編程框架，為并行算法提供了一個(gè)容錯(cuò)和可靠的編程環(huán)境。為了處理大量的數(shù)據(jù)，該框架支持一個(gè)可擴(kuò)展的文件系統(tǒng)，稱為Hadoop分布式文件系統(tǒng)(HDFS)，用于在硬件群集中的機(jī)器上存儲大量文件。

MapReduce計(jì)算過程分成Map和Reduce兩個(gè)階段[6]，其中數(shù)據(jù)的格式以鍵值對呈現(xiàn)，其處理過程如圖1所示。

圖1 MapReduce編程模型的處理過程

2 問題描述

2.1 Top-k連接查詢

給定具有n個(gè)得分屬性的輸入表或關(guān)系T，使用τ代表T的記錄(或元組)，τ[i]是指第i個(gè)得分屬性(i∈[1,n])。Top-k查詢q(k,f)基于單調(diào)評分函數(shù)f返回k個(gè)最佳查詢結(jié)果。當(dāng)應(yīng)用于關(guān)系T時(shí)，Top-k查詢q(k,f)的結(jié)果是T中一組k個(gè)記錄τi，…，τk中得分最小的Δk，即f(θ)的值。在不失一般性的情況下，分?jǐn)?shù)最低的記錄被認(rèn)為是最好的[7]。

通常在排名感知處理中，需要兩個(gè)(或更多)輸入關(guān)系連接的Top-k結(jié)果，視為一個(gè)運(yùn)算符，稱之為Top-k連接查詢[8]。可以通過先執(zhí)行連接，然后通過評分函數(shù)對連接記錄進(jìn)行排名，并輸出前k個(gè)排名結(jié)果。然而，這會導(dǎo)致處理過程的資源浪費(fèi)，所以需要提出高效的算法來解決交織排序和連接的問題[9]。

在本文中，認(rèn)為輸入關(guān)系Ti包含了一個(gè)連接屬性ai，一個(gè)得分屬性si，以及其他一些屬性。因此，Ti由唯一標(biāo)識符(τ,id)、連接屬性值或連接值(τ,ai)和得分屬性值(τ,si)所描述的記錄組成。本文關(guān)注二元多對多的Top-k等值連接，其中輸入表T0和T1連接在連接屬性a0=a1上，得分屬性(s0和s1)的組合是為了生成Top-k連接記錄，作為得分函數(shù)f的輸入。

2.2 問題描述

考慮兩個(gè)輸入表T0和T1，它們分別在一組機(jī)器上被水平分割，并具有連接屬性a0，a1和得分屬性s0，s1。給定由整數(shù)k定義的Top-k連接查詢q(k,f,T0,T1)，和用于組合得分屬性s0和s1產(chǎn)生連接記錄的單調(diào)得分函數(shù)f。并行Top-k連接問題要求產(chǎn)生具有最低分?jǐn)?shù)的Top-k連接記錄。

在MapReduce環(huán)境中，輸入表T0和T1被拆分為HDFS塊，并按照水平分區(qū)的概念存儲在HDFS中。一個(gè)記錄τ在每個(gè)文件中都是(τ.id,τ.ai,τ.si)形式，其中τ.id是唯一標(biāo)識符，τ.ai是連接屬性，τ.si是得分屬性。除了這個(gè)三元組之外，每一行可能都包含其他任意長度的記錄元素的屬性τ。因此，在一般情況下，每個(gè)節(jié)點(diǎn)只存儲每個(gè)關(guān)系記錄的一個(gè)子集。問題在于設(shè)計(jì)一個(gè)由Map和Reduce階段組成的算法，通過并行方式有效計(jì)算Top-k連接方式。

最后，本文注意到Top-k連接并行處理中最昂貴的部分是計(jì)算每個(gè)連接值的Top-k連接記錄。因此，在本文中，我們著重于提供一個(gè)完全并行的解決方案來解決這個(gè)問題。獲得Top-k連接結(jié)果的最后一步需要處理k·m個(gè)連接記錄(其中m表示不同連接值的個(gè)數(shù))，這通常比初始表Ti的值小幾個(gè)數(shù)量級，即k·m<<|Ti|。因此，可利用一個(gè)集中程序來處理這些單獨(dú)的Top-k結(jié)果，而沒有顯著的開銷。

3 提出的并行Top-k連接查詢算法

3.1 方法概述

上傳兩個(gè)輸入表T0和T1，并作為單獨(dú)的文件存儲在HDFS中，根據(jù)得分屬性以升序排序。此外，對于每個(gè)輸入表，計(jì)算并存儲在HDFS直方圖H(T0)和H(T1)中，它們維護(hù)一系列連接屬性值的記錄數(shù)。需要注意的是，這些信息可以在輸入表上傳到HDFS的過程中構(gòu)建，而開銷可以忽略不計(jì)。

給定一個(gè)Top-k連接查詢，計(jì)算每個(gè)輸入表(基于直方圖)的分?jǐn)?shù)范圍，這些范圍決定了作業(yè)執(zhí)行前足以產(chǎn)生正確結(jié)果的記錄子集。因此，可以選擇性地在Map階段加載和處理存儲數(shù)據(jù)的一小部分，一旦遇到分?jǐn)?shù)值大于邊界的記錄，就終止Mappers的處理。此外，通過引入數(shù)據(jù)過濾和負(fù)載均衡機(jī)制來優(yōu)化Reduce端連接的性能，該機(jī)制均衡地將連接值分配給Reduce任務(wù)。

3.2 直方圖構(gòu)建

在Hadoop中處理數(shù)據(jù)需要上傳數(shù)據(jù)，整個(gè)數(shù)據(jù)集從外部源按順序讀取并存儲在HDFS中[10]。這個(gè)階段主要是I/O密集型任務(wù)，CPU沒有充分利用，可以利用這個(gè)階段在后臺建立直方圖。通常情況下，直方圖的大小比初始數(shù)據(jù)集要小幾個(gè)數(shù)量級，但是在準(zhǔn)確性和磁盤大小之間權(quán)衡，即在構(gòu)建過程中更大直方圖可以實(shí)現(xiàn)更高的準(zhǔn)確性，同時(shí)會消耗更多磁盤空間的。

為達(dá)到預(yù)期的目的，本文選擇了構(gòu)建等寬直方圖，其構(gòu)造簡單且符合一次通過的要求。更詳細(xì)地說，當(dāng)一個(gè)記錄τ(τ.ai,τ.si)在上傳階段被讀取，可以通過增加對應(yīng)分?jǐn)?shù)值τ.si的bin的內(nèi)容來更新連接值為τ.ai的直方圖。

圖2描繪了相同連接屬性值下，T0和T1的等寬直方圖。對于每個(gè)輸入表Ti，創(chuàng)建與連接屬性中單獨(dú)值數(shù)量一樣多的直方圖。每個(gè)直方圖被表示為H(Ti)。例如，所描述的T1的直方圖H(T1)表示它總共包含11個(gè)具有連接值a1=x的記錄。此外，第一個(gè)直方圖框表示存在2條記錄，得分在0-10之間(表示為[0-10]：2)，剩下的bin是：[10-20]：3，[20-30]：2以及[30-40]：4。

圖2 相同的連接屬性值(a0=a1=x)下，T0和T1的等寬直方圖的例子

3.3 提前終止機(jī)制

為了減少連接的處理成本，本文只處理兩個(gè)表的輸入記錄子集，來保證提供正確的Top-k連接結(jié)果。直觀地說，只有表Ti中分?jǐn)?shù)低于bi的記錄才會參與連接，用來產(chǎn)生Top-k連接結(jié)果。 因此，為了實(shí)現(xiàn)提前終止操作，需要有一種方法來確定分?jǐn)?shù)范圍b0和b1，以便盡可能早地放棄高于bi分?jǐn)?shù)的記錄。

1)分?jǐn)?shù)界限估計(jì)：將兩個(gè)表的直方圖作為輸入，問題在于要計(jì)算每個(gè)表Ti中輸入記錄得分的正確分?jǐn)?shù)界限bi。為此，本文使用文獻(xiàn)[11]中提出的算法來進(jìn)行分?jǐn)?shù)界限估計(jì)。在實(shí)踐中，這個(gè)算法對兩個(gè)表格的直方圖執(zhí)行連接，并估計(jì)連接結(jié)果的數(shù)量和分?jǐn)?shù)范圍。這個(gè)算法的用處為：第一，識別直方圖bin和相應(yīng)分?jǐn)?shù)范圍用來產(chǎn)生k個(gè)連接記錄；第二，確保沒有其他直方圖bin組合可以產(chǎn)生具有比這第k個(gè)連接記錄更小分?jǐn)?shù)值的連接記錄。為此直方圖bin不斷被訪問和加入，直到加入記錄的數(shù)量超過k，或者任何直方圖bin產(chǎn)生的連接記錄得分都不小于當(dāng)前第k個(gè)記錄的得分。用一個(gè)例子來解釋算法的操作，描述如下。

示例1:考慮圖2中描述的直方圖，并假設(shè)Top-k連接結(jié)果(k=1)被要求使用作為評分函數(shù)的總和。通過檢查每個(gè)直方圖的第一個(gè)bin，可以知道在[0-15]范圍內(nèi)存在2(= 1×2)個(gè)連接記錄，即[0-15]：2。通過每個(gè)直方圖，還可以知道存在[10-25]：3，[5-20]：4和[15-30]：6。只有在T0的第三個(gè)bin被檢查后(產(chǎn)生的連接記錄沒有顯示在這里)，才可以安全地停止處理，并且報(bào)告得分范圍b0=15和b1=20。這是因?yàn)榈梅諿0-15]內(nèi)已經(jīng)有至少2條記錄(即多于k=1)，并且T0或T1bin組合產(chǎn)生的任何連接記錄的分?jǐn)?shù)都將大于15。

2)在Hadoop中實(shí)現(xiàn)提前終止操作：假設(shè)輸入表以HDFS格式存儲，并且直方圖也可用，創(chuàng)建一個(gè)提前終止機(jī)制，在Map階段有選擇地只處理分?jǐn)?shù)比各自界限低的輸入記錄。需要注意的是，提前終止機(jī)制是通過擴(kuò)展Hadoop來實(shí)現(xiàn)的，也就是說，不會更改Hadoop核心。

3.4 數(shù)據(jù)過濾

Map任務(wù)會處理一組輸入記錄(以鍵值對的形式)并生成一組輸出記錄。限制輸出記錄的數(shù)量非常重要，這會影響整體性能，因?yàn)檫@些記錄需要通過Reduce任務(wù)進(jìn)行混洗(消耗通信成本)和處理(消耗處理成本)。數(shù)據(jù)過濾技術(shù)通常是通過消除不影響結(jié)果的輸入記錄來限制Map輸出記錄的數(shù)量。應(yīng)該注意的是，數(shù)據(jù)過濾是依賴于作業(yè)的，這意味著每個(gè)作業(yè)都需要基于查詢類型的不同過濾機(jī)制。

Top-k查詢的過濾過程中，考慮在n維空間Rn中定義的多維數(shù)據(jù)集S(例如，p∈S且p=[p1,…pn])，以及一個(gè)Map任務(wù)，即訪問完整數(shù)據(jù)集S的子集S′。另外，讓一個(gè)偏好函數(shù)f(p)=ω1·p1+…+ωn·pn為數(shù)據(jù)對象賦值。目標(biāo)是檢索出得分最高的top-k對象。對于由Map任務(wù)讀取的每個(gè)對象p∈S′，分配一個(gè)分?jǐn)?shù)f(p)。通過在優(yōu)先隊(duì)列中保存k個(gè)最高得分對象來執(zhí)行Map任務(wù)中的過濾。只有這些k個(gè)對象需要發(fā)送到Reduce階段，而不是由Map任務(wù)訪問的|S′|個(gè)對象。

圖3所示為一個(gè)2維數(shù)據(jù)集中的Top-k查詢過濾例子。白點(diǎn)和黑點(diǎn)對應(yīng)于由兩個(gè)不同Map任務(wù)訪問的對象。 黑點(diǎn)對象的局部Skyline集合用虛線連接。這些是一個(gè)Map任務(wù)中唯一需要發(fā)送到Reduce階段的象，而剩余的黑點(diǎn)則被過濾。

圖3 具有兩個(gè)Map任務(wù)的Top-k示例，空心點(diǎn)對應(yīng)第1個(gè)Mapper，實(shí)心點(diǎn)對應(yīng)第2個(gè)Mapper

3.5 負(fù)載均衡機(jī)制

Reduce任務(wù)的工作量由其需要處理和連接的記錄數(shù)決定[12]。為了執(zhí)行負(fù)載均衡，本文目標(biāo)是將一些連接值分配給Reduce任務(wù)，以最小化每個(gè)Reduce任務(wù)的最大記錄數(shù)，這個(gè)問題相當(dāng)于多處理器調(diào)度問題。然而，多處理器調(diào)度問題是一種NP-hard問題，因此本文使用了一種名為LPT(最長處理時(shí)間)的啟發(fā)式算法來進(jìn)行調(diào)度。該算法根據(jù)連接記錄的數(shù)量對連接值進(jìn)行排序，然后將它們分配給迄今為止連接總數(shù)最低的處理器(Reducer)。

3.6 基于 MapReduce的并行實(shí)現(xiàn)

算法1展示了如何在Map階段實(shí)現(xiàn)提前終止、數(shù)據(jù)過濾和負(fù)載均衡機(jī)制。該算法將每個(gè)輸入表的分?jǐn)?shù)界限作為輸入，并訪問排序的輸入表。另外，如上所述，HashMapH用來捕獲一些連接值分配給Reduce任務(wù)。只要表T1中的輸入記錄τ的得分低于得分邊界b1，即∑si≤bi，則將該記錄傳遞給Reduce任務(wù)。以此確保沒有得分高于邊界的記錄可以產(chǎn)生屬于Top-k連接的連接結(jié)果，從而可以棄用高得分記錄的連接結(jié)果，顯著減少需要傳遞和處理的記錄數(shù)量。

算法1:P-TKJ Map階段輸入：T0，T1，b0，b1，H

輸出：T0，T1中分?jǐn)?shù)低于b0，b1的記錄

Function Map(τ(τ.ai,τ.si))//表Ti中的記錄

1：r←H.get(τ.ai)

2:if (τ∈T0) then

3： if(τ.s0≤b0) then

4:τ.tag←0

5： output[(τ.ai,τ.si,τ.tag,r),τ]

6； else

7： if(τ.s0≤b1) then

8：τ.tag←1

9： output[(τ.ai,τ.si,τ.tag,r),τ]

10： 執(zhí)行數(shù)據(jù)過濾

11：end

算法2展示了Reduce階段的流程。將Map階段的輸出鍵值對根據(jù)連接值(τ.ai)分組，并使用自定義分區(qū)程序分配給Reduce任務(wù)。在每個(gè)Reducer中，需要按照得分(τ.si)的升序?qū)γ總€(gè)組中的記錄進(jìn)行排序，這是通過使用組合鍵排序來實(shí)現(xiàn)的。Reduce階段的輸出形式為a,τ.id,τ'.id,f(τ.τ')。

每個(gè)Reduce任務(wù)將與特定連接屬性值相關(guān)的所有記錄作為輸入，并獨(dú)立于其他Reduce任務(wù)，對每個(gè)這樣的連接值執(zhí)行Top-k連接。而且，由于按升序?qū)τ涗涍M(jìn)行排序訪問，因此只要在存儲器(M0和M1)中，從每個(gè)輸入表(第6行)中只讀取與k相同數(shù)量的記錄即可，因?yàn)槿魏纹渌涗浂疾荒墚a(chǎn)生Top-k連接結(jié)果。

算法2 :P-TKJ Reduce 階段

輸出:連接值key的Top-k記錄。

Function Reduce(key,V)

1:for (τ∈V) do

2:if(τ.tag=0) then

3:載入τinM0

4:else

5:載入τinM1

6:if(M0.size()≥k)and(M1.size()≥k)則

隨著互聯(lián)網(wǎng)和云計(jì)算技術(shù)的急速發(fā)展和普及，云計(jì)算在提高使用效率的同時(shí)，為數(shù)字內(nèi)容安全和用戶個(gè)人敏感信息保護(hù)帶來了很大的挑戰(zhàn)。

7:執(zhí)行提前終止機(jī)制

8:output[RankJoin(k,f,M0,M1)]

9:end

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

4.1 實(shí)驗(yàn)設(shè)置

將算法部署在由8個(gè)服務(wù)器節(jié)點(diǎn)組成的內(nèi)部Hadoop集群[13]中。對于Map和Reduce任務(wù)，JVM堆大小設(shè)置為2GB。HDFS大小配置為128MB，默認(rèn)復(fù)制因子為3。

使用了兩種Hadoop平臺上計(jì)算Top-k連接的算法進(jìn)行比較，分別為傳統(tǒng)RSJ算法和本文提出的P-TKJ算法。這兩種算法的區(qū)別在于，本文P-TKJ算法具有提前終止、數(shù)據(jù)過濾和負(fù)載均衡機(jī)制。

對于記錄數(shù)據(jù)集，使用了一個(gè)合成數(shù)據(jù)生成器來生成大量的輸入數(shù)據(jù)集。輸入表Ti的大小從1 GB到50 GB。根據(jù)偏態(tài)分布(ZIPF分布)來生成評分屬性，其中偏度為0.5，表示為ZI0.5。改變每個(gè)表中不同連接值的數(shù)量(從100到2000)，從而影響連接選擇性，以研究它對本文算法的影響。為了驗(yàn)證算法的可擴(kuò)展性，本文創(chuàng)建了4個(gè)不同大小的數(shù)據(jù)集，記為DS1-DS4。這些數(shù)據(jù)集的各個(gè)參數(shù)顯示在表1中。另外，各種算法中都設(shè)置Top-k中的k=10。

對于性能指標(biāo)，本文使用的主要度量是每個(gè)作業(yè)的總執(zhí)行時(shí)間。另外，還測量了在Map和Reduce階段消耗的CPU時(shí)間。

表1 用于可擴(kuò)展性研究的數(shù)據(jù)集

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

圖4給出不同數(shù)據(jù)集大小下，兩種算法的總執(zhí)行時(shí)間。圖5給出了分別在Map和Reduce階段所消耗的CPU處理時(shí)間。

圖4 算法的總執(zhí)行時(shí)間

圖5 Map和Reduce階段所消耗的CPU處理時(shí)間

可以看出，P-TKJ算法的執(zhí)行時(shí)間優(yōu)于RSJ 算法將近1倍。而且，當(dāng)數(shù)據(jù)集的大小增加時(shí)，優(yōu)勢更加明顯。以上實(shí)驗(yàn)這有力證明了本文算法支持大量輸入的可擴(kuò)展性。

這是因?yàn)镽SJ雖然為并行Top-k連接問題提供了一個(gè)正確的解決方案，但是它在性能方面有嚴(yán)重的局限性。首先，盡管直觀上一小部分列表記錄就足以產(chǎn)生正確的結(jié)果，但是它需要完整地訪問兩個(gè)輸入表。換句話說，就磁盤訪問、處理成本以及通信而言，這明顯導(dǎo)致資源的浪費(fèi)。理想情況下，如果確定已經(jīng)訪問過的記錄能夠產(chǎn)生正確的結(jié)果，只需要有選擇地只訪問幾個(gè)HDFS塊，并終止Map階段的處理。其次，由于RSJ不使用與每個(gè)連接值關(guān)聯(lián)的記錄數(shù)量知識，為此其將Map輸出鍵(連接值)分配給Reduce任務(wù)是隨機(jī)執(zhí)行的，這可能會導(dǎo)致不均衡的工作分配，從而延遲了工作的完成。

相比而言，本文使用了提前終止策略，使Map階段輸入記錄的數(shù)量減少，所以算法比RSJ執(zhí)行更快。另外，由于本文方法很好地對Reducer任務(wù)進(jìn)行了負(fù)載平衡。在沒有負(fù)載均衡機(jī)制時(shí)，使用Hadoop默認(rèn)的基于散列的分區(qū)，將Map輸出鍵分配給Reducers，這本質(zhì)上是一種隨機(jī)分區(qū)。而由于本文的負(fù)載均衡機(jī)制，以更統(tǒng)一的方式將連接結(jié)果分配給Reducers，從而以更公平的方式分配工作。另外，本文融入了數(shù)據(jù)過濾操作，減少了Reducer任務(wù)數(shù)量，這也一定程度上提高了算法執(zhí)行速度。

為了驗(yàn)證不同k對算法性能的影響，這里設(shè)定k=5、10、15、20和25。在DS1上分別進(jìn)行實(shí)驗(yàn)，并統(tǒng)計(jì)相應(yīng)的執(zhí)行時(shí)間，結(jié)果如圖6所示?？梢钥闯?，不同k值下兩種算法的執(zhí)行時(shí)間幾乎不受影響。這是因?yàn)檫B接查詢是消耗時(shí)間最高的操作。但Top-k通過在連接階段實(shí)行部分合并，不同k值下所維護(hù)的元組數(shù)量基本相同，所以執(zhí)行時(shí)間也基本不變。

圖6 不同k值下的執(zhí)行時(shí)間

5 結(jié)論

本文介紹了一種在MapReduce框架上處理Top-k連接的并行化計(jì)算框架。使用數(shù)據(jù)匯總，以直方圖的形式表示，并將這些操作在數(shù)據(jù)上傳過程中通過后臺CPU處理，以此提高CPU利用率。同時(shí)利用提前終止策略、數(shù)據(jù)過濾和負(fù)載均衡策略提高了算法對數(shù)據(jù)分析訪問和處理的效率。實(shí)驗(yàn)結(jié)果證明了提出算法的可擴(kuò)展性和有效性。