作者：Grzegorz Malewicz, Matthew H. Austern .etc.Google Inc 2010-6

原文：http://people./~edwardyoon/documents/pregel.pdf

譯者：phylips@bmy 2012-09-14

譯文：http://duanple.blog.163.com/blog/static/70971767201281610126277/

[說明：Pregel這篇是發(fā)表在2010年的SIGMOD上，Pregel這個名稱是為了紀念歐拉，在他提出的格尼斯堡七橋問題中，那些橋所在的河就叫Pregel。最初是為了解決PageRank計算問題，由于MapReduce并不適于這種場景，所以需要發(fā)展新的計算模型去完成這項計算任務，在這個過程中逐步提煉出一個通用的圖計算框架，并用來解決更多的問題。核心思想源自BSP模型，這個就更早了，是在上世紀80年代由Leslie Valiant(2010年圖靈獎得主)提出，之后在1990的Communications of the ACM 上，正式發(fā)表了題為A bridging model for parallel computation的文章。目前實際上已經(jīng)有針對Pregel這篇文章的翻譯版本了，不過只翻譯了出了前半部分關(guān)于Pregel的設(shè)計與實現(xiàn)部分。其實后半部分也很重要，有助于理解整個圖計算的歷史背景，以及Pregel本身的性能和項目本身的演化等，另外最近越來越多的人開始關(guān)注這一文章，所以還是抽出時間重新閱讀了一遍，并重新翻譯出來，以供參考]

很多現(xiàn)實中的計算問題都會涉及到大規(guī)模的圖。經(jīng)典的例子像網(wǎng)頁鏈接關(guān)系和各種社交關(guān)系等。這些圖的規(guī)模—某些情況下，可能達到數(shù)十億的頂點和數(shù)萬億的邊—使得如何對它們進行高效的處理成為一個巨大的挑戰(zhàn)。在這篇論文中，我們將提出一種適于處理這類問題的計算模型。程序使用一系列的迭代過程來表達，在每一次迭代中，每個頂點會接收來自上一次迭代的信息，并發(fā)送信息給其它頂點，同時可能修改其自身狀態(tài)以及以它為頂點的出邊的狀態(tài)，或改變整個圖的拓撲結(jié)構(gòu)。這種以頂點為中心的策略非常靈活，足以用來表達一大類的算法。該模型的設(shè)計目標就是可以高效，可擴展，和容錯地在由上千臺機器組成的集群中得以實現(xiàn)。此外它的隱式的同步性(implied synchronicity)使得程序本身很容易理解。分布式相關(guān)的細節(jié)被隱藏在一組抽象出來的API下面。這樣展現(xiàn)給人們的就是一個具有豐富表現(xiàn)力，易于編程的大規(guī)模圖處理框架。

分布式計算，圖算法

Internet使得Web graph成為一個人們爭相分析和研究的熱門對象。Web 2.0更是激發(fā)了人們對社交網(wǎng)絡的關(guān)注。其他的一些大型圖對象（如交通路線圖，新聞文章的相似性，疾病爆發(fā)路徑，以及發(fā)表的科學研究文章中的引用關(guān)系等），也已經(jīng)被研究了數(shù)十年了。經(jīng)常被用到的一些算法包括最短路徑算法，不同種類的聚類算法，各種page rank算法變種。還有其他許多具有實際價值的圖計算問題，比如最小切割，連通分支。

對大型圖對象進行高效的處理，是非常具有挑戰(zhàn)性的。圖算法常常表現(xiàn)出比較差的內(nèi)存訪問局部性，針對單個頂點的處理工作過少，以及計算過程中伴隨著的并行度的改變等問題[31,39]。分布式的介入更是加劇了locality的問題，并且增加了在計算過程中機器發(fā)生故障的概率。盡管大型圖對象無處不在，及其在商業(yè)上的重要性，但是據(jù)我們所知，目前還不存在一種在大規(guī)模分布式環(huán)境下，可以基于各種圖表示方法來實現(xiàn)任意圖算法的，可擴展的通用系統(tǒng)。

要實現(xiàn)一種處理大規(guī)模圖對象的算法通常意味著要在以下幾點中作出選擇:

1. 為特定的圖應用定制相應的分布式實現(xiàn)。在面對新的圖算法或者圖表示方式時，就需要做大量的重復實現(xiàn)，不通用。

2. 基于現(xiàn)有的分布式計算平臺，而這種情況下，通常它們并不適于做圖處理。比如mapreduce就是一個對許多大規(guī)模計算問題都非常合適的計算框架。有時它也被用來對大規(guī)模圖對象進行挖掘[11,30]，但是通常在性能和易用性上都不是最優(yōu)的。盡管這種對數(shù)據(jù)處理的基本模式經(jīng)過擴展，已經(jīng)可以使用方便的聚合(facilitate aggregation)[Sawzall]以及類SQL的查詢方式[Pig,DryadLINQ]，但這些擴展對于圖算法這種更適合用消息傳遞模型的問題來說，通常并不理想。

3. 使用單機的圖算法庫，如BGL[43]，LEAD[35]，NetworkX[25]，JDSL[20]，Standford GraphBase[29]，F(xiàn)GL[16]等，但對可以解決的問題的規(guī)模提出了很大的限制。

4. 使用已有的并行圖計算系統(tǒng)。Parallel BGL[22]和CGMgraph[8]這些庫實現(xiàn)了很多并行圖算法，但是并沒有解決對大規(guī)模分布式系統(tǒng)中來說非常重要的容錯等一些問題。

以上的這些選擇都或多或少的存在一些局限性。為了解決大型圖的分布式計算問題，我們搭建了一套可擴展，有容錯機制的平臺，該平臺提供了一套非常靈活的API，可以描述各種各樣的圖計算。這篇論文將描述這套名為Pregel的系統(tǒng)，并分享我們的經(jīng)驗。

對Pregel計算系統(tǒng)的靈感來自Valiant提出的BSP(Bluk Synchronous Parallell)模型[45]。Pregel的計算過程由一系列被稱為超級步(superstep)的迭代(iterations)組成。在每一個超級步中，計算框架都會針對每個頂點調(diào)用用戶自定義的函數(shù)，這個過程是并行的{!即不是一個一個頂點的串行調(diào)用，同一時刻可能有多個頂點被調(diào)用}。該函數(shù)描述的是一個頂點V在一個superstep S中需要執(zhí)行的操作。該函數(shù)可以讀取前一個超級步(S-1)中發(fā)送給V的消息，并發(fā)送消息給其他頂點，這些消息將會在下一個超級步(S+1)中被接收，并且在此過程中修改頂點V及其出邊的狀態(tài)。消息通常沿著頂點的出邊發(fā)送，但一個消息可能會被發(fā)送到任意已知ID的頂點上去。

這種以頂點為中心的策略很容易讓人聯(lián)想起MapReduce，因為他們都讓用戶只需要關(guān)注其本地的執(zhí)行邏輯，每條記錄的處理都是獨立的{!相互之間不需要通信}，系統(tǒng)將這些行為組合起來就可以完成大規(guī)模數(shù)據(jù)的處理。根據(jù)設(shè)計，這種計算模型非常的適合分布式的實現(xiàn)：它沒有將任何檢測執(zhí)行順序的機制暴露在單個超級步中，所有的通信都僅限于S到S+1之間。

模型的同步性使得在實現(xiàn)算法時很容易理解程序的語義，并且使得Pregel程序天生對異步系統(tǒng)中經(jīng)常出現(xiàn)的死鎖以及臨界資源競爭就是免疫的。理論上，Pregel程序的性能即使在與足夠并行化的異步系統(tǒng)[28,34]的對比中都有一定的競爭力。因為通常情況下圖計算的應用中頂點的數(shù)量要遠遠大于機器的數(shù)量，所以必須要平衡各機器之間的負載，這樣各個superstep間的同步就不會增加過多的延遲{!負載平衡會引入大量的通信開銷，就使得超級步間的同步開銷并不那么顯眼了}。

本文接下來的結(jié)構(gòu)如下：第2節(jié)主要描述該模型；第3節(jié)描述其C++ API；第4節(jié)討論實現(xiàn)方面的情況，包括性能和容錯等；第5節(jié)將列舉幾個實際應用；第6節(jié)將提供一些性能的對比結(jié)果；最后我們會討論下相關(guān)的研究工作和未來的方向。

在Pregel計算模型中，輸入是一個有向圖，該有向圖的每一個頂點都有一個相應的由String描述的頂點標識符。每一個頂點都有一個與之對應的可修改的用戶自定義值。每一條有向邊都和其源頂點關(guān)聯(lián)，并且也擁有一個可修改的用戶自定義值，并同時還記錄了其目標頂點的標識符。

一個典型的Pregel計算過程如下：讀取輸入初始化該圖，當圖被初始化好后，運行一系列的超級步直到整個計算結(jié)束，這些超級步之間通過一些全局的同步點分隔，輸出結(jié)果結(jié)束計算。

在每個超級步中，頂點的計算都是并行的，每個頂點執(zhí)行相同的用于表達給定算法邏輯的用戶自定義函數(shù)。每個頂點可以修改其自身及其出邊的狀態(tài)，接收前一個超級步(S-1)中發(fā)送給它的消息，并發(fā)送消息給其他頂點(這些消息將會在下一個超級步中被接收)，甚至是修改整個圖的拓撲結(jié)構(gòu)。邊，在這種計算模式中并不是核心對象，沒有相應的計算運行在其上。

算法是否能夠結(jié)束取決于是否所有的頂點都已經(jīng)“vote”標識其自身已經(jīng)達到“halt”狀態(tài)了。在第0個超級步，所有頂點都處于active狀態(tài)，所有的active頂點都會參與所有對應superstep中的計算。頂點通過將其自身的status設(shè)置成“halt”來表示它已經(jīng)不再active。這就表示該頂點沒有進一步的計算需要執(zhí)行，除非被再次被外部觸發(fā)，而Pregel框架將不會在接下來的superstep中執(zhí)行該頂點，除非該頂點收到其它頂點傳送的消息。如果頂點接收到消息被喚醒進入active狀態(tài)，那么在隨后的計算中該頂點必須顯式的deactive {!?是說頂點此后會一直處于active狀態(tài)，然后要想不active必須顯示deactive，但是也可以選擇不deactive，還是說頂點必須要顯式deactive呢，感覺前者更合理}。整個計算在所有頂點都達到“inactive”狀態(tài)，并且沒有message在傳送的時候宣告結(jié)束。這種簡單的狀態(tài)機如下圖所示：

整個Pregel程序的輸出是所有頂點輸出的集合。通常來都是一個跟輸入同構(gòu)的有向圖，但是這并不是系統(tǒng)的一個必要屬性，因為頂點和邊可以在計算的過程中進行添加和刪除。比如一個聚類算法，就有可能是從一個大圖中生成的非連通頂點組成的小集合；一個對圖的挖掘算法就可能僅僅是輸出了從圖中挖掘出來的聚合數(shù)據(jù)等。

圖2通過一個簡單的例子來說明這些基本概念：給定一個強連通圖，圖中每個頂點都包含一個值，它會將最大值傳播到每個頂點。在每個超級步中，頂點會從接收到的消息中選出一個最大值，并將這個值傳送給其所有的相鄰頂點。當某個超級步中已經(jīng)沒有頂點更新其值，那么算法就宣告結(jié)束。

 

圖算法其實也可以被寫成是一系列的鏈式MapReduce調(diào)用[11,30]。我們選擇了另外一種不同的模式的原因在于可用性和性能。Pregel將頂點和邊保存在執(zhí)行計算的那臺機器上，而僅僅利用網(wǎng)絡來傳輸信息。而MapReduce本質(zhì)上是函數(shù)式的，所以將圖算法用鏈式MapReduce來實現(xiàn)就需要將整個圖的狀態(tài)從一個階段傳輸?shù)搅硗庖粋€階段，這樣就需要許多的通信和隨之而來的序列化和反序列化的開銷。另外，這一連串的MapReduce作業(yè)各執(zhí)行階段需要的協(xié)同工作也增加了編程復雜度，而在Pregel中通過引入超級步避免了這樣的情況。

這一節(jié)主要介紹Pregel C++ API中最重要的幾個方面，暫時忽略相關(guān)其他機制。編寫一個Pregel程序需要繼承Pregel中已預定義好的一個基類——Vertex類(見圖3)。 

該類的模版參數(shù)中定義了三個值類型參數(shù)，分別表示頂點，邊和消息。每一個頂點都有一個對應的給定類型的值。這種形式可能看上有很多限制，但用戶可以用protocol buffer來管理增加的其他定義和屬性。而邊和消息類型的行為比較類似。

用戶覆寫Vertex類的虛函數(shù)Compute()，該函數(shù)會在每一個超級步中對每一個頂點進行調(diào)用。預定義的Vertex類方法允許Compute()方法查詢當前頂點及其邊的信息，以及發(fā)送消息到其他的頂點。Compute()方法可以通過調(diào)用GetValue()方法來得到當前頂點的值，或者通過調(diào)用MutableValue()方法來修改當前頂點的值。同時還可以通過由出邊的迭代器提供的方法來查看修改出邊對應的值。這種狀態(tài)的修改是立時可見的。由于這種可見性僅限于被修改的那個頂點，所以不同頂點并發(fā)進行的數(shù)據(jù)訪問是不存在競爭關(guān)系的。

頂點和其對應的邊所關(guān)聯(lián)的值是唯一需要在超級步之間持久化的頂點級狀態(tài)。將由計算框架管理的圖狀態(tài)限制在一個單一的頂點值或邊值的這種做法，簡化了主計算流程，圖的分布以及故障恢復。

頂點之間的通信是直接通過發(fā)送消息，每條消息都包含了消息值和目標頂點的名稱。消息值的數(shù)據(jù)類型是由用戶通過Vertex類的模版參數(shù)來指定。

在一個超級步中，一個頂點可以發(fā)送任意多的消息。當頂點V的Compute()方法在S+1超級步中被調(diào)用時，所有在S超級步中發(fā)送給頂點V的消息都可以通過一個迭代器來訪問到。在該迭代器中并不保證消息的順序，但是可以保證消息一定會被傳送并且不會重復。

一種通用的使用方式為：對一個頂點V，遍歷其自身的出邊，向每條出邊發(fā)送消息到該邊的目標頂點，如圖4中PageRank算法(參見5.1節(jié))所示的那樣。但是，dest_vertex并不一定是頂點V的相鄰頂點。一個頂點可以從之前收到的消息中獲取到其非相鄰頂點的標識符，或者頂點標識符可以隱式的得到。比如，圖可能是一個clique(一個圖中兩兩相鄰的一個點集,或是一個完全子圖)，頂點的命名規(guī)則都是已知的(從V1到Vn)，在這種情況下甚至都不需要顯式地保存邊的信息。

當任意一個消息的目標頂點不存在時，便執(zhí)行用戶自定義的handlers。比如在這種情況下，一個handler可以創(chuàng)建該不存在的頂點或從源頂點中刪除這條邊。

發(fā)送消息，尤其是當目標頂點在另外一臺機器時，會產(chǎn)生一些開銷。某些情況可以在用戶的協(xié)助下降低這種開銷。比方說，假如Compute() 收到許多的int 值消息，而它僅僅關(guān)心的是這些值的和，而不是每一個int的值，這種情況下，系統(tǒng)可以將發(fā)往同一個頂點的多個消息合并成一個消息，該消息中僅包含它們的和值，這樣就可以減少傳輸和緩存的開銷。

Combiners在默認情況下并沒有被開啟，這是因為要找到一種對所有頂點的Compute()函數(shù)都合適的Combiner是不可能的。而用戶如果想要開啟Combiner的功能，需要繼承Combiner類，覆寫其virtual函數(shù)Combine()。框架并不會確保哪些消息會被Combine而哪些不會，也不會確保傳送給Combine()的值和Combining操作的執(zhí)行順序。所以Combiner只應該對那些滿足交換律和結(jié)合律的操作打開。

對于某些算法來說，比如單源最短路徑(參見5.2節(jié))，我們觀察到通過使用Combiner將流量降低了4倍多。

Pregel的aggregators是一種提供全局通信，監(jiān)控和數(shù)據(jù)查看的機制。在一個超級步S中，每一個頂點都可以向一個aggregator提供一個數(shù)據(jù)，系統(tǒng)會使用一種reduce操作來負責聚合這些值，而產(chǎn)生的值將會對所有的頂點在超級步S+1中可見。Pregel包含了一些預定義的aggregators，如可以在各種整數(shù)和string類型上執(zhí)行的min，max，sum操作。

Aggregators可以用來做統(tǒng)計。例如，一個sum aggregator可以用來統(tǒng)計每個頂點的出度，最后相加就是整個圖的邊的條數(shù)。更復雜的一些reduce操作還可以產(chǎn)生統(tǒng)計直方圖。

Aggregators也可以用來做全局協(xié)同。例如， Compute()函數(shù)的一些邏輯分支可能在某些超級步中執(zhí)行，直到and aggregator表明所有頂點都滿足了某條件，之后執(zhí)行另外的邏輯分支直到結(jié)束。又比如一個作用在頂點ID之上的min和max aggregator，可以用來選定某頂點在整個計算過程中扮演某種角色等。

要定義一個新的aggregator，用戶需要繼承預定義的Aggregator類，并定義在第一次接收到輸入值后如何初始化，以及如何將接收到的多個值最后reduce成一個值。Aggregator操作也應該滿足交換律和結(jié)合律。

默認情況下，一個aggregator僅僅會對來自同一個超級步的輸入進行聚合，但是有時也可能需要定義一個sticky aggregator，它可以從所有的supersteps中接收數(shù)據(jù)。這是非常有用的，比如要維護全局的邊條數(shù)，那么就僅僅在增加和刪除邊的時候才調(diào)整這個值了。

還可以有更高級的用法。比如，可以用來實現(xiàn)一個△-stepping最短路徑算法所需要的分布式優(yōu)先隊列[37]。每個頂點會根據(jù)它的當前距離分配一個優(yōu)先級bucket。在每個超級步中，頂點將它們的indices匯報給min aggregator。在下一個超級步中，將最小值廣播給所有worker，然后讓在最小index的bucket中的頂點放松它們的邊。{!說明此處的核心在于說明aggregators用法，關(guān)于△-stepping最短路徑算法不再解釋，感興趣的可以參考這篇文章：Δ-Stepping: A Parallel Single Source Shortest Path Algorithm }

有一些圖算法可能需要改變圖的整個拓撲結(jié)構(gòu)。比如一個聚類算法，可能會將每個聚類替換成一個單一頂點，又比如一個最小生成樹算法會刪除所有除了組成樹的邊之外的其他邊。正如用戶可以在自定義的Compute()函數(shù)能發(fā)送消息，同樣可以產(chǎn)生在圖中增添和刪除邊或頂點的請求。

多個頂點有可能會在同一個超級步中產(chǎn)生沖突的請求(比如兩個請求都要增加一個頂點V，但初始值不一樣)。Pregel中用兩種機制來決定如何調(diào)用：局部有序和handlers。

由于是通過消息發(fā)送的，拓撲改變在請求發(fā)出以后，在超級步中可以高效地執(zhí)行。在該超級步中，刪除會首先被執(zhí)行，先刪除邊后刪除頂點，因為頂點的刪除通常也意味著刪除其所有的出邊。然后執(zhí)行添加操作，先增加頂點后增加邊，并且所有的拓撲改變都會在Compute()函數(shù)調(diào)用前完成。這種局部有序保證了大多數(shù)沖突的結(jié)果的確定性。

剩余的沖突就需要通過用戶自定義的handlers來解決。如果在一個超級步中有多個請求需要創(chuàng)建一個相同的頂點，在默認情況下系統(tǒng)會隨便挑選一個請求，但有特殊需求的用戶可以定義一個更好的沖突解決策略，用戶可以在Vertex類中通過定義一個適當?shù)膆andler函數(shù)來解決沖突。同一種handler機制將被用于解決由于多個頂點刪除請求或多個邊增加請求或刪除請求而造成的沖突。我們委托handler來解決這種類型的沖突，從而使得Compute()函數(shù)變得簡單，而這樣同時也會限制handler和Compute()的交互，但這在應用中還沒有遇到什么問題。

我們的協(xié)同機制比較懶，全局的拓撲改變在被apply之前不需要進行協(xié)調(diào){!即在變更請求的發(fā)出端不會進行任何的控制協(xié)調(diào)，只有在它被接收到然后apply時才進行控制，這樣就簡化了流程，同時能讓發(fā)送更快}。這種設(shè)計的選擇是為了優(yōu)化流式處理。直觀來講就是對頂點V的修改引發(fā)的沖突由V自己來處理。

Pregel同樣也支持純local的拓撲改變，例如一個頂點添加或刪除其自身的出邊或刪除其自己。Local的拓撲改變不會引發(fā)沖突，并且頂點或邊的本地增減能夠立即生效，很大程度上簡化了分布式的編程。

可以采用多種文件格式進行圖的保存，比如可以用text文件，關(guān)系數(shù)據(jù)庫，或者Bigtable[9]中的行。為了避免規(guī)定死一種特定文件格式，Pregel將從輸入中解析出圖結(jié)構(gòu)的任務從圖的計算過程中進行了分離。類似的，結(jié)果可以以任何一種格式輸出并根據(jù)應用程序選擇最適合的存儲方式。Pregel library本身提供了很多常用文件格式的readers和writers，但是用戶可以通過繼承Reader和Writer類來定義他們自己的讀寫方式。

Pregel是為Google的集群架構(gòu)[3]而設(shè)計的。每一個集群都包含了上千臺機器，這些機器都分列在許多機架上，機架之間有這非常高的內(nèi)部通信帶寬。集群之間是內(nèi)部互聯(lián)的，但地理上是分布在不同地方的。

應用程序通常通過一個集群管理系統(tǒng)執(zhí)行，該管理系統(tǒng)會通過調(diào)度作業(yè)來優(yōu)化集群資源的使用率，有時候會殺掉一些任務或?qū)⑷蝿者w移到其他機器上去。該系統(tǒng)中提供了一個名字服務系統(tǒng)，所以各任務間可以通過與物理地址無關(guān)的邏輯名稱來各自標識自己。持久化的數(shù)據(jù)被存儲在GFS[19]或Bigtable[9]中，而臨時文件比如緩存的消息則存儲在本地磁盤中。

Pregel library將一張圖劃分成許多的partitions，每一個partition包含了一些頂點和以這些頂點為起點的邊。將一個頂點分配到某個partition上去取決于該頂點的ID，這意味著即使在別的機器上，也是可以通過頂點的ID來知道該頂點是屬于哪個partition，即使該頂點已經(jīng)不存在了。默認的partition函數(shù)為hash(ID) mod N，N為所有partition總數(shù)，但是用戶可以替換掉它。

將一個頂點分配給哪個worker機器是整個Pregel中對分布式不透明的主要地方。有些應用程序使用默認的分配策略就可以工作地很好，但是有些應用可以通過定義更好地利用了圖本身的locality的分配函數(shù)而從中獲益。比如，一種典型的可以用于Web graph的啟發(fā)式方法是，將來自同一個站點的網(wǎng)頁數(shù)據(jù)分配到同一臺機器上進行計算。

在不考慮出錯的情況下，一個Pregel程序的執(zhí)行過程分為如下幾個步驟：

1. 用戶程序的多個copy開始在集群中的機器上執(zhí)行。其中有一個copy將會作為master，其他的作為worker，master不會被分配圖的任何一部分，而只是負責協(xié)調(diào)worker間的工作。worker利用集群管理系統(tǒng)中提供的名字服務來定位master位置，并發(fā)送注冊信息給master。

2. Master決定對這個圖需要多少個partition，并分配一個或多個partitions到worker所在的機器上。這個數(shù)字也可能由用戶進行控制。一個worker上有多個partition的情況下，可以提高partitions間的并行度，更好的負載平衡，通常都可以提高性能。每一個worker負責維護在其之上的圖的那一部分的狀態(tài)(頂點及邊的增刪)，對該部分中的頂點執(zhí)行Compute()函數(shù)，并管理發(fā)送出去的以及接收到的消息。每一個worker都知道該圖的計算在所有worker中的分配情況。

3. Master進程為每個worker分配用戶輸入中的一部分，這些輸入被看做是一系列記錄的集合，每一條記錄都包含任意數(shù)目的頂點和邊。對輸入的劃分和對整個圖的劃分是正交的，通常都是基于文件邊界進行劃分。如果一個worker加載的頂點剛好是這個worker所分配到的那一部分，那么相應的數(shù)據(jù)結(jié)構(gòu)就會被立即更新。否則，該worker就需要將它發(fā)送到它所應屬于的那個worker上。當所有的輸入都被load完成后，所有的頂點將被標記為active狀態(tài)， 

4. Master給每個worker發(fā)指令，讓其運行一個超級步，worker輪詢在其之上的頂點，會為每個partition啟動一個線程。調(diào)用每個active頂點的Compute()函數(shù)，傳遞給它從上一次超級步發(fā)送來的消息。消息是被異步發(fā)送的，這是為了使得計算和通信可以并行，以及進行batching，但是消息的發(fā)送會在本超級步結(jié)束前完成。當一個worker完成了其所有的工作后，會通知master，并告知當前該worker上在下一個超級步中將還有多少active節(jié)點。

不斷重復該步驟，只要有頂點還處在active狀態(tài)，或者還有消息在傳輸。

5. 計算結(jié)束后，master會給所有的worker發(fā)指令，讓它保存它那一部分的計算結(jié)果。

容錯是通過checkpointing來實現(xiàn)的。在每個超級步的開始階段，master命令worker讓它保存它上面的partitions的狀態(tài)到持久存儲設(shè)備，包括頂點值，邊值，以及接收到的消息。Master自己也會保存aggregator的值。

worker的失效是通過master發(fā)給它的周期性的ping消息來檢測的。如果一個worker在特定的時間間隔內(nèi)沒有收到ping消息，該worker進程會終止。如果master在一定時間內(nèi)沒有收到worker的反饋，就會將該worker進程標記為失敗。

當一個或多個worker發(fā)生故障，被分配到這些worker的partitions的當前狀態(tài)信息就丟失了。Master重新分配圖的partition到當前可用的worker集合上，所有的partition會從最近的某超級步S開始時寫出的checkpoint中重新加載狀態(tài)信息。該超級步可能比在失敗的worker上最后運行的超級步 S’早好幾個階段，此時失去的幾個superstep將需要被重新執(zhí)行{!應該是所有的partition都需要重新分配，而不僅僅是失敗的worker上的那些，否則如何重新執(zhí)行丟失的超級步，也正是這樣才有了下面的confined recovery}。我們對checkpoint頻率的選擇基于某個故障模型[13]的平均時間，以平衡checkpoint的開銷和恢復執(zhí)行的開銷。

為了改進恢復執(zhí)行的開銷和延遲， Confined recovery已經(jīng)在開發(fā)中。除了基本的checkpoint，worker同時還會將其在加載圖的過程中和超級步中發(fā)送出去的消息寫入日志。這樣恢復就會被限制在丟掉的那些 partitions上。它們會首先通過checkpoint進行恢復，然后系統(tǒng)會通過回放來自正常的partitions的記入日志的消息以及恢復過來的partitions重新生成的消息，更新狀態(tài)到S’階段。這種方式通過只對丟失的partitions進行重新計算節(jié)省了在恢復時消耗的計算資源，同時由于每個worker只需要恢復很少的partitions，減少了恢復時的延遲。對發(fā)送出去的消息進行保存會產(chǎn)生一定的開銷，但是通常機器上的磁盤帶寬不會讓這種IO操作成為瓶頸。

Confined recovery要求用戶算法是確定性的，以避免原始執(zhí)行過程中所保存下的消息與恢復時產(chǎn)生的新消息并存情況下帶來的不一致。隨機化算法可以通過基于超級步和partition產(chǎn)生一個偽隨機數(shù)生成器來使之確定化。非確定性算法需要關(guān)閉Confined recovery而使用老的恢復機制。

一個worker機器會在內(nèi)存中維護分配到其之上的graph partition的狀態(tài)。概念上講，可以簡單地看做是一個從頂點ID到頂點狀態(tài)的Map，其中頂點狀態(tài)包括如下信息：該頂點的當前值，一個以該頂點為起點的出邊(包括目標頂點ID，邊本身的值)列表，一個保存了接收到的消息的隊列，以及一個記錄當前是否active的標志位。該worker在每個超級步中，會循環(huán)遍歷所有頂點，并調(diào)用每個頂點的Compute()函數(shù)，傳給該函數(shù)頂點的當前值，一個接收到的消息的迭代器和一個出邊的迭代器。這里沒有對入邊的訪問，原因是每一條入邊其實都是其源頂點的所有出邊的一部分，通常在另外的機器上。

出于性能的考慮，標志頂點是否為active的標志位是和輸入消息隊列分開保存的。另外，只保存了一份頂點值和邊值，但有兩份頂點active flag和輸入消息隊列存在，一份是用于當前超級步，另一個用于下一個超級步。當一個worker在進行超級步S的頂點處理時，同時還會有另外一個線程負責接收從處于同一個超級步的其他worker接收消息。由于頂點當前需要的是S-1超級步的消息，那么對superstep S和superstep S+1的消息就必須分開保存。類似的，頂點V接收到了消息表示V將會在下一個超級步中處于active，而不是當前這一次。

當Compute()請求發(fā)送一個消息到其他頂點時，worker首先確認目標頂點是屬于遠程的worker機器，還是當前worker。如果是在遠程的worker機器上，那么消息就會被緩存，當緩存大小達到一個閾值，最大的那些緩存數(shù)據(jù)將會被異步地flush出去，作為單獨的一個網(wǎng)絡消息傳輸?shù)侥繕藈orker。如果是在當前worker，那么就可以做相應的優(yōu)化：消息就會直接被放到目標頂點的輸入消息隊列中。

如果用戶提供了Combiner，那么在消息被加入到輸出隊列或者到達輸入隊列時，會執(zhí)行combiner函數(shù)。后一種情況并不會節(jié)省網(wǎng)絡開銷，但是會節(jié)省用于消息存儲的空間。

Master主要負責的worker之間的工作協(xié)調(diào)，每一個worker在其注冊到master的時候會被分配一個唯一的ID。Master內(nèi)部維護著一個當前活動的worker列表，該列表中就包括每個worker的ID和地址信息，以及哪些worker被分配到了整個圖的哪一部分。Master中保存這些信息的數(shù)據(jù)結(jié)構(gòu)大小與partitions的個數(shù)相關(guān)，與圖中的頂點和邊的數(shù)目無關(guān)。因此，雖然只有一臺master，也足夠用來協(xié)調(diào)對一個非常大的圖的計算工作。

絕大部分的master的工作，包括輸入 ，輸出，計算，保存以及從 checkpoint中恢復，都將會在一個叫做barriers的地方終止：Master在每一次操作時都會發(fā)送相同的指令到所有的活著的worker，然后等待從每個worker的響應。如果任何一個worker失敗了，master便進入4.2節(jié)中描述的恢復模式。如果barrier同步成功，master便會進入下一個處理階段，例如master增加超級步的index，并進入下一個超級步的執(zhí)行。

Master同時還保存著整個計算過程以及整個graph的狀態(tài)的統(tǒng)計數(shù)據(jù)，如圖的總大小，關(guān)于出度分布的柱狀圖，處于active狀態(tài)的頂點個數(shù)，在當前超級步的時間信息和消息流量，以及所有用戶自定義aggregators的值等。為方便用戶監(jiān)控，Master在內(nèi)部運行了一個HTTP服務器來顯示這些信息。

每個Aggregator(見3.3節(jié))會通過對一組value值集合應用aggregation函數(shù)計算出一個全局值。每一個worker都保存了一個aggregators的實例集，由type name和實例名稱來標識。當一個worker對graph的某一個partition執(zhí)行一個超級步時，worker會combine所有的提供給本地的那個aggregator實例的值到一個local value：即利用一個aggregator對當前partition中包含的所有頂點值進行局部規(guī)約。在超級步結(jié)束時，所有workers會將所有包含局部規(guī)約值的aggregators的值進行最后的匯總，并匯報給master。這個過程是由所有worker構(gòu)造出一棵規(guī)約樹而不是順序的通過流水線的方式來規(guī)約，這樣做的原因是為了并行化規(guī)約時cpu的使用。在下一個超級步開始時，master就會將aggregators的全局值發(fā)送給每一個worker。

本節(jié)包含四個例子，它們是由Pregel用戶開發(fā)地用來解決如下實際問題的簡化版算法：PageRank，最短路徑，二分圖匹配和Semi-Clustering算法。

{!首先來簡要介紹下PageRank算法：將文獻檢索中的引用理論用到Web中，引用網(wǎng)頁的鏈接數(shù)一定程度上反映了該網(wǎng)頁的重要性和質(zhì)量。PageRank發(fā)展了這種思想，網(wǎng)頁間的鏈接是不平等的。PageRank定義如下: 我們假設(shè)T1…Tn指向網(wǎng)頁A（例如，被引用）。參數(shù)d是制動因子，取值在0，1之間。通常d等于0.85。網(wǎng)頁A的PageRank值由下式給出： 

PR(A) = (1-d) + d (PR(T1)/C(T1) + … + PR(Tn)/C(Tn))。

PageRank或PR（A）可以用簡單的迭代算法計算，計算過程是收斂的，隨著迭代次數(shù)的增加，各網(wǎng)頁的PageRank值趨于平穩(wěn)。可以從如下角度進行理解：

1. 假設(shè)網(wǎng)上沖浪是隨機的，不斷點擊鏈接，從不返回，最終煩了，另外隨機選一個網(wǎng)頁重新開始沖浪。隨機訪問一個網(wǎng)頁的可能性就是它的PageRank值。制動因子d是隨機訪問一個網(wǎng)頁煩了的可能性，隨機另選一個網(wǎng)頁。對單個網(wǎng)頁或一組網(wǎng)頁，一個重要的變量加入到制動因子d中。這允許個人可以故意地誤導系統(tǒng)，以得到較高的PageRank值。

2. 直覺上判斷，一個網(wǎng)頁有很多網(wǎng)頁指向它，或者一些PageRank值高的網(wǎng)頁指向它，則這個網(wǎng)頁很重要。直覺地，在Web中，一個網(wǎng)頁被很多網(wǎng)頁引用，那么這個網(wǎng)頁值得一看。一個網(wǎng)頁被象Yahoo這樣重要的主頁引用即使一次，也值得一看。如果一個網(wǎng)頁的質(zhì)量不高，或者是死鏈接，象Yahoo這樣的主頁不會鏈向它。PageRank處理了這兩方面因素，并通過網(wǎng)絡鏈接遞歸地傳遞。

關(guān)于該公式需要說明的是，Google后來調(diào)整時使用了1-d/N，公式的其他部分未作任何變動，這里的N是互聯(lián)網(wǎng)中全部的網(wǎng)頁數(shù)量，也就是本論文中使用的公式，據(jù)說這樣做使得PageRank變?yōu)榱吮浑S機訪問的期望值。關(guān)于PageRank更具體的解釋可以參考這篇文章：數(shù)據(jù)挖掘10大算法(1)：PageRank。

}

PageRank算法[7]的Pregel實現(xiàn)如圖4所示。PageRankVertex繼承自Vertex類。頂點value類型是double，用來保存PageRank中間值，消息類型也是double，用來傳輸PageRank分數(shù)，邊的value類型是void，因為不需要存儲任何信息。我們假設(shè)，在第0個超級步時，圖中各頂點的value值被初始化為1/NumVertices()。在前30個超級步中，每個頂點都會沿著它的出邊發(fā)送它的PageRank值除以出邊數(shù)后的結(jié)果值。從第1個超級步開始，每個頂點會將到達的消息中的值加到sum值中，同時將它的PageRank值設(shè)為0.15/ NumVertices()+0.85*sum。到了第30個超級步后，就沒有需要發(fā)送的消息了，同時所有的頂點VoteToHalt。在實際中，PageRank算法需要一直運行直到收斂，可以使用aggregators來檢查是否滿足收斂條件。

最短路徑問題是圖論中最有名的問題之一了，同時具有廣泛的應用[10,24]，該問題有幾個形式：單源最短路徑，是指要找出從某個源頂點到其他所有頂點的最短路徑；s-t最短路徑，是指要找出給定源頂點s和目標頂點t間的最短路徑，這個問題具有廣泛的實驗應用比如尋找駕駛路線，并引起了廣泛關(guān)注，同時它也是相對簡單的；全局最短路徑，對于大規(guī)模的圖對象來說通常都不太實際，因為它的空間復雜度是O(V*V)的。為簡化起見，我們這里以非常適于用Pregel解決的單源最短路徑為例，實現(xiàn)代碼見圖5。

在該算法中，我們假設(shè)與頂點關(guān)聯(lián)的那個值被初始化為INF(比從源點到圖中其他頂點的所有可能距離都大的一個常量)。在每個超級步中，每個頂點會首先接收到來自鄰居傳送過來的消息，該消息包含更新過的從源頂點到該頂點的潛在的最短距離{!鄰居節(jié)點發(fā)送過來的已經(jīng)是源頂點到它本身的當前已知的最短距離+它到該頂點的距離了，所以該頂點接收到的已經(jīng)是源頂點到它的距離了}。如果這些更新里的最小值小于該頂點當前關(guān)聯(lián)值，那么頂點就會更新這個值，并發(fā)送消息(該消息包含了該頂點的關(guān)聯(lián)值+每個出邊的關(guān)聯(lián)值)給它的鄰居。在第一個超級步中，只有源頂點會更新它的關(guān)聯(lián)值(從INF改為0)，然后發(fā)送消息給它的直接鄰居。然后這些鄰居會更新它們的關(guān)聯(lián)值，然后繼續(xù)發(fā)送消息給它們的鄰居，如此循環(huán)往復。當沒有更新再發(fā)生的時候，算法就結(jié)束，之后所有頂點的關(guān)聯(lián)值就是從源頂點到它的最短距離，若值為INF表示該頂點不可達。如果所有的邊權(quán)重都是非負的，就可以保證該過程肯定會結(jié)束。

該算法中的消息保存都是潛在的最小距離。由于接收頂點實際上只關(guān)注最小值，因此該算法是可以通過combiner進行優(yōu)化的，combiner實現(xiàn)如圖6所示，它可以大大減少worker間的消息量，以及在執(zhí)行下一個超級步前所需要緩存的數(shù)據(jù)量。圖5中的實現(xiàn)只是計算出了最短距離，如果要計算最短路徑生成樹也是很簡單的。

與其他類似的串行算法比如Dijkstra或者是Bellman-Ford[5,15,17,24]相比，該算法需要更多的比較次數(shù)，但是它可以用來解決對于單機版實現(xiàn)很難解決的那個規(guī)模上的最短路徑問題。還有一些更高級的并行算法，比如，Thorup[44]和△-stepping算法[37]，這些高級算法也可以在Pregel系統(tǒng)中實現(xiàn)。但是圖5中的那個實現(xiàn)，由于其比較簡單同時性能也還可以接受，對于那些普通用戶來說也還是很具有吸引力的。

二分匹配算法的輸入由兩個不同的頂點集合組成，所有邊的兩頂點分別位于兩個集合中，輸出是邊的一個子集，它們之間沒有公共頂點。極大匹配(Maximal Matching)是指在當前已完成的匹配下，無法再通過增加未完成匹配的邊的方式來增加匹配的邊數(shù)。我們實現(xiàn)了一個隨機化的極大匹配算法[1]，以及一個最大權(quán)匹配算法[4]；我們只在此描述下前者。

在該算法的Pregel實現(xiàn)中，頂點的關(guān)聯(lián)值是由兩個值組成的元組(tuple)：一個是用于標識該頂點所處集合(L or R)的flag，一個是跟它所匹配的頂點名稱。邊的關(guān)聯(lián)值類型為void，消息的類型為boolean。該算法是由四個階段組成的多個循環(huán)組成，用來標識當前所處階段的index可以通過用當前超級步的index mod 4得到。

在循環(huán)的階段0，左邊集合中那些還未被匹配的頂點會發(fā)送消息給它的每個鄰居請求匹配，然后會無條件的VoteToHalt。如果它沒有發(fā)送消息(可能是因為它已經(jīng)找到了匹配，或者沒有出邊)，或者是所有的消息接收者都已經(jīng)被匹配，該頂點就不會再變?yōu)閍ctive狀態(tài)。

在循環(huán)的階段1，右邊集合中那些還未被匹配的頂點隨機選擇它接收到的消息中的其中一個，并發(fā)送消息表示接受該請求，然后給其他請求者發(fā)送拒絕消息。然后，它也無條件的VoteToHalt。

在循環(huán)的階段2，左邊集合中那些還未被匹配的頂點選擇它所收到右邊集合發(fā)送過來的接受請求中的其中一個，并發(fā)送一個確認消息。左邊集合中那些已經(jīng)匹配好的頂點永遠都不會執(zhí)行這個階段，因為它們不會在階段0發(fā)送任何消息。

最后，在階段3，右邊集合中還未被匹配的頂點最多會收到一個確認消息。它會通知匹配頂點，然后無條件的VoteToHalt，它的工作已經(jīng)完成。

。

我們使用5.2節(jié)描述的單源最短路徑(SSSP)實現(xiàn)在一個由300臺多核PC組成的集群上進行了多次實驗。得到了在所有邊的權(quán)重為1情況下，針對不同大小規(guī)模下的二叉樹(為了研究可擴展屬性)和對數(shù)正態(tài)隨機圖(為了研究更接近真實環(huán)境下的性能)的運行時間。

測量結(jié)果沒有包含用于初始化集群，在內(nèi)存中生成測試圖以及進行結(jié)果驗證的時間。因為所有的實驗運行的時間都相對比較短，因此出錯的概率比較低，同時關(guān)閉了checkpointing。

圖7展示了一個具有十億個頂點(由于是樹，故邊數(shù)應為十億-1)的二叉樹最短路徑算法的在Pregel worker數(shù)目從50到800之間的情況下的運行時間。通過該圖可以看出Pregel伴隨著worker數(shù)增加的擴展性?？梢钥吹竭\行時間從170秒降到了17.3秒，相當于使用16倍的worker數(shù)獲得了大概10倍的加速。

 

前面的實驗只是展示了Pregel隨著worker數(shù)目和圖大小增加的情況下的可擴展性，但是二叉樹很明顯無法代表實際中經(jīng)常碰到的那些圖。因此我們需要繼續(xù)實驗，通過使用一個出度具有對數(shù)正態(tài)分布的隨機生成的圖來進行，同時我們令μ=4，σ=1.3，此時的平均出度為127.1。這個分布與很多現(xiàn)實中的大型圖都很類似，比如web graph或者是社交網(wǎng)絡，在這些情況下，大多數(shù)頂點的度都相對較小，但是存在少數(shù)的一些頂點的度非常大—可能成千上萬甚至更多。

 圖9展示了這種類型的圖在頂點數(shù)從一千萬到十億(此時邊數(shù)將超過127.1B)的情況下的最短路徑算法運行時間，也是總共800個worker被調(diào)度300臺機器上。最大規(guī)模的情況下的運行時間超過了10分鐘。

在所有的實驗中，圖在worker上的partition方式都是采用的默認的基于隨機hash的partition函數(shù)；如果使用一個topolog-aware的partition函數(shù)應該可以得到更好的性能。此外這里只是使用了一個最基本的最短路徑算法；如果使用一個更高級的算法還能再達到更好的性能。因此，本節(jié)的實驗結(jié)果并不代表了使用Pregel所能達到的最好性能。這些結(jié)果只是用來展示：只需要編寫非常少的代碼就可以達到令人滿意的性能。實際上，我們的系統(tǒng)在1B頂點和邊的情況下的性能結(jié)果與下一節(jié)提到的運行在112個處理單元組成的集群上，處理具有256M頂點和1B條邊的圖的Parallel BGL[31]的△-stepping算法結(jié)果不相上下，同時Pregel還可以擴展到更大的規(guī)模上。

Pregel是一個分布式編程框架，專注于為用戶編寫圖算法提供自然的API，同時將消息機制和容錯等底層分布式細節(jié)隱藏起來。從概念上看，它非常類似于MapReduce[14]，但是具有更自然的面前圖的API和更高效的在圖上進行迭代計算的支持。由于專注于圖對象使得它與其他的一些分布式框架比如Sawzall[41],Pig Latin[40],Drayad[27,47]區(qū)別開來。Pregel之所以不同，還因為它實現(xiàn)了一個有狀態(tài)的模型，在這個模型中進程會一直存活著，不斷地進行計算，通信和修改本地狀態(tài)等等，這與數(shù)據(jù)流模型不同，在數(shù)據(jù)流模型中進程只是在輸入數(shù)據(jù)上進行計算，然后產(chǎn)生輸出數(shù)據(jù)再交由其他進程處理。

Pregel借鑒了BSP模型的思想，比如它里面的由計算和通信組成的超級步概念。目前已經(jīng)存在大量的普通BSP庫實現(xiàn)，比如Oxford BSP[38],GreenBSP[21],BSPlib[26]和Paderborn University BSP Library[6]。這些庫本身在提供的通信原語，處理分布式環(huán)境下的問題(比如容錯)的方式，負載平衡等方面都有些不同。據(jù)我們所知，這些BSP實現(xiàn)都是只在幾十臺機器上運行過，可擴展性和容錯性都比較有限，同時都沒有提供一個面向圖處理的API。

跟Pregel最接近的應該算是Parallel Boot Graph Library和CGMGraph了。Parallel BGL[22,23]為分布式的圖定義了一些關(guān)鍵概念，提供了一個基于MPI[18]的實現(xiàn)，同時基于此實現(xiàn)了大量的圖算法。并試圖維護與BGL(串行的)[43]的兼容性，以方便算法的移植。它內(nèi)部實現(xiàn)采用一個property map來存儲與圖的頂點和邊相關(guān)的信息，采用ghost cells存放與遠程組件相關(guān)的值。在具有很多遠程組件時這會影響可擴展性。Pregel使用了一種顯式的消息機制來獲取遠程信息，同時不會將這些值存放在本地。最關(guān)鍵的區(qū)別是Pregel提供了容錯機制來處理計算環(huán)境中發(fā)生的故障，這就使得它可以部署在一個很大的集群環(huán)境中，在這種規(guī)模的集群中故障是很常見的，比如硬件產(chǎn)生了故障或者高優(yōu)先級的作業(yè)發(fā)生了資源搶占。

CGMGraph[8]在概念上非常類似于Pregel，它通過基于MPI的CGM(Coarse Grained Multicomputer)模型提供了大量并行圖算法實現(xiàn)。它暴露給用戶更多的底層分布式機制，更關(guān)注于提供圖算法實現(xiàn)而不是提供一個實現(xiàn)這些算法的基礎(chǔ)設(shè)施。CGMGraph使用了面向?qū)ο蟮木幊田L格，與Parallel BGL和Pregel的泛型編程風格相比，會有一些性能損失。

除了Pregel和Parallel BGL之外，基本沒有其他系統(tǒng)提供過在billions這個規(guī)模的頂點數(shù)的實驗結(jié)果。目前已發(fā)表的最大規(guī)模的實驗結(jié)果來自于一個針對s-t最短路徑的定制化實現(xiàn)，而不是來自通用框架。Yoo 等[46]發(fā)表的廣度優(yōu)先搜索實現(xiàn)(s-t最短路徑)在BlueGene/L上的實現(xiàn)，是運行在32786個PowerPC處理器上，同時采用了高性能torus網(wǎng)絡，對于一個具有3.2B個頂點和32B 條邊滿足泊松分布的隨機圖的處理用了1.5秒。Bader和Madduri[2]發(fā)表了該類似問題在一個10節(jié)點Cray MTA-2上的結(jié)果，對于一個具有134M個頂點和805M條邊的R-MAT隨機圖的處理用了0.43秒。Lumsadaine等[31]用在一個具有200個處理器的x86-64 Opteron集群上的Parallel BGL結(jié)果，與BlueGene/L實現(xiàn)進行了對比，對于一個具有4B個頂點和20B條邊的Erdos-Renyi隨機圖的處理用了0.43秒。

對于一個具有256M個頂點和平均出度為4的Erdos-Renyi隨機圖的上的單源最短路徑問題來說，在使用△-stepping算法的情況下，結(jié)果如下：Cray MTA-2(40個處理器，2.37秒，[32])，在Opterons上的Parallel BGL(112個處理器，35秒，[31])。后面的這個時間比較接近于我們針對1B頂點和邊的規(guī)模在400個worker上的結(jié)果。我們還沒有任何其他的在1B頂點和127.1B邊這個規(guī)模上的log-normal隨機圖上的相關(guān)結(jié)果。

另外的一個研究方向是在單臺機器上通過擴展內(nèi)存磁盤來處理更大規(guī)模的問題，比如[33,36]。但是這些實現(xiàn)，對于1B個頂點的規(guī)模的圖的處理要花幾個小時。

本文的貢獻是提出了一個適用于大規(guī)模圖計算的模型，并描述了它的高質(zhì)量的，可擴展的，容錯實現(xiàn)。

來自用戶的數(shù)據(jù)顯示，我們已經(jīng)成功地讓該模型被使用起來，并具有了不錯的可用性。目前已經(jīng)有很多Pregel應用被部署，同時還要更多地處于設(shè)計和實現(xiàn)的過程中。用戶反映當它們將思維方式成功轉(zhuǎn)換到”think like a vertex”后，發(fā)現(xiàn)提供的API是如此直觀，靈活，太好用了。這并不令人吃驚，因為我們從一開始就是跟早期用戶一起做這項工作的，從那時起他們就影響著這些API的設(shè)計了。比如，aggregators之所以被支持就是為了解決用戶在早期Pregel模型發(fā)現(xiàn)的一些限制。此外還有其他的關(guān)于Pregel可用性方面的改進都是源自用戶的使用經(jīng)驗，比如關(guān)于Pregel程序執(zhí)行過程的詳細信息的狀態(tài)頁面，unittesting框架，以及用來幫助用戶進行快速原型開發(fā)和debug的單機運行模式。

Pregel已經(jīng)在性能，可擴展性和容錯方面滿足了具有billios規(guī)模的邊的圖的處理。同時我們也在繼續(xù)進行調(diào)研以擴展到更大的規(guī)模，比如放松模型的同步性，避免讓那些運行的快的worker總是等待在超級步之間。

當前整個的計算狀態(tài)都是駐留在內(nèi)存中的。我們已經(jīng)開始將一些數(shù)據(jù)存到本地磁盤，同時我們會繼續(xù)在這個方向進行深入的研究，希望可以支持規(guī)模太大以至于內(nèi)存無法完全存下的情況。

通過調(diào)整頂點在機器間的分配以最小化機器間的通信開銷非常具有挑戰(zhàn)性。根據(jù)圖的拓撲結(jié)構(gòu)對輸入進行劃分有時可能行地通，有時可能不行，圖的拓撲結(jié)構(gòu)可能會動態(tài)地發(fā)生改變。我們希望可以引入動態(tài)的re-parttioning機制。

Pregel是為那種通信主要發(fā)送在邊上的稀疏圖設(shè)計的，我們會一直堅持這個假設(shè)。盡管我們已經(jīng)投入一些經(jīng)歷用于支持高的消息收發(fā)流量的情況，但是當大部分的頂點都在持續(xù)地向大部分的其他頂點發(fā)送消息時，性能問題會變得很嚴重。當然了，實際中dense的圖還是很少的，同時需要在稀疏圖上進行dense通信的算法也是很少發(fā)生的。對于這樣的算法來說，其中一些可以轉(zhuǎn)換成Pregel能夠比較好的支持的變種，比如通過使用combiners，aggregators或者拓撲變更，當然了這樣的計算對于任何高度分布式的系統(tǒng)來說都會是很難的。

需要注意的一點是，Pregel已經(jīng)正成為我們的基礎(chǔ)設(shè)施的一部分?，F(xiàn)在已經(jīng)不能隨意地去修改API而不考慮兼容性了。但是，我們相信現(xiàn)有的編程接口已經(jīng)是足夠抽象和靈活了，足以應對底層系統(tǒng)未來的演化。

We thank Pregel's very early users--Lorenz Huelsbergen,Galina Shubina,Zoltan Gyongyi--for their contributions to the model. Discussions with Adnan Aziz,Yossi Matias,and Steffen Meschkat helped refine several aspects of Pregel. Our interns, Punyashloka Biswal and Petar Maymounkov, provided initial evidence of Pregel's applicability to matchings and clustering, and Charles Reiss automated checkpointing

decisions. The paper benefited from comments on its earlier drafts from Jeff Dean, Tushar Chandra, Luiz Barroso, Urs Holzle, Robert Henry, Marian Dvorsky, and the anonymous reviewers. Sierra Michels-Slettvet advertised Pregel to various teams within Google computing over interesting but less known graphs. Finally, we thank all the users of Pregel for feedback and many great ideas.

[1] Thomas Anderson, Susan Owicki, James Saxe, and Charles Thacker, High-Speed Switch Scheduling for Local-Area Networks. ACM Trans. Comp. Syst. 11(4),1993, 319{352.

[2] David A. Bader and Kamesh Madduri, Designing multithreaded algorithms for breadth-_rst search and st-connectivity on the Cray MTA-2, in Proc. 35th Intl. Conf. on Parallel Processing (ICPP'06), Columbus,OH, August 2006, 523|530.

[3] Luiz Barroso, Je_rey Dean, and Urs Hoelzle, Web search for a planet: The Google Cluster Architecture.IEEE Micro 23(2), 2003, 22{28.

[4] Mohsen Bayati, Devavrat Shah, and Mayank Sharma, Maximum Weight Matching via Max-Product Belief Propagation. in Proc. IEEE Intl. Symp. On Information Theory, 2005, 1763{1767.

[5] Richard Bellman, On a routing problem. Quarterly of Applied Mathematics 16(1), 1958, 87{90.

[6] Olaf Bonorden, Ben H.H. Juurlink, Ingo von Otte, andIngo Rieping, The Paderborn University BSP (PUB) Library. Parallel Computing 29(2), 2003, 187{207.

[7] Sergey Brin and Lawrence Page, The Anatomy of a Large-Scale Hypertextual Web Search Engine. in Proc.7th Intl. Conf. on the World Wide Web, 1998,107{117.

[8] Albert Chan and Frank Dehne,CGMGRAPH/CGMLIB: Implementing and Testing CGM Graph Algorithms on PC Clusters and Shared Memory Machines. Intl. J. of High Performance Computing Applications 19(1), 2005, 81{97.

[9] Fay Chang, Je_rey Dean, Sanjay Ghemawat, Wilson C. Hsieh, Deborah A. Wallach, Mike Burrows, Tushar Chandra, Andrew Fikes, Robert E. Gruber, Bigtable:A Distributed Storage System for Structured Data. ACM Trans. Comp. Syst. 26(2), Art. 4, 2008.

[10] Boris V. Cherkassky, Andrew V. Goldberg, and Tomasz Radzik, Shortest paths algorithms: Theory and experimental evaluation. Mathematical Programming 73, 1996, 129{174.

[11] Jonathan Cohen, Graph Twiddling in a MapReduce World. Comp. in Science & Engineering, July/August 2009, 29{41.

[12] Joseph R. Crobak, Jonathan W. Berry, Kamesh Madduri, and David A. Bader, Advanced Shortest Paths Algorithms on a Massively-Multithreaded Architecture. in Proc. First Workshop on Multithreaded Architectures and Applications, 2007,1{8.

[13] John T. Daly, A higher order estimate of the optimum checkpoint interval for restart dumps. Future Generation Computer Systems 22, 2006, 303{312.

[14] Je_rey Dean and Sanjay Ghemawat, MapReduce:Simpli_ed Data Processing on Large Clusters. in Proc.6th USENIX Symp. on Operating Syst. Design and Impl., 2004, 137{150.

[15] Edsger W. Dijkstra, A Note on Two Problems in Connexion with Graphs. Numerische Mathematik 1,1959, 269{271.

[16] Martin Erwig, Inductive Graphs and Functional Graph Algorithms. J. Functional Programming 1(5), 2001,467{492.

[17] Lester R. Ford, L. R. and Delbert R. Fulkerson, Flows in Networks. Princeton University Press, 1962.

[18] Ian Foster and Carl Kesselman (Eds), The Grid 2:Blueprint for a New Computing Infrastructure (2nd edition). Morgan Kaufmann, 2003.

[19] Sanjay Ghemawat, Howard Gobio_, and Shun-Tak Leung, The Google File System. in Proc. 19th ACM Symp. on Operating Syst. Principles, 2003, 29{43.

[20] Michael T. Goodrich and Roberto Tamassia, Data Structures and Algorithms in JAVA. (second edition).John Wiley and Sons, Inc., 2001.

[21] Mark W. Goudreau, Kevin Lang, Satish B. Rao,Torsten Suel, and Thanasis Tsantilas, Portable and E_cient Parallel Computing Using the BSP Model.IEEE Trans. Comp. 48(7), 1999, 670

[22] Douglas Gregor and Andrew Lumsdaine, The Parallel BGL: A Generic Library for Distributed Graph Computations. Proc. of Parallel Object-Oriented Scienti_c Computing (POOSC), July 2005.

[23] Douglas Gregor and Andrew Lumsdaine, Lifting Sequential Graph Algorithms for Distributed-Memory Parallel Computation. in Proc. 2005 ACM SIGPLAN Conf. on Object-Oriented Prog., Syst., Lang., and Applications (OOPSLA'05), October 2005, 423{437.

[24] Jonathan L. Gross and Jay Yellen, Graph Theory and Its Applications. (2nd Edition). Chapman and Hall/CRC, 2005.

[25] Aric A. Hagberg, Daniel A. Schult, and Pieter J. Swart, Exploring network structure, dynamics, and function using NetworkX. in Proc. 7th Python in Science Conf., 2008, 11{15.

[26] Jonathan Hill, Bill McColl, Dan Stefanescu, Mark Goudreau, Kevin Lang, Satish Rao, Torsten Suel,Thanasis Tsantilas, and Rob Bisseling, BSPlib: The BSP Programming Library. Parallel Computing 24,1998, 1947.

[27] Michael Isard, Mihai Budiu, Yuan Yu, Andrew Birrell,and Dennis Fetterly, Dryad: Distributed Data-Parallel Programs from Sequential Building Blocks. in Proc.European Conf. on Computer Syst., 2007, 59.

[28] Paris C. Kanellakis and Alexander A. Shvartsman,Fault-Tolerant Parallel Computation. Kluwer Academic Publishers, 1997.

[29] Donald E. Knuth, Stanford GraphBase: A Platform for Combinatorial Computing. ACM Press, 1994.

[30] U Kung, Charalampos E. Tsourakakis, and Christos Faloutsos, Pegasus: A Peta-Scale Graph Mining System - Implementation and Observations. Proc. Intl.Conf. Data Mining, 2009, 229-238.

[31] Andrew Lumsdaine, Douglas Gregor, Bruce Hendrickson, and Jonathan W. Berry, Challenges in Parallel Graph Processing. Parallel Processing Letters 17, 2007, 5

[32] Kamesh Madduri, David A. Bader, Jonathan W.Berry, and Joseph R. Crobak, Parallel Shortest Path Algorithms for Solving Large-Scale Graph Instances.DIMACS Implementation Challenge { The Shortest Path Problem, 2006.

[33] Kamesh Madduri, David Ediger, Karl Jiang, David A.Bader, and Daniel Chavarria-Miranda, A Faster Parallel Algorithm and E_cient Multithreaded Implementation for Evaluating Betweenness Centrality on Massive Datasets, in Proc. 3rd Workshop on Multithreaded Architectures and Applications (MTAAP'09), Rome, Italy, May 2009.

[34] Grzegorz Malewicz, A Work-Optimal Deterministic Algorithm for the Certi_ed Write-All Problem with a Nontrivial Number of Asynchronous Processors. SIAM J. Comput. 34(4), 2005, 993

[35] Kurt Mehlhorn and Stefan Naher, The LEDA Platform of Combinatorial and Geometric Computing.Cambridge University Press, 1999.

[36] Ulrich Meyer and Vitaly Osipov, Design and Implementation of a Practical I/O-e_cient Shortest Paths Algorithm. in Proc. 3rd Workshop on Multithreaded Architectures and Applications (MTAAP'09), Rome, Italy, May 2009.

[37] Ulrich Meyer and Peter Sanders, _-stepping: A Parallelizable Shortest Path Algorithm. J. Algorithms 49(1), 2003, 114

[38] Richard Miller, A Library for Bulk-Synchronous Parallel Programming. in Proc. British Computer Society Parallel Processing Specialist Group Workshop on General Purpose Parallel Computing, 1993.

[39] Kameshwar Munagala and Abhiram Ranade, I/O-complexity of graph algorithms. in Proc. 10th Annual ACM-SIAM Symp. on Discrete Algorithms,1999, 687

[40] Christopher Olston, Benjamin Reed, Utkarsh Srivastava, Ravi Kumar, and Andrew Tomkins, Pig Latin: A Not-So-Foreign Language for Data Processing. in Proc. ACM SIGMOD Intl. Conf. on Management of Data, 2008, 1099

[41] Rob Pike, Sean Dorward, Robert Griesemer, and Sean Quinlan, Interpreting the Data: Parallel Analysis with Sawzall. Scienti_c Programming Journal 13(4), Special Issue on Grids and Worldwide Computing Programming Models and Infrastructure, 2005,227

[42] Protocol Bu_ers|Google's data interchange format.http://code.google.com/p/protobuf/ 2009.

[43] Jeremy G. Siek, Lie-Quan Lee, and Andrew Lumsdaine, The Boost Graph Library: User Guide and Reference Manual. Addison Wesley, 2002.

[44] Mikkel Thorup, Undirected Single-Source Shortest Paths with Positive Integer Weights in Linear Time.J. ACM 46(3), May 1999, 362

[45] Leslie G. Valiant, A Bridging Model for Parallel Computation. Comm. ACM 33(8), 1990, 103

[46] Andy Yoo, Edmond Chow, Keith Henderson, William McLendon, Bruce Hendrickson, and Umit Catalyurek,A Scalable Distributed Parallel Breadth-First Search Algorithm on BlueGene/L, in Proc. 2005 ACM/IEEE Conf. on Supercomputing (SC'05), 2005, 25|43.

[47] Yuan Yu, Michael Isard, Dennis Fetterly, Mihai Budiu,Ulfar Erlingsson, Pradeep Kumar Gunda, and Jon Currey, DryadLINQ: A System for General-Purpose Distributed Data-Parallel Computing Using a High-Level Language. in Proc. 8th USENIX Symp. On Operating Syst. Design and Implementation, 2008,