Spark實現離線不相交集計算

• 2020 年 4 月 8 日
• 筆記

最近在開發一個ID Mapping業務系統——識別數據上報中社交帳號的關聯關係，找到系統中哪些社交帳號屬於現實世界中的同一個人。簡單來講，如果同一條上報數據中出現了兩個社交帳號（比如一個手機號和一個QQ號），就認為這兩個社交帳號在現實世界屬於同一個人。那麼，如何計算這個關聯關係呢？

一開始我們解決這個問題的思路很直接：現實世界的每個人在系統中用唯一的UUID標識，每次社交帳號（Account）上報，在Redis中記錄一個UUID->Account的正向索引，同時記錄一個Account->UUID的反向索引。每個UUID下可能有多個社交帳號，每個社交帳號只會歸屬於一個UUID。每次收到2個社交帳號相關的上報時，先通過Account->UUID的反向索引查到這兩個社交帳號對應的UUID，如果兩個帳號分別屬於兩個不同的UUID，就把這兩個UUID合併為一個新的UUID，同時原來歸屬於這兩個UUID的所有社交帳號都歸屬於新的UUID；如果查到一個UUID，那麼把這兩個帳號歸屬於這個UUID；如果未查到UUID，則新生成一個UUID，最後把新的正向索引和反向索引再寫回Redis即可。

當然實際的ID Mapping可能有更複雜的關聯關係，並不是兩個社交帳號在同一條上報數據中出現這麼簡單的邏輯，這就不在本文討論的範圍內了。

這種方式乍一看也沒什麼問題，但仔細想來卻忽略了一個很基本的問題：這種方式只能不斷的將帳號關聯在一起，而不能解除關聯。舉例來說，給定A-B、B-C兩組社交帳號的關聯關係，A、B、C應該全部關聯在一個UUID下，此時若B點被刪除，或B-C關聯關係解除，系統無法將A和C解除關聯。而這種解除關聯的場景在業務系統中也是很常見的，比如在企業的客戶管理系統中，往往會記錄一個客戶的一些社交帳號，有時企業會刪除客戶的一些社交帳號，甚至刪除某一個客戶。

其實這個業務抽象成一個數據結構的問題，就是典型的不相交集問題。每一個社交帳號看做一個點，兩個社交帳號的關聯關係看做兩點之間的無向邊，上面的問題就變成了在一個圖裡劃分不相交集的問題。既然很難處理動態變化的圖，就每天批量計算一下某一時刻所有帳號的關聯關係吧。本文就是要介紹一下如何用Spark的RDD API實現靜態圖不相交集的計算。

一、問題定義

前言描述了問題的背景，這裡再明確定義下本文要解決的問題。

演算法輸入是一張離線Hive表，每行有兩個欄位P_i、P_j，表示無向圖中節點P_i和P_j之間存在一條邊。輸出同樣得到一張Hive表，每行也是P_i、P_j兩個欄位，P_j均表示不相交集的根節點，即所有P_j相同的行中P_i的集合加上P_j就構成了與圖中其他點不相關聯的一個獨立集合。如圖一所示，對於左側的輸入，計算結果將得到右側的輸出，可以看出P1、P2、P3、P6、P8、P9構成一個獨立集合，P4、P5、P7構成另一個獨立集合。

二、計算過程

1. 使每行數據中P_i > P_j

為了保證迭代過程最終可以收斂，不妨將圖中所有邊都當做有向邊處理，方向都是節點ID較大的節點指向節點ID較小的節點，這樣最終計算得到的不相交集必是以集合中ID最小的點為根，即所有節點都指向所在集合中ID最小的點。因此，不妨將原始數據中的每一行當做由P_i指向P_j的有向邊，若P_j>P_i，則交換P_i和P_j。如圖二所示，這一步修改了第三行和第五行數據。

edge_rdd = edge_rdd.map(row => if (row._1 > row._2) (row._1, row._2) else (row._2. row._1))

2. 保證Pi不重複

第一步處理完之後，原始數據中還會存在一個問題：多條邊相交於一個ID較大的節點，這會導致ID較大的節點成為潛在的根節點。解決這個問題需要將局部相交於ID較大節點的邊轉化為相交於ID最小的節點。比如存在P6->P1和P6->P3兩條邊，這兩條邊交於P6，P1、P3、P6組成一個獨立集合。需要將這個關聯關係轉換為P6->P1和P3->P1兩條邊，即以P1為根節點。前一步的處理已經保證數據中每一行都滿足P_i>P_j，因此多條邊交於ID較大的節點等價於多行的P_i相同。所以只需要在保證原有關聯關係的條件下將表處理為P_i不重複即可。

這一步保證了P_i中的節點不會作為根節點，所有有向邊都由葉子節點指向潛在的根節點。

在演算法實現上，首先將P_i相同的行聚合在一起，然後輸出這個集合中每個點指向集合中ID最小點的有向邊。

while(edge_rdd.keys.count() != edge_rdd.keys.distinct.count()) {      edge_rdd = edge_rdd.groupbykey()          .flatMap(row => {              val vertex_list = row._2.toSeq.sorted.distinct              val result = new ArrayBuffer[(String, String)]()              result.append(row => (row._1, vertex_list.head))              vertex_list.tail.foreach(vertex => result.append((vertex, vertex_list.head)))              result          })  }

從圖三種可以看出，每次迭代之後，可能會產生新的P_i相同的數據，因此需要用迭代的方式，多次執行以上邏輯，迭代的終止條件就是P_i中的點不重複。

3. 將每一行中的P_j替換為集合中最小的節點ID

最後一步就是演算法的核心，通過自關聯，將所有葉子節點關聯到根節點上。演算法原理很好理解：若同時存在P_z->P_y的有向邊和P_y->P_x的有向邊，就將P_z->P_y替換為P_z->P_x。

如圖四所示，第一次迭代由P8->P2和P2->P1兩條邊得到P8->P1，由P9->P8和P8->P2兩條邊得到P9->P2；第二次迭代由P9->P2和P2->P1兩條得到P9->P1。

如果我們定義P2與根節點P1的距離是1，P8和根節點P1的距離是2，P9和根節點P1的距離是3，依次類推。該演算法每n次迭代可以將距離為2^n之內的節點全部關聯到根節點上。

for (i <- 1 to iterateNum) {      val inverse_edge = edge_rdd.map(row => (row._2, row._1))      val edge_rdd = inverse_edge.leftOuterJoin(edge_rdd).map(row => (row._2._1, if (row._2._2.isDefined) row._2._2.get else row._1))  }

通過以上3個步驟的處理，可以看到原始數據集被劃分成了2個不相交集，根節點分別為P1和P4。

三、執行優化

問題也並不是這樣順利的就解決了，將上述邏輯轉化為工程程式碼時還遇到了一些其他問題，下面也分享下遇到的問題，以及採取的優化方案。

1. 推測執行

由於數據量比較大，計算時會使用較大的並發來執行，對於同一個job經常出現大部分task都執行很快，只是在等2-3個task的執行，而觀察發現並沒有明顯的數據傾斜。查閱相關資料了解到這種情況可能是由於各個executor上的運行環境不同（有可能同時運行了其他任務、或者是硬體原因），導致計算時間差異較大，這個問題可以通過推測執行解決。

spark.speculation=true  spark.speculation.interval=100  spark.speculation.multiplier=1.5

2. checkpoint

由於該演算法是一個迭代演算法，執行流程較長，有時會出現某個executor丟失，使RDD中的一部分丟失，導致整個任務需要重新計算，甚至失敗。查閱相關資料後，最終通過checkpoint的方式解決了這個問題。checkpoint運算元將RDD寫入到HDFS某個目錄下，因此也一個Action，所以一般會先執行cache再執行checkpoint。

edge_rdd = edge_rdd.groupbykey().flatMap(row => {...}).cache  edge_rdd.checkpoint()

3. RDD cache釋放

程式碼運行過程中還發現任務會佔用很多記憶體，遠比預期大的多，通過查看Spark任務的Storage頁，發現其實是迭代的方式導致了「記憶體泄漏」。

在迭代的過程中，演算法對每一次迭代得到的edge_rdd進行了cache，而事實上每次計算出新的edge_rdd後，前一次迭代的cache就沒用了。但如果沒有手動釋放的話，這些RDD的cache在任務終止之前都不會被釋放掉，會一直佔用著集群記憶體，導致「記憶體泄漏」。從圖五中可以看出，每次迭代都會生成一個RDD，並cache在記憶體中，如果迭代次數比較多，這部分記憶體浪費對集群資源的佔用就很可觀了。甚至如果新的RDD沒有記憶體可以cache，會導致RDD的重複計算，這樣會嚴重影響任務執行的時間。

這個問題可以通過在每次計算生成新的RDD時手動unpersist上一個RDD來解決，在記憶體無效時立刻釋放掉這部分記憶體。

val tmp = edge_rdd.groupbykey().flatMap(row => {...}).cache  tmp.checkpoint()  edge_rdd.unpersist()  edge_rdd = tmp

四、執行性能

用上述演算法計算業務收集到的社交帳號關聯關係，數據量在5000萬條左右，第二步計算需要7次迭代收斂，第三步計算需要3-4次迭代收斂。程式運行使用16核64G記憶體的分散式Spark運行環境，迭代過程中partition個數為64，整體運行時間在20分鐘左右，基本達到了業務使用的要求。