? ? 今天要講的文章是SIGMOD 2010年的一篇文章，Pregel: A System for Large- Scale Graph Processing。本文主要想解決的問題就是：隨著如今技術的發展，圖數據規模是不斷地增長的。現有的圖處理系統采用單機處理大圖數據，但是單機處理大圖數據存在很差的可擴展性，因為單機機器內存是由限制的。然而使用MapReduce處理大圖數據，效率又很差。因為每個階段中必須存儲圖狀態，這樣在每個階段中會造成大量的磁盤IO和網絡通信帶來的開銷。所以本文就針對現有的圖處理系統這種狀況，提出了基于BSP的大規模圖處理系統：Pregel。

1.Motivation

????圖算法常常表現出比較差的內存訪問局部性，針對單個頂點的處理工作過少，以及計算過程中伴隨著的并行度的改變等問題。分布式的介入更是加劇了locality的問題，并且增加了在計算過程中機器發生故障的概率。盡管大型圖對象無處不在，及其在商業上的重要性，但是據我們所知，目前還不存在一種在大規模分布式環境下，可以基于各種圖表示方法來實現任意圖算法的，可擴展的通用系統。如圖所示：

當今現有的圖計算存在的問題：

1.現有的圖處理系統不能滿足需求 重新寫一個基礎設施，需要大量的工程性。
2.使用MapReduce算法。沒有效率，因為必須存儲圖狀態到每個階段中來，這樣會造成大量的磁盤IO和網絡通信帶來的開銷在每個階段中。
3.使用單機圖處理庫 scalable不好
4.使用現有的并行圖處理系統 沒有很好的容錯

????所以說就需要一個可以伸縮的、容錯的解決方案：Pregel: A System for Large- Scale Graph Processing，并且Pregel采用一種同步批量模型。

1.Scalable and Fault-tolerant platform
2.API with flexibility to express arbitrary algorithm
3.Inspired by Valiant’s Bulk Synchronous Parallel model[4]

4.Vertex centric computation (Think like a vertex)

2.Pregel的計算模型

首先圖數據從文件中輸入數據，經過若干次SuperSteps后，也就是若干輪迭代之后，最終數據將會寫入到文件中。每一輪迭代都叫做一個Superstep。輸入數據經過compute之后進入下一輪迭代。并且每輪迭代計算都需要同步等待其他數據都計算完成后才能進入下一輪迭代。
BSP計算模型由通過通信網絡連接的處理器組成。 每個處理器都有一個快速的本地內存，并可能遵循不同的計算線程。 BSP計算以一系列全局超級步驟進行。 一個superstep包含三個組件：
并發計算：在每個參與處理器上進行幾次計算。 每個進程只使用存儲在處理器本地存儲器中的值。 這些計算是獨立的，它們是異步發生的。
通信：進程之間交換數據。 這種交換采取單方面的形式，并獲得通話，而不是雙方發送和接收電話。
屏障同步：當一個進程到達這個點（屏障），它等待，直到所有其他進程已經完成他們的通信行為。
計算和溝通行為不必按時排序。 屏障同步然后結束本輪superstep。
Pregel計算和通信不必按時排序，當一個任務的計算完成之后，這個任務產生的消息就能馬上發送到相應的任務中去。Pregel消息采用一種Message-Passing的方式。每個頂點接受上一個superstep產生的消息，每個頂點程序執行相同的用戶頂i有的函數，然后修改它的頂點值和出邊的值。接著頂點程序就會發送消息給其他頂點（這個消息會在下一個superstep中被處理）。每個頂點程序都有激活和不激活狀態之分，每個頂點程序在執行一輪頂點迭代后，會經過判斷是否在下一輪迭代中變成激活或者不激活狀態。每個頂點程序就是一個狀態機，如下圖所示：

????? ? 當整體圖數據中，所有的頂點都變為不激活狀態后，這個圖迭代計算才計算完成。

3.?Differences from MapReduce

Pregel?
1.保持每臺機器中的頂點和邊進行計算
2.僅將網絡傳輸用于消息
MapReduce
1.將圖形的整個狀態從一個階段傳遞到下一個階段
2.需要協調一個鏈接的MapReduce的步驟

4.Program API

Pregel 頂點程序定義的編程接口

????Pregel 程序Example：SSSP

4. System Architecture

????? ? Pregel 采用的也是一種Master& Worker的模型。Master用來協調各個Worker工作，并且當集群發生宕機的情況下，Master還可以調度Worker，使得Worker能夠從錯誤中恢復出來。Worker就是用來處理各個Task任務，并且每輪迭代的通信都是在不同的Worke之間進行通信。

????? ? Pregel將圖數據放在分布式存儲系統中（比如GFS或者BigTable），臨時數據存儲在本地磁盤中。

4.1Pregel執行步驟

1. Pregel初始化過程 Master將會復制很多個點程序執行到各個集群上面去

2. Master 初始化每個Worker分區的個數和Grpah分區個數，并將圖graph劃分到每個worker中一個或者多個分區上
3. DataInput 數據輸入?
Master進程為每個worker分配用戶輸入中的一部分，這些輸入被看做是一系列記錄的集合，每一條記錄都包含任意數目的頂點和邊。并且每個worker加載這些點集并把它標記為active 并且更新woker上的數據結構

4. Master指示每個woker執行一個superstep階段 worker輪詢在其之上的頂點，會為每個partition啟動一個線程。調用每個active頂點的Compute()函數，傳遞給它從上一次超級步發送來的消息。消息是被異步發送的，這是為了使得計算和通信可以并行，以及進行batching，但是消息的發送會在本次superstep結束前完成。 每個step會不斷重復該步驟，只要有頂點還處在active狀態，或者還有消息在傳輸。 5. 在每個點program計算都終止后，Master會指示每個worker去保存一部分執行結果

5.?Fault Tolerance

5.1Checkpoints

Pregel容錯采用一種Checkpoints機制，在每次進行superstep開始的時候，the master會指示每個workers去保存每個分區的狀態到持久化系統中去。這個分區的狀態包括：點的values 邊的values 和incoming messages。然后Master自己也會保存aggregator的值。 當一個或多個worker發生故障，被分配到這些worker的partitions的當前狀態信息就丟失了。Master重新分配圖的partition到當前可用的worker集合上，所有的partition會從最近的某超級步S開始時寫出的checkpoint中重新加載狀態信息。該超級步可能比在失敗的worker上最后運行的超級步 S’早好幾個階段，此時失去的幾個superstep將需要被重新執行我們對checkpoint頻率的選擇基于某個故障模型[13]的平均時間，以平衡checkpoint的開銷和恢復執行的開銷。

5.2 Log outgoing Message

除了基本的checkpoint，worker同時還會將其在加載圖的過程中和superstep中發送出去的消息(來自它們已經分配后的分區)寫入日志。這樣恢復就會被限制在丟掉的那些 partitions上。它們會首先通過checkpoint進行恢復，然后系統會通過回放來自正常的partitions的記入日志的消息以及恢復過來的partitions重新生成的消息，更新狀態到S’階段。這種方式通過只對丟失的partitions進行重新計算節省了在恢復時消耗的計算資源，同時由于每個worker只需要恢復很少的partitions，減少了恢復時的延遲。

6.Aggregator

? ? ????Pregel提供一種全局通信的機制。用于全局通信，全局數據和監測，從頂點報告的值計算匯總統計信息。在superstep期間，每個工作人員從其頂點聚合值以形成部分聚合的值，在superstep結束時，來自每個worker的部分聚合值被聚合在一個樹形結構中。并且樹結構允許并行化，全局聚合被發送到Master節點上。Master在下一個superstep開始時將全局值發送給所有Worker。

7. Experiments

總結

以上是生活随笔為你收集整理的[SIGMOD 10] Pregel 基于BSP的大规模图处理系统 学习总结的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。