差別依賴驗證的分布式算法

覃 昇 談子敬 肖永松

(復旦大學計算機科學技術學院 上海 200433)

0 引 言

數據質量問題在大數據時代變得更為突出。無論系統能夠以多么快的速度處理多么大的數據量，如果數據質量得不到保證，一切得到的分析結果可能都是無意義的。在對數據質量進行評估的多個維度中，數據一致性是一個常見的指標。數據一致性通常基于用邏輯表達式的形式所給出的數據依賴來進行描述。簡而言之，若數據不滿足給定的數據依賴，則稱該數據為不一致，這通常意味著數據中存在錯誤或誤差。

因為數據依賴的重要性，有大量的研究針對各種數據依賴展開。常見的數據依賴多數強調的是數據間的相等關系，例如函數依賴、條件函數依賴[1]等。在現實中，相關數據之間可能存在一定差異，而并非嚴格完全相等，也可能需要在依賴定義中引入大于、小于等序列關系。差別依賴[2]是一個表達能力較強的數據依賴定義形式，它滿足了以上這些需求。常見的函數依賴是差別依賴的一個特例。

在數據一致性的相關工作中，數據集上的數據依賴驗證是一個基礎且重要的步驟：其目標是在數據集中找到不滿足數據依賴的部分數據，以對其進行進一步的分析和修復。隨著數據量的變大，數據依賴驗證所需的內存和對處理器的要求越來越高，單機無法實現。這需要引入分布式計算技術，將問題并行處理。和單機算法不同，分布式算法需將整個問題分成若干個可并行的子問題，每個子問題由一臺計算機作為一個節點(reducer)進行并行處理，最終整合結果。和設計單機算法只注重時間空間消耗不同，設計一個分布式算法需要考慮的主要因素，包括如何將問題盡可能平均地并行化分解；如何在保證正確的情況下減少并發運行時間，同時減少分發和收集過程中的數據量等。

本文研究的問題是：基于大規模分布式計算平臺，在數據集上進行分布式差別依賴的驗證。算法將基于差別依賴的特征，對部分情況進行優化，以提出更優的算法。

1 背景知識

1.1 差別依賴

首先回顧一下差別依賴的具體定義[2]。

對于關系數據表R中的一個屬性B，在其值域dom(B)上定義一個二元差別dB(a,b)(a,b∈dom(B))。這個二元差別滿足：(1) 非負性：dB(a,b)≥0，dB(a,b)=0當且僅當a=b；(2) 對稱性：dB(a,b)=dB(b,a)。例如，實數域上的絕對值運算dB(a,b)=|a-b|是一個符合上述要求的二元差別，字符串的最小編輯距離也同樣屬于二元差別。

對于多屬性上的差別函數φ[Z]，其中Z為若干個屬性組成的集合。則有：

(1)

定義2一個數據表R的差別依賴DD，其形式為:

DD:φ1(X)→φ2(Y)

(2)

其中：X、Y是R中屬性集的子集，φ1(X)和φ2(Y)是兩個不同的差別函數。

對一個信用卡交易的數據庫，可以有如下差別依賴：

DD1[cardno(= 0)]∩[position(≥60)]→[transtime(≥20)]

其含義為：當兩筆交易的卡號(cardno)相同，并且發生地點(position)相距不小于60時，則這兩筆交易的時間(transtime)一定不小于20。

對于一個產品價格記錄的數據庫，可能有如下約束：當兩條記錄的日期(date)相距在7～30天時，則價格之差將在100～900的范圍內，其表達式為：

DD2[date(>7,≤30)]→[price(≥100,≤900)]

可以看到，差別依賴是一個具有較強表達能力的依賴類型，并且函數依賴是它的一個特例。由差別依賴的定義可知，差別依賴的約束驗證，需要對數據集中的任意元組對進行比較。該算法的復雜度是元組個數的平方。當數據集具有較大規模時，單機的內存和時間將無法承受該負荷，所以使用分布式系統進行差別依賴的驗證。

1.2 分布式系統及算法

在MapReduce[3]和Spark系統中，一個分布式系統由若干無共享存儲空間的計算機構成，每一臺計算機被視作一個節點(reducer)，所有節點之間的交互通過網絡中的發送和收集信息來實現。一個分布式算法由若干輪映射歸約(MapReduce)操作組成，每一輪操作分為映射(Map)、轉移(Shuffle)和歸約(Reduce)三部分。其中：Map為數據傳輸做準備，產生包含數據內容的鍵值對(Key-Value)；Shuffle根據數據的鍵值進行實際的數據傳輸，使得所有鍵值相同的數據被歸約到同一個節點；當每一個reducer將具有相同鍵值的數據收集之后，便繼續完成Reduce中之后的步驟，可能進行統計或輸出，也可能開始下一輪的MapReduce操作。三個操作之中，以Map和Reduce為算法核心。

1.3 分布式驗證相關工作

文獻[4]中主要討論了當數據表以水平分割或垂直分割的形式存儲時，如何進行條件函數依賴的檢驗，使得數據傳輸量或并發運行時間最小。文獻[5]中提出了一種基于等價類的分布式環境多函數依賴沖突檢測的方法，給出了沖突檢測的響應時間代價模型，并且將問題化為整數規劃問題，給出近似解。文獻[6]中提出了一種分布式環境多函數依賴不一致性檢測方法，依靠最小集合覆蓋的理論，通過一次數據遍歷，對多個函數依賴進行并行檢測。文獻[7]中將數據依賴的檢測和數據清洗化歸為一系列的原子操作，并針對多操作提出了并行或合并的改進。

本文研究的是差別依賴在分布式環境下的驗證。如前所述，差別依賴不僅包括函數依賴，還包括一系列建立在不相等或相近的數據上的依賴條件。本文的算法設計考慮了差別依賴的定義形式，同時針對部分的數據分布特征，進一步給出優化策略。

2 三角分布算法

差別依賴的驗證基于表中元組的兩兩比較、直觀理解，這和數據庫中一個表上的自連接(join)運算有相似之處。在分布式的設定下，為了實現元組的兩兩比較，并且保證不重復、不遺漏以及時間上的可接受，需要一定的運算技巧。文獻[8]針對大數據下元組的查重問題，使用了一種三角形分布的算法。算法的本質是模仿向量的叉乘，同時根據運算的對稱性，刪去其中約一半的運算。從結果上看，三角分布算法是將reducer排列成一個三角形，元組根據一定要求，通過MapReduce算法，分發到對應的reducer中。該算法不僅保證了正確性和比較次數的最優化，同時也保證了每一臺機器的時間復雜度都相同。

以文獻[8]中的方法為基礎，本文給出一個適用于差別依賴檢測的隨機三角發表算法。

2.1 隨機三角算法

當reducer數目給定時，比如為m，取最大的正整數l，使得l(l+1)/2≤m成立，此時l為三角分布的邊長。圖1為m=21、l=6的情況，其中每一個小方塊表示一個reducer，左上角的數字為其編號。按照圖1對每行每列進行編號，每一臺reducer可以由一個整數對來表示。例如編號為9的reducer同樣可以表示為(4，2)。

圖1 三角分布節點示意圖

利用三角分布策略，可以有效實現元組的成對比較，其具體算法見算法1。

算法1隨機三角分布算法

輸入：數據表的元組t；

三角分布邊長l；

差別依賴DD。

輸出：違背DD的所有元組對。

1: class MAPPER

2: method MAP(Tuple t)

3: Iht a ← a random value from [1,l]

4: for all p∈[1,a) do

5: EMIT(Reducer(p,a),(L,Tuple t))

6: EMIT(Reducer(a,a),(S,Tuple t))

7: for all q∈(a,I] do

8: EMIT(Reducer(a,q),(R,TupIe t))

9: class Partitioner

10: method PARTITION(Key rid,Pair(c,t))

11: Return rid

12: class REDUCER

13: method REDUCE(Key rid,Pairs[(c1,t1),(c2,t2)…])

14: Left,Right,Self ← ?

15: for all Pair (c,t)∈Pairs[(c1,t1),(c2,t2)…〗

16: if (c=L) then Left ← Left+t

17: else if (c=R) then Right ← Right+t

18: else if (c=S) then Self ← Self+t

19: if Self ≠ ? then

20: for all t1∈Self do

21: for aIl t2∈Self do

22: check t1and t2

23: else

24: for all t1∈Left do

25: for all t2∈Right do

26 : check t1and t2

27: Return Pairs(t1,t2) violating the DD

整個算法建立在分布式結構中，共有一次MapReduce操作，包含Map、Shuffle和Reduce三個階段。

在Map階段(第1-8行)，對每一個元組，選取一個1到l的隨機數a(第3行)，將元組發射到第a行和第a列的所有reducer中(第4-8行)。發射的過程中，需要進行標記，例如一個元組的隨機數為4，則在發送到第4列的4、9、13號機器上時，標記為L；發送到對角線上的16號機器上時，標記為S；發送到第4行的17、18號機器上時，標記為R。

在Shuffle階段(第9-11行)，數據根據機器序號進行分組。

在Reduce階段(第12-27行)，每一臺非對角線上的機器，接收到標記為L和R的兩類數據(第16-17行)，雙重循環比較L和R集合之間的元組對(第23-26行)，是否符合需要驗證的差別依賴；而在對角線上的機器，則只會收到標記為S的數據(第18行)，收集之后對其內部的所有元組對進行兩兩比較(第19-22行)。全部比較結束之后，返回違背DD的元組對(第27行)。

2.2 算法正確性和時間復雜度

算法的正確性等價于任意兩條元組都必須進行過比較。對于任意兩條元組，設它們在第3行得到的隨機數為i和j，由于差別依賴具有對稱性，設i≤j。若i=j時，元組會在機器(i,i)上進行比較；若i

從時間上看，算法的整體耗時主要集中在Reduce階段的數據比較上。設元組總數為n，則取到相同隨機數的元組個數平均值為n/l。

對于非對角線上的機器，其L和R集合中的元素數量均為n/l，所以總比較次數為n2/l2；

對于對角線上的機器，其S集合中元素數量為n/l，所以總比較次數為n2/l2；考慮到差別依賴的對稱性，(t1,t2)和(t2,t1)的比較完全一致，所以總比較次數可以減少至n2/2t2。

綜上，三角分布策略時間復雜度為O(n2/l2)，并且每一臺機器上的時間平均復雜度相同。

2.3 排序三角算法

對于一部分數據集，其在某些屬性上的值的上下界是已知的，并且分布大致均勻。若能通過這些已有的信息，在Map過程中就進行初篩，則可以減少部分運算時間。基于這樣的想法，可以對部分數據集上的差別依賴的檢驗，使用如下的排序三角算法。

考慮一條左側含有[A(θp)]的差別依賴，其中A是屬性，θ是判斷運算符(θ∈{≥,>,=,<,≤)，p是常數。同時A屬性的值存在上下界，大致地均勻分布在區間[smin,smax]中。令:

(3)

t表示單位區間長度，則整個[smin,smax)區間可以被分成l個長度為t的單位區間。在算法1中，a為1到l中的隨機數(算法1第3行)，而在排序三角算法中，a的取值為:

(4)

式中：t[A]為該元組在A屬性上的取值，則a∈[1,l]，并且所有元組被有序分散到reducer中。令k=p/t，則可以根據k的值免去部分reducer上的元組比較。

如圖2所示，取k=1，l=4。記difi為在編號為i的reducer中，需要比較的兩個元組的屬性A上值的差的范圍。例如dif7=(p,3p)，因為在這個reducer中需要比較的兩個元組的A屬性的值分別屬于[3p,4p)和[p,2p)，所以兩個數據之差的范圍為(p,3p)。

圖2 節點數據示意圖

圖3顯示了在圖2條件下各reducer中可能進行的運算，顯然針對特定的θ,并非所有reducer都有工作。記numθ為θ取不同運算符時所需要的reducer數量，在忽略少量的邊界情況和暫定k為整數時，計算可得：

num≤=num<=(l+l-k)(k+1)/2

(5)

num==l-k

(6)

num≥=num>=(l-k)(l-k+1)/2

(7)

圖3 各節點和運算符的對應關系

當初始條件都給定時，可以計算出對應的numθ，所以只需要這些數量的reducer工作就可以完成所有的比較，剩下的reducer無需運行。所以可以在map階段，直接減少數據的發射量，數據發射(emit)時，若目標reducer不需要進行運算，則可以跳過這個當前的發射步驟，減少Shuffle階段的數據運輸量。

另一方面，既然可以減少需要使用的reducer，那自然也可以用這些reducer來加速整個過程。

在已知reducer數目情況下(數目為m)，可以令邊長l′為變量，根據之前k和numθ的定義式，解出滿足numθ≤m的最大l′值，顯然有l′≥l。當三角分布邊長從l變大到l′以后，由于每一臺機器分配到的運算量的復雜度相同，而總運算量不變，所以每一臺使用到的機器上分到的運算量會相應變少，總時間減少。

如果k=p/t不為整數，則需要在上述numθ的基礎上進行一定的增加。從圖2中可以看出，若k=1.5，θ取<，則編號為3和7的reducer同樣需要進行運算，num<會增大。在這個情況下，在一部分的行上需要額外增加一個reducer，所以每個numθ都會增加，增加的數目上界為l。在這些額外加入的reducer中，并不是所有的元組對都符合[A(θp)]，所以還需要進行檢驗。

3 實驗分析

實驗使用的分布式環境是Spark 2.0.0，主要的啟動參數見表1。

表1 Spark啟動參數表

實驗所使用的測試數據集有兩個，均為人為構造的大數據集。兩個數據集均含有屬性A，該屬性的值域為[0,100)，不同點在于，數據集Data1在屬性A上的值為均勻分布，數據集Data2在屬性A上的值呈正態分布，平均值為50，方差為20，即在[30,70]范圍內包含了總數據的約70%。

在實驗1和實驗2中，只使用Data1作為測試數據測試三角算法的性能，而在實驗3中，使用Data1和Data2作為測試數據，比較隨機三角算法和排序三角算法的優缺點。

實驗所使用的差別依賴DD，其左側共有兩個差別函數，其中包含有差別函數φ(A)=[A(θ,30)]。在實驗1和2中，指定θ為<，在實驗3中，將取θ分別為<、=和>，進行兩種三角分布的性能比較。

3.1 時間比較

目前在分布式平臺上，用來處理結構化數據的常見工具有SparkSQL和Hive。

SparkSQL是Spark上的一個模塊，可以從外部讀入數據內容，將其轉化為特有的DataFrame數據結構，分布式地進行存儲和運算。同時它包含了部分數據庫常見的操作函數，可在分布式平臺上進行用戶的查詢或修改操作。

Hive是基于分布式平臺的一個數據倉庫工具，可以將結構化的數據文件映射為一張數據庫表。在Hive上可以進行結構化查詢語言SQL的語句查詢。當輸入SQL語句之后，Hive會把SQL語句轉換為MapReduce任務，然后在分布式系統上進行運行。

通過將差別約束的驗證改寫為SQL語句，在Hive和SparkSQL下運行。

例如，在表格Table1上有差分依賴：

DD3[A(> 10)]→[B≤20)]

現需要返回違背DD3的元組對的數量，可以用如下SQL語句來描述：

SELECT COUNT(*)

FROM Table AS X,Table AS Y

WHERE(X.A-Y.A)>10 AND ABS(X.B-Y.B)>20

實驗1使用隨機三角分布、SparkSQL和Hive進行差別依賴的驗證，比較不同數據量下的時間。此處三角分布的邊長取l=14。

需要注意的是，由于Hive的運行時間過長，圖4中使用了對數縱坐標來顯示運行時間，橫坐標為實驗數據集的元組個數。當數據量為20萬行時，Hive的執行時間已經超過了7個小時，由此省略了40萬行時Hive的運行結果。從圖4可知，三角算法的時間非?？欤荢parkSQL和Hive時間的1/10～1/100。當數據量達到40萬行時，SparkSQL的耗時已經達到了近2個小時，而三角分布只需要不到5分鐘，時間優勢非常明顯。

圖4 實驗1的實驗結果

3.2 不同邊長下性能比較

以隨機三角分布的邊長和數據量作為變量，比較各情況下檢驗差別依賴所需要的時間。

觀察圖5中同一條曲線，當邊長一定的時候，reducer數目不變，顯然數據量的增加會導致總時間增加。另一方面，在同數據量的情況下，隨著邊長的增加，參與的reducer數目會增加，時間隨之下降，這和預期一致。

圖5 實驗2的實驗結果

需要注意reducer數目的增加并不會使得總時間一直下降。從實驗結果看，當reducer數目已經較大時，再增加其數目(即增加邊長)對繼續縮短時間的幫助變得不再明顯。這是因為中間的Shuffle過程是需要有實際的數據傳輸的。若過度增加邊長和reducer數目，會造成數據傳輸量增大，而每個reducer上的工作減少卻不明顯，這可能反而會造成總時間的增加。

3.3 兩種三角分布算法性能比較

排序三角算法相較于隨機三角算法，可以在邊長相同的情況下省去部分的reducer，或者可以增加reducer利用率，減少時間，但這兩點都建立在數據分布較為平均的情況下?，F對于兩個數據集，進行兩種算法的性能比較。

3.3.1 邊長相同時reducer的減少

當邊長固定時，隨機三角算法的reducer數目是不變的，而排序三角算法需要的reducer數目可以通過計算得出。以邊長l=10為例，計算得出θ取不同運算符時的reducer數量，和隨機三角算法進行比較，如表2所示。

表2 reducer數目計算結果

分析表2得出，使用優化算法，對于同一個差別依賴，其使用的reducer數目得到了顯著的下降，特別是在取值為=的時候。實際上，reducer可減少的數目與常數值和該屬性的取值范圍有很大關系：當常數值接近于屬性值取值范圍的中間值時，不論θ的取值多少，都可以減少很大占比的reducer數目；而當常數值接近于屬性值的一端時，至少也有兩個θ的取值可以使得reducer數目減少很多。

3.3.2 時間上的比較

排序三角算法相較于隨機三角算法，依賴于原始數據的具體分布。雖然三角邊長可以增大，但若原始數據分布不均勻，會使得某些機器上的比較次數變多，進而影響總體運行時間。本次實驗的數據集為均勻分布數據集Data1和正態分布數據集Data2，行數為20萬行；隨機三角算法的邊長l=10，與排序三角算法在θ的不同取值下進行比較，如表3所示。

表3 時間比較結果

由表3可得，在均勻分布的數據集上，排序三角分布算法非常出色，可以更加高效地利用所有的reducer進行計算。而在正態分布的數據集上，當運算符為<和>時，時間和隨機分布的耗時差距不大。這說明排序三角算法相對來說依賴于初始數據和初始條件，以及此時的方差和的取值是采取何種三角分布算法的分界點。

對于排序三角算法，若數據分布不均勻，會導致部分reducer上的工作量超過平均值，而MapReduce算法的時間，依賴于所有reducer中最晚結束的時間，所以非平均數據集的排序三角算法的時間會長于使用相同數量reducer的隨機三角算法的時間。對于一個數據集更適合于何種三角分布算法，可以先通過抽樣來對數據分布進行分析。

另外，對比表2和表3可以得出，即使是均勻分布的數據，時間減少的百分比，也不如上一個實驗中預估reducer減少的百分比。這主要是因為在之前的分析中，部分因素被忽略了：一部分原因是在Shuffle步驟的耗時是固定的(大致只與元組數量有關)，這部分時間無法按比例降低；另一方面，邊長增加后，由于不為整數，所以也無法做到將所有的有效比較完全均勻分配到每一個reducer中，依舊會存在部分的冗余比較無法避免。

4 結 語

差別依賴用于描述數據表中元組對間屬性變化量之間的關系。差別依賴的約束驗證需進行元組的兩兩比較，通過引入分布式計算技術可以有效提高該過程的可用性。本文針對該問題進行研究，在分布式系統上提出了隨機三角分布的分布式算法，實現了差別依賴的大數據檢驗。本文提出了排序三角分布，在均勻分布的數據集上，能夠更快速高效地完成約束驗證。通過實驗，兩個算法的時間優勢非常明顯。

在未來的工作中，本文將考慮多差別依賴的分布式檢測算法。針對多依賴，通過合并檢測的方式以減少重復的工作量。另外，經分析，數據傳輸量這一要素相對被忽略。在Spark等并行處理的系統中，Shuffle步驟會進行數據的傳輸，數據傳輸量在reducer間的分派情況也會實際影響到最終的總計算時間。我們將由此探討邊長、數據轉移量以及時間上的關系，以期進一步進行算法優化。