圖解分散式圖演算法 Pregel: 模型簡介與實戰案例

03-31

本文為原創文章，轉載請著明出處。
作者： @張茄子
原文：圖解圖演算法 Pregel: 模型簡介與實戰案例
版權：CC BY-NC-SA 3.0 許可協議

這篇文章是對之前在 SHLUG 月度分享活動上所作演講 Pregel in Graphs 的總結。為使分享內容清晰易懂，本人繪製了大量原創示意圖，這篇文字版的總結也會盡量以這些圖示為主。除了對 Pregel 演算法的簡單介紹，本文還附加了一個用戶追蹤畫像的實戰案例，用以證明圖計算模型的重要意義。

Pregel 簡介

Pregel 是 Google 自 2009 年開始對外公開的圖計算演算法和系統，主要用於解決無法在單機環境下計算的大規模圖論計算問題。與其說 Pregel 是圖計算的演算法，不如說它是一系列演算法、模型和系統設計組合在一起形成的一套圖模型處理方案。

圖計算的實際應用非常廣泛，因此自 Pregel 公開之後，一些開源的方案也被實現出來，比如說來自 Spark 的 GraphX 就實現了 PregelAPI。值得注意的是，Pregel 作為一個近十年前起就為人所知的演算法，雖然新近也已經提出了不少增強和改進的技術，但在現實場景下仍然是很有生命力的。

本文假定讀者有基本的圖論知識。

模型簡介

Pregel 模型的基本要素

上圖展示了 Pregel 計算模型的基本要素，主要包括:

節點(Vertex)。在 Pregel 中，每個節點都有全局唯一的 ID
邊(Edge)。在 Pregel 中，每個邊可以被 Assign 一個屬性，這個屬性可以是邊的權值等信息
消息(Message)。消息是 Pregel 計算模型的核心。每個 Vertex 在初始狀態以及之後的每一個計算步驟當中都被 Attach 一個 Message 值作為 Vertex 當前的狀態，演算法的迭代通過 Vertex 之間互相發送的消息來完成
超迭代(Superstep)。一個 Superstep 是 Pregel 在執行演算法過程當中進行的一次迭代。一次 Pregel 計算過程可能包括多個 Superstep

在 Pregel 當中，Edge 一般是有向的。同時節點 Vertex 還存在 Active 和 Inactive 兩種狀態。之後可以看到，節點的狀態將會決定一些演算法是否結束。

Pregel 演算法以 Vertex 為中心

Pregel 的圖計算過程與 MapReduce 非常接近：在迭代的每一個步驟當中，將會以圖的節點為中心進行 Map 操作，這意味著在 Map 函數當中我們只有以某一節點為中心的局部信息，這包括:

一個 Vertex 和它當前 Attach 的 Message
當前 Vertex 向外指向的 Edge 和 Edge 上的屬性
當前 Vertex 在上一步計算當中所接收到的全部 Message

對於圖中的每一個 Vertex，在 Pregel 當中的一次 Superstep 包括接收消息、合併消息和發送消息三個步驟。上圖當中，節點 N1 接收到了兩條 Message ，加上自己原有的 Message，合併出一個最大值，在最後它會把這個值以消息的形式發送給與之相鄰的 Vertex。

Vertex 的狀態變化

前面提到 Vertex 會有狀態變化，這個概念也十分簡單:

當一個 Vertex 在上一步當中沒有接收到消息，或者演算法自己決定不再向外發送消息，它可以被轉變為 Inactive 的。在 Pregel 的術語當中，這被稱為 Vote to halt
當一個在之前已經 Inactive 的 Vertex 又接受到一條新的消息，它會在新的計算中轉變為 Active 的狀態

在大多數演算法當中，所有的 Vertex 都進入 Inactive 狀態就意味著演算法結束。

下圖給出了節點之間傳遞消息的一個示意，此時 Pregel 還面臨一個比較重要的問題：通過網路發送大量 Message 的成本較高。

通過網路傳遞 Message

在有些情況下，我們可以在 Message 發送前先對他們進行一步聚合。比方說，在演算法中我們只關心接收到消息的最大值，那麼與其把所有消息都發送到目的地再計算，不如先將最大值求出，這樣可以極大地減少需要發送的消息數量。 Pregel 允許用戶自定義 Combiner 來實現這一目的。

下圖展示了使用 Combiner 聚合左邊節點發送的消息的最大值 4 ，之後再發往目標節點的過程。

使用 Combiner 實現預先聚合

Pregel 需要解決的另一個問題是部分圖論演算法無法使用上述 Vertex 狀態來判斷是否結束。有些時候我們可能需要全圖的所有節點共同提供一些信息，統計出一些指標來進行判斷。在另一些情況下，我們也希望對演算法的進展進行衡量和追蹤。因此，Pregel 還引入了 Aggregator。

使用 Aggregator 跟蹤全局信息

最後一個需要解決的問題是改變拓撲的問題。有些圖演算法在迭代過程中需要增刪節點和邊。 Pregel 並沒有中心服務掌控整個圖的狀態，這一需求也被抽象為 Message 發送機製得以解決。

下圖中節點 N1 向節點 N2 發送了刪除 E1 的指令和向節點 N3 發送了刪除節點的指令。當 N3 被刪除之後，其向外的邊也都會被一併刪除。

通過發送特殊的消息實現圖拓撲的修改

為了防止接收到的拓撲修改的消息相互衝突，這些消息會按照一定的順序被應用，用戶也可以定義函數用來進行衝突處理。

系統設計

Pregel 的計算模型不單單只有上面介紹的抽象模型而已，為了能在大規模分散式環境下執行這一演算法， Pregel 還包含系統設計上的具體考量，比較重要的四條是:

將圖分區到不同機器進行計算
使用主從模型進行任務調度和管理
使用 Message 緩衝近一步提高通訊吞吐量
使用 Checkpoint 和 Confined Recovery 實現容錯性

圖分區

圖分區的方法十分容易理解，前文提到 Pregel 是以 Vertex 為中心的計算模型，因此在分區的時候也是以 Vertex 為中心。當一個節點被劃分到一個區，與之相連的局部信息(邊、邊屬性、消息)也都會被分配到這個區上。由於對圖進行分區的函數是全局一致的，各個計算節點對消息的轉發並不需要通過某一中心服務進行協調。

默認的分區方法就是對 VertexId 的 Hash 值進行取模操作。用戶也可以自定義分區函數以增強分區的局部性(Locality)，減少計算節點之間的網路流量。

主從系統架構

Pregel 在分散式系統當中的任務調度是簡單的主從模型。每個計算任務有一個 Master 進程協調所有計算，在每個 Superstep 當中，Master 會決定圖分區、發送 RPC 調用到 Worker 節點激發任務以及監控任務完成。所有圖分區都在 Worker 上，Master 不管理任何圖分區。不過，Pregel 的 Aggregator 運行在 Master 上。因此 Worker 需要將 Aggregator 所需信息發送到 Master 上進行聚合。

Message 緩衝

Message 緩衝是在計算節點(Worker)的層面上提高吞吐量的一個優化。 Message 在 Worker 之間傳遞時並不是來一個發一個，而是通過緩衝積攢一些 Message，之後以 Batch 的形式批量發送。這一優化可以減少網路請求的 Overhead。

Pregel 的容錯方法

Pregel 使用兩種方法來實現容錯性:

Checkpoint 在 Superstep 執行前進行，用來保存當前系統的狀態。當某一圖分區計算失敗但 Worker 仍然可用時，可以從 Checkpoint 執行快速恢復
當某一 Worker 整體失敗當機使得它所記錄的全部狀態丟失時，新啟動的 Worker 可能要重新接收上一步發送出來的消息。為了避免無限制的遞歸重新計算之前的步驟，Pregel 將 Worker 將要發送的每一條消息寫入 Write Ahead Log。這種方法被稱為 Confined Recovery

可以看到，Confined Recovery 的思想和 Spark 等基於 DAG 的計算模型有很多相似的地方。

演算法案例

為了使大家對 Pregel 具有更加直觀的認識，這一章將介紹一些抽象的演算法在 Pregel 之下的運行過程。 對實戰案例更感興趣的朋友可以直接跳到下一章。

計算連通分量

計算連通分量的經典單機演算法是並查集，在 Pregel 中，這一問題通過發送消息的方法解決。

Pregel 計算連通分量

上圖提供了 Pregel 計算連通分量的簡單過程，其步驟是:

為每個節點初始化一個唯一的 Message 值作為初始值
在每一個步驟當中，一個 Vertex 將其本身和接收到的 Message 聚合為它們之中的最大值(最小值)
如果 Attach 在某一個 Vertex 上的 Message 在上一步當中變大(變小)了，它就會把新的值發送給所有相鄰的節點，否則它會執行 Vote to halt 來 Inactivate 自己
演算法一直執行直到所有節點都 Inactive 為止

上述連通分量的演算法假定邊都是雙向的(可以通過兩條相反的邊實現)。可以想像，由於同一連通分量當中的節點都可以互相傳播消息，因此最終在同一個連通分量里的 Vertex，必定都會擁有這一連通分量內 Message 的最大值(最小值)。這個最後的值就可以作為這一連通分量的 Identifier。

值得注意的是，Pregel 實現的連通分量演算法在超大規模的圖上仍然有可擴展性的瓶頸， Google 之後發表了論文 Connected Components in MapReduce and Beyond 加以解決。

計算單源最短路

最短路的經典單機演算法包括 Dijkstra、 Bellman-Ford 等。在圖規模巨大無法被放入單機內存的場景下， Pregel 的消息傳遞模型仍然適用。具體的方法是:

初始化 Vertex 的 Message 為 INF (無窮大)
將源點的 Message 設為 0
每次每個節點將自己目前的 Message 加上邊的權值發送到相鄰節點，每個節點聚合出自身所有消息的最小值
當某一步當中一個節點的 Message 值沒有變小，這個節點 Vote to halt
當所有節點都 Inactive 時，演算法結束

Pregel 計算最短路: 步驟1

上圖展示了一個簡單的圖最短路的第一步計算步驟，在這一步中，N2、N3 接收到 N1 發送的消息，從而更新了自己的消息為更小的值。執行結束後，N1 因為沒有變化而 Inactive。

Pregel 計算最短路: 步驟2

在這一步中，N2 和 N3 互相發送消息，由於 N1 -> N3 -> N2 的路徑更短，N3 的 Message 被更新， N2 則變為 Inactive。

Pregel 計算最短路: 步驟3

儘管這時所有的最短距離已經求出，由於 N2 在上一步仍然是 Active 的狀態，它將會向 N3 發送最後一次消息。由於 N3 沒有更新自己的值，此時圖中三個節點都變為 Inactive，演算法結束。

PageRank

PageRank 可能是 Pregel 最典型的應用案例。因為它本身的原理就是通過不斷將自身的權值以消息的形式發送出去而完成計算的。

這裡拿 PageRank 作為典型案例的原因在於: PageRank 不能簡單的以 Vertex 的狀態作為演算法終止的條件。除了設定固定的迭代次數之外，另一個方法就是利用 Pregel 的 Aggregator 來跟蹤計算過程。

關於 PageRank 的消息傳遞的公式這裡不再贅述，下圖給出了在一個簡單的圖上進行計算過程的前半部分:

PageRank：前半部分

在上圖中，N1 不斷地將權值發送給自己和 N2，而 N2 則只會發給自己。隨著逐步的迭代，N1 的權值越來越小，N2 的權值越來越大。我們用 Aggregator 跟蹤每個節點上權值變化絕對值的最大值 delta，這個最大值會隨著迭代的進程而越變越小。

如果我們設定 delta 小於 0.05 時演算法結束，則我們將在接下來的步驟中看到如下過程:

PageRank：後半部分

在最後 Aggregator 得到 delta 滿足終止條件之後，Master 就可以結束演算法了。

實戰案例: 用戶追蹤畫像

Pregel 演算法的普適性非常強，幾乎可以應用各類經典圖論演算法，然而我們實際使用它時，主要看中的是它可以解決規模超大的圖計算問題。在這一章我們研究一個實戰案例以說明其應用價值。

用戶追蹤畫像問題

上圖是我們實戰案例的問題描述，具體來說：

我們的網站可以收集到各個用戶一系列離散的訪問請求
這些請求可以被以 Session 形式組織起來
在一個 Session 的請求當中，我們可能收集到用戶各種信息的一個子集
我們希望通過一個用戶的多個 Session，儘可能聚合到這個用戶完整的信息

上面的第三條是我們達成目標的最大阻礙，其原因在於，為了最好的用戶體驗，我們不希望在一開始就強迫用戶登錄或提供定位許可權以獲得他的全部信息。設想有一個用戶，平時一直匿名訪問我們的網站，當有一天他終於認為我們的網站很有價值，決定註冊登陸我們的網站，這時一個典型的需求就是將其之前匿名的訪問記錄和之後登錄的用戶聯繫起來，從而獲得更加有用的信息。

用戶追蹤畫像的基本過程可以分為多個步驟，下圖展示了對相互穿插的離散的用戶信息的初步處理。

用戶離散訪問信息的初步處理

這一步處理的主要目的是將離散的請求組織成 Session 的形式並提取出特徵。在這一步之後，我們可以得到很多相互獨立的 Session，但是我們仍然不知道哪些 Session 是屬於同一個用戶的。但是依靠每個 Session 之中提取出來的特徵，發現 Session 之間相互的關係是可能的。如果我們將 Session 作為圖的節點，將相互匹配的 Session 用邊連接起來，就可以把用戶追蹤畫像的問題轉換為尋找無向圖連通分量的問題。

通過計算無向圖連通分量聚合用戶信息

通過聯繫和聚合多個 Session 包含的特徵，我們得到了比僅著眼單個 Session 更全面的用戶信息，這樣用戶追蹤和畫像的問題就解決了。由於網站的用戶訪問可能是海量的，這一問題可能必須交由大規模分散式系統進行計算，這即是 Pregel 的用武之處。

在這裡有一點擴展優化。設想我們有 $O(N)$ 個 Session，要找出它們兩兩之間的關係，需要耗費的是 $O(N^2)$ 的時間，這對於僅有 $10^6$ 個 Session 的規模都是相當不可行的。我所想到的一種優化是：

只選擇 $n$ 個重要的特徵作為參考
定義兩個 Session 是匹配的，當有大於等於 $x$ 個特徵相匹配
對於收集到的一個 Session 的特徵，枚舉它恰好 $x$ 個特徵的子集合
匹配這些包含恰好 $x$ 個特徵的子集合，而不是 Session 本身

下圖展示了這個過程:

枚舉固定數量特徵子集合進行匹配

假設 $n=5$ 且 $x=2$ ，也就是說對於任意兩個 Session，我們只考慮它們所擁有的 5 個固定的特徵，當等於或大於 2 個特徵匹配，我們就認為兩個 Session 匹配。如果我們有 $O(N)$ 個 Session，因 $mathtt{C}^2_5=10$ 那麼我們將會得到 $O(10×N)$ 個這樣的子集合。

這麼做難道不是使得問題的規模上升了么？為什麼我們可以算的更快呢？這是因為這些子集合都恰好包含兩個元素，假如我們對這些子集合進行排序，相同的子集合一定排在一起，這樣我們就可以通過在排好序的序列當中，鏈接相鄰相同子集合所代表的 Session，快速將所有匹配的 Session 連接起來。

對 $O(10 imes N)$ 個元素進行排序，將會花費 $O(10 imes Ncdotlog(10 imes N))$ ，最後處理整個序列還需要 $O(10 imes N)$ 的時間。但是總體的時間複雜度卻降低到了 $O(Nlog N)$ 的級別，對於 $O(N^2)$ 來說是一個巨大的進步。

換句話說，我們利用類似在終端中使用 sort | uniq 的方式，降低了尋找 Session 匹配的複雜度。

總結

本文簡要介紹了 Pregel 的基本模型和系統設計，給出了一些簡單演算法案例的執行過程。在最後使用用戶追蹤畫像這樣的實戰案例，說明了圖計算在現實世界當中的具體應用。

隨著我們需要處理的問題規模越來越大，像 Pregel 這樣的分散式計算模型的應用價值也會越來越高，甚至有可能會逐漸超過傳統的演算法。

然而，在實戰案例的擴展優化當中可以看到，那些傳統和經典的演算法技巧，在大規模分散式數據處理當中仍然可以發揮啟發性的作用。我認為，無論統治未來是 AI 還是區塊鏈，這些基礎的演算法技能都是非常重要的。