參考廈門課程linziyu-bigdata book chapter9graph computing

1、獲取和講義 PPT 等各種課程資料請http:/node/422=課程由林子雨老師根據(jù)網(wǎng)絡(luò)資料編著=廈門大學(xué)計(jì)算機(jī)科學(xué)系教師 林子雨 編著 HYPERLINK http:/w/ http:/w/linziyu2013 年 9 月1 / 33前言由廈門大學(xué)計(jì)算機(jī)科學(xué)系教師林子雨編著，可以作為計(jì)算機(jī)專業(yè)本課程大數(shù)據(jù)技術(shù)基礎(chǔ)的輔助。本的主要內(nèi)容包括：大數(shù)據(jù)概述、大數(shù)據(jù)處理模型、大數(shù)據(jù)、大數(shù)據(jù)時(shí)代的新、NoSQL 數(shù)據(jù)庫、云數(shù)據(jù)庫、Spanner、Hadoop、HDFS、HBase、MapReduce、Zookeeper、流計(jì)算、圖計(jì)算和Dremel 等。本是林子雨通過大量閱讀、收集、整理各種資料后精

2、心制作的學(xué)習(xí)材料，與廣大數(shù)據(jù)庫者共享。中的內(nèi)容大部分來自網(wǎng)絡(luò)資料和書籍，一部分是自己撰寫。對于自寫內(nèi)容，林子雨老師擁有著作權(quán)。本PDF 文檔及其教學(xué)PPT 可以通過網(wǎng)絡(luò)免費(fèi)和使用（地址：http:/node/422）中可能存在一些問題，歡迎讀者提出寶貴意見和建議！本已經(jīng)應(yīng)用于廈門大學(xué)計(jì)算機(jī)科學(xué)系課程大數(shù)據(jù)技術(shù)基礎(chǔ)，歡迎2013 班級http:/node/423。林子雨的是：z。林子雨的個(gè)人主頁是： HYPERLINK http:/w/ http:/w/linziyu。林子雨于廈門大學(xué)園2013 年 9 月2 / 33第 9 章圖計(jì)算廈門大學(xué)計(jì)算機(jī)科學(xué)系教師林子雨 編著個(gè)人主頁： HYPERLI

3、NK http:/w/ http:/w/linziyu課程：http:/node/4222013 年 9 月3 / 33第 9 章圖計(jì)算隨著大數(shù)據(jù)時(shí)代的到來，圖的規(guī)模越來越大，有的甚至有數(shù)十億的頂點(diǎn)和數(shù)千億的邊，這就給高速地處理圖數(shù)據(jù)帶來了。一臺(tái)機(jī)器已經(jīng)不能存放所有需要計(jì)算的數(shù)據(jù)了，所以需要一個(gè)分布式的計(jì)算環(huán)境。而已有的圖計(jì)算框架和圖算法庫不能很好地滿足計(jì)算需求，因此，新的圖計(jì)算框架應(yīng)運(yùn)而生。本章內(nèi)容首先簡單介紹了圖計(jì)算，然后詳細(xì)介紹了當(dāng)前熱門的圖計(jì)算框架Pregel，包括 Pregel 圖計(jì)算模型、Pregel 中的C+ API、Pregel 的執(zhí)行過程和Pregel 的算法實(shí)現(xiàn)，內(nèi)容要點(diǎn)如

4、下：圖計(jì)算簡介Pregel 簡介Pregel 圖計(jì)算模型Pregel 的C+ APIPregel 模型的基本體系結(jié)構(gòu)Pregel 模型的應(yīng)用實(shí)例改進(jìn)的圖計(jì)算模型9.1 圖計(jì)算簡介在實(shí)際應(yīng)用中，存在許多圖計(jì)算問題，比如最短路徑、集群、網(wǎng)頁、最小切割、連通分支等等。圖計(jì)算算法的性能，直接關(guān)系到應(yīng)用問題解決的高效性，尤其對于大型圖（比如社交網(wǎng)絡(luò)和網(wǎng)絡(luò)圖）而言，更是如此。下面首先傳統(tǒng)圖計(jì)算解決方案的之處，然后介紹兩大類通用圖計(jì)算。9.1.1 傳統(tǒng)圖計(jì)算解決方案的之處在很長一段時(shí)間內(nèi)，都缺少一個(gè)可擴(kuò)展的通用系統(tǒng)來解決大型圖的計(jì)算問題。很多傳統(tǒng)的圖計(jì)算算法都存在以下幾個(gè)典型問題：（1）常常局部性；（2）針

5、比較差的內(nèi)存4 / 33對單個(gè)頂點(diǎn)的處理工作過少；（3）計(jì)算過程中伴隨著并行度的改變。針對大型圖（比如社交網(wǎng)絡(luò)和網(wǎng)絡(luò)圖）的計(jì)算問題，可能的解決方案及其之處具體如下：為特定的圖應(yīng)用定制相應(yīng)的分布式實(shí)現(xiàn)。之處是，在面對新的圖算法或者圖表示方式時(shí)，就需要做大量的重復(fù)實(shí)現(xiàn)，不通用。基于現(xiàn)有的分布式計(jì)算進(jìn)行圖計(jì)算。但是，在這種情況下，它們往往并不適于做圖處理。比如，MapReduce 就是一個(gè)對許多大規(guī)模計(jì)算問題都非常合適的計(jì)算框架。有時(shí)，它也被用來對大規(guī)模圖對象進(jìn)行挖掘，但是，通常在性能和易用性上都不是最優(yōu)的。盡管這種對數(shù)據(jù)處理的基本模式經(jīng)過擴(kuò)展，已經(jīng)可以使用方便的聚合以及類似于 SQL 的查詢方式，

6、但是，這些擴(kuò)展對于圖算法這種更適合用消息傳遞模型來說，通常并不理想。使用單機(jī)的圖算法庫。比如 BGL、LEAD、NetworkX、JDSL、Standford GraphBase和 FGL 等等；但是，這種方式對可以解決的規(guī)模提出了很大的限制。使用已有的并行圖計(jì)算系統(tǒng)。Parallel BGL 和 CGMgraph 這些庫實(shí)現(xiàn)了很多并行圖算法，但是，并沒有解決對大規(guī)模分布式系統(tǒng)中來說非常重要的容錯(cuò)等一些問題。9.1.2 圖計(jì)算通用正是因?yàn)閭鹘y(tǒng)的圖計(jì)算解決方案無法解決大型圖的計(jì)算問題，因此，就需要設(shè)計(jì)能夠用來解決這些問題的通用圖計(jì)算。針對大型圖的計(jì)算，目前通用的圖處理主要包括兩種：第一種主要是基

7、于遍歷算法和實(shí)時(shí)的圖數(shù)據(jù)庫，如 Neo4j 、OrientDB、DEX 和InfiniteGraph。第二種則是以圖頂點(diǎn)為中心的消息傳遞批處理的并行引擎，如 Hama、GoldenOrb、Giraph 和Pregel。第一種圖處理，基本都基于 Tinkop 的圖基礎(chǔ)框架，Tinkop 項(xiàng)目關(guān)系如圖 9-1所示。5 / 33圖 9-1 Tinkop 項(xiàng)目關(guān)系圖下面具體介紹一下 Tinkop 框架的各層功能：Blueprs：是一組針對屬性圖數(shù)據(jù)模型的接口、實(shí)現(xiàn)、測試套件，它和 JDBC 類似，但是，它是基于圖數(shù)據(jù)庫的。就其本身而言，它提供了一組通用的接口，允許開發(fā)者對其圖數(shù)據(jù)庫即插即用。另外，在

8、Blueprs 上編寫的可以運(yùn)行于所有的 Blueprs 開啟的圖數(shù)據(jù)庫。在Tinkop 的圖基礎(chǔ)框架中，Blueprs 為其他幾層提供基礎(chǔ)技術(shù)服務(wù)。Pipes：是一個(gè)應(yīng)用流程圖的數(shù)據(jù)流框架。一個(gè)流程圖由很多通過“通信邊”相連的 pipe 頂點(diǎn)組成。一個(gè) pipe 實(shí)現(xiàn)了一個(gè)簡單的計(jì)算步驟，它可以和其他的 pipe 相組合，一起產(chǎn)生一個(gè)更大的計(jì)算。這樣的數(shù)據(jù)流圖允許拆分、合并、循環(huán)和輸入輸出數(shù)據(jù)的相互轉(zhuǎn)換。伴隨著主 Pipe 的分布，會(huì)產(chǎn)生大量的 Pipe 類。只要了解了每個(gè) Pipe 的實(shí)現(xiàn)，就可以直接使用 Pipe 框架。Gremlin：是一種圖遍歷語言，可以用于圖的查詢、分析和處理。它工

9、作在那些執(zhí)行 Blueprs 性能圖數(shù)據(jù)模型的圖數(shù)據(jù)庫和框架上。Gremlin 是能用于各種 JVM 語言的一種圖遍歷。Gremlin 的布局為 Java 和Groovy 提供了支持。Frames：把 Blueprs 圖為一組相互關(guān)聯(lián)的域?qū)ο蠹?。Frames 中經(jīng)常使用InvocationHandlorxy 類和 Annoions，使開發(fā)者可以用一個(gè)特殊的 Java 接口來構(gòu)造一個(gè)圖元素（頂點(diǎn)或邊）。通過 Frames，非常容易確認(rèn)圖中數(shù)據(jù)各自的圖解，對應(yīng)哪一個(gè)帶有注釋的 Java 接口。Furnace：是能啟用 Blueprs 圖的一個(gè)算法包。在圖理論和分析的歷史進(jìn)程中開發(fā)6 / 33了很多

10、的圖算法。這些算法中的大多數(shù)是為無標(biāo)號(hào)的、單一關(guān)系的圖而設(shè)計(jì)的。Furnace 的目的就是在單一關(guān)系圖算法中揭示屬性圖（像屬性化圖或多關(guān)系圖）。另外，F(xiàn)urnace 提供了針對不同圖計(jì)算場景，各種圖算法的優(yōu)化實(shí)現(xiàn)，像單機(jī)圖和分布圖。Rexster：是一個(gè)圖服務(wù)器，通過 REST 和一個(gè)被稱為 RexPro 的二進(jìn)制協(xié)議展現(xiàn)任意的 Blueprs 圖。HTTP 網(wǎng)頁服務(wù)器提供了標(biāo)準(zhǔn)的低層 GET、T、PUT 和 DELETE方法，一個(gè)靈活的、像開發(fā)一個(gè)外部服務(wù)器（如通過 Gremlin 的特殊圖查詢）一樣允許插件法的擴(kuò)展模型，用 Gremlin 編寫的服務(wù)器端程序和一個(gè)基于瀏覽器的接口Dog H

11、ouse。Rexster Console 使對 Rexster 服務(wù)器中的配置圖的評估成為可能。Rexster Kibbles 是由 Tinkop 提供的各種 Rexster 服務(wù)器的擴(kuò)展集。第二種圖處理，則主要是基于 BSP 模型所實(shí)現(xiàn)的并行圖處理包。BSP 是由哈佛大學(xué) Viliant 和牛津大學(xué) Bill McColl并行計(jì)算模型，全稱為“整體同步并行計(jì)算模型”(Bulk Synchronous Parallel Computing M,簡稱 BSP 模型),又名“大同步模型”。創(chuàng)始人希望BSP 模型像馮體系結(jié)構(gòu)那樣，架起計(jì)算機(jī)程序語言和體系結(jié)構(gòu)間的橋梁，故又稱作“橋模型” (Bridg

12、e M)。一個(gè)BSP 模型由大量相互關(guān)聯(lián)的處理器所組成，它們之間形成了一個(gè)通信網(wǎng)絡(luò)。每個(gè)處理器都有快速的本地內(nèi)存和不同的計(jì)算線程。一次 BSP 計(jì)算過程包括一系列全局超步，所謂的超步就是計(jì)算中的一次迭代。每個(gè)超步主要包括三個(gè)組件：并發(fā)計(jì)算：每個(gè)參與的處理器都有自身的計(jì)算任務(wù)，它們只在本地內(nèi)存的值，這些計(jì)算都是異步并且獨(dú)立的；通訊：處理器群相互交換的形式是，由一方發(fā)起推送(put)和獲取(get)操作；柵欄同步(Barrier synchronisation): 當(dāng)一個(gè)處理器遇到“路障”，會(huì)等到其他所有處理器完成它們的計(jì)算步驟；每一次同步也是一個(gè)超步的完成和下一個(gè)超步的開始；圖 9-2 是一個(gè)超

13、步的垂直結(jié)構(gòu)圖。7 / 33圖 9-2 一個(gè)超步的垂直結(jié)構(gòu)圖Pregel 簡介9.2Pregel就是一種基于BSP模型所實(shí)現(xiàn)的并行圖處理包。為了解決大型圖的分布式計(jì)算問題，Pregel搭建了一套可擴(kuò)展的、有容錯(cuò)機(jī)制的，該提供了一套非常靈活的API，可以描述各種各樣的圖計(jì)算。Pregel是一個(gè)用于分布式圖計(jì)算的計(jì)算框架，主要用于圖遍歷（BFS）、最短路徑（SSSP）、PageR計(jì)算等等。共享內(nèi)存的運(yùn)行庫有很多，但是，對于來說，一臺(tái)機(jī)器早已經(jīng)放不下需要計(jì)算的數(shù)據(jù)了，所以，需要分布式這樣一個(gè)計(jì)算環(huán)境。沒有Pregel之前， 你可以選擇用MapReduce 來做，但是效率很低。圖算法如果用MapRed

14、uce實(shí)現(xiàn)，需要一系列的MapReduce的調(diào)用。從一個(gè)階段到下一個(gè)階段，它需要傳遞整個(gè)圖的狀態(tài)，會(huì)產(chǎn)生大量不必要的序列化和反序列化開銷。而Pregel使用超步簡化了這個(gè)過程。下面以一個(gè)實(shí)例來闡述采用Pregel和MapReduce來執(zhí)行圖計(jì)算的區(qū)別。實(shí)例：PageR算法在Pregel和MapReduce中的實(shí)現(xiàn)。PageR算法作為的網(wǎng)頁算法，具體公式如下：對于任意一個(gè)，其PR值為鏈入到該的源的PR值對該的貢獻(xiàn)和（分母8 / 33Ni為第i個(gè)源的鏈出度）。Pregel是專門為圖計(jì)算所設(shè)計(jì)的計(jì)算模型，主要來源于BSP并行計(jì)算模型的啟發(fā)。要用Pregel計(jì)算模型實(shí)現(xiàn)PageR算法，也就是將網(wǎng)頁算法

15、到圖計(jì)算中，這其實(shí)是很自然的，因?yàn)?，網(wǎng)絡(luò)是通圖。圖 9-3通圖圖9-3就是四個(gè)網(wǎng)頁（A,B,C,D）互相鏈入鏈出組成的連通圖。根據(jù)Pregel的計(jì)算模型，將計(jì)算定義到頂點(diǎn)（即A，B，C，D）上來，每個(gè)頂點(diǎn)對應(yīng)一個(gè)對象，即一個(gè)計(jì)算單元。每一個(gè)計(jì)算單元包含三個(gè)成員變量：Vertex value：頂點(diǎn)對應(yīng)的PR值；Out edge：只需要表示一條邊，可以不取值；Message：傳遞的消息，因?yàn)樾枰獙⒈卷旤c(diǎn)對其它頂點(diǎn)的PR貢獻(xiàn)值，傳遞給目標(biāo)頂點(diǎn)。每一個(gè)計(jì)算單元包含一個(gè)成員函數(shù)：Compute：該函數(shù)定義了頂點(diǎn)上的運(yùn)算，包括該頂點(diǎn)的PR值計(jì)算，以及從該頂點(diǎn)發(fā)送消息到其鏈出頂點(diǎn)。PageR算法的Prege

16、l實(shí)現(xiàn)代碼如下：9 / 33class PageRVertex: public Vertex public:virtual void Compute(MessageIterator* msgs) if (superstep() = 1) double sum = 0;for (; !msgs-Done(); msgs-Next() sum += msgs-Value();*MutableValue() =0.15 / NumVerti() + 0.85 * sum;if (superstep() 30) const64 n = GetOutEdgeIterator().size();PageR

17、Vertex 繼承自 Vertex 類。頂點(diǎn) value 類型是 double，用來保存 PageR中間值，消息類型也是 double，用來傳輸 PageR分?jǐn)?shù)，邊的 value 類型是 void，因?yàn)椴恍枰魏涡畔?。假設(shè)，在第 0 個(gè)超步時(shí)，圖中各頂點(diǎn)的 value 值被初始化為1/NumVerti()。30 個(gè)超步中，每個(gè)頂點(diǎn)都會(huì)沿著它的出射邊，發(fā)送它的 PageR值除以出射邊數(shù)目以后的結(jié)果值。從第 1 個(gè)超步開始，每個(gè)頂點(diǎn)會(huì)將到達(dá)的消息中的值加到sum 值中，同時(shí)將它的 PageR值設(shè)為 0.15/ NumVerti()+0.85*sum。到了第 30 個(gè)超步后，就沒有需要發(fā)送的消息了，

18、同時(shí)所有的頂點(diǎn) VoteToHalt。在實(shí)際中，PageR算法需要一直運(yùn)行直到收斂，可以使用 aggregators 來檢查是否滿足收斂條件。MapReduce 也是一種計(jì)算模型，它是為全量計(jì)算而設(shè)計(jì)。采用 MapReduce實(shí)現(xiàn) PageR的計(jì)算過程需要包括以下三個(gè)階段：階段 1：網(wǎng)頁Map task 把（URL，page content）為（URL，（PRinit，list-of-urls），其中，PRinit是URL 的“seed”PageR，list-of-urls 包含了該 URL 頁面中的外鏈所指向的所有頁。Reducetask 只是恒等函數(shù)。在階段 1 中，對于 MapReduc

19、e 程序，Map 函數(shù)的輸入是用戶指定的文件（這里就是一個(gè)網(wǎng)頁），那么，輸入的 value 值就是網(wǎng)頁內(nèi)容，key 值是網(wǎng)頁的 url。程序?qū)斎氲?key 和value 進(jìn)行一系列操作，然后，按(key,value)鍵值對的形式輸出。系統(tǒng)獲取 Map 函數(shù)的輸出，把相同的 key 合并，再把 key 和 value 集合作為鍵值對作為 reduce 函數(shù)的輸入。程序員自定義 reduce 函數(shù)中的處理方法（這里是一個(gè)恒等函數(shù)），輸出(key,value)鍵值對到磁盤文件（這里的鍵值對中的 key 仍是該網(wǎng)頁的 url，value 是該頁的初始PR 值和鏈出頁的 url 列表）。階段 2：Pa

20、geR分配Map task 得到（URL，（cur_r，url_list）， 對于每一個(gè) url_list 中的 u，輸出（u，cur_r/|url_list|），并輸出關(guān)系（URL，url_list），用于迭代。Reduce task 獲得（URL，url_list）和很多（URL，var）值對，對于具有相同 key 值的10 / 33SendMessageToAllNeighbors(GetValue() / n); else VoteToHalt();value 進(jìn)行匯總，并把匯總結(jié)果乘以 d(這里是 0.85)，然后輸出輸出（URL，（new_r，url_list）。階段 2 是一個(gè)迭

21、代過程，每次將磁盤上的文件讀入，key 為每一個(gè)網(wǎng)頁，value 的第一項(xiàng)是當(dāng)前網(wǎng)頁的 PR 值，第二項(xiàng)是該網(wǎng)頁中的外鏈列表。Map 函數(shù)將輸入的每一對(key,value)“反轉(zhuǎn)”對(value,key)輸出，也就是說，每個(gè)輸出的 key 為輸入 value 中的每一個(gè)外鏈，而輸出的 value 則為輸入 key 的、輸入 key 的 PR 值除以輸入 key 的對外鏈接數(shù)。在 Reduce 函數(shù)中，就可以根據(jù)輸入的 value 中的參數(shù)來計(jì)算每一個(gè) key 新的 PageR值，并按 Map 函數(shù)的格式輸出到磁盤，以作為下一次迭代的輸入。迭代多次后，PageR值趨于穩(wěn)定，就得出了較為精確的

22、PageR值。階段 3：最后階段一個(gè)非并行組件決定是否達(dá)到收斂。如果達(dá)到收斂，寫出 PageR生成的列表。否則，回退到第 2 階段的輸出，進(jìn)行另一個(gè)第 2 階段的迭代。下面具體解釋一下階段 2 的偽代碼實(shí)現(xiàn)：11 / 33Mapper 函數(shù)的偽碼：input - PageA, PageB, PageC . /關(guān)系beginNn := the number of outlinks for PageN; for each outlink PageKoutput PageK - / 同時(shí)輸出關(guān)系，用于迭代output PageN - PageA, PageB, PageC .endMapper 的輸

23、出如下（已經(jīng)排序，所以 PageK 的數(shù)據(jù)排在一起，最后一列則是關(guān)系對）：PageK - PageK - .PageK - 上述偽代碼只是一次迭代的代碼，多次迭代需要重復(fù)運(yùn)行，需明的是這里可以優(yōu)化的地方很多，比如把 Mapper 的內(nèi)容緩存起來，這樣就不用再output 作為 Reducer 的 input 了，同時(shí)，Reducer 在 output 的時(shí)候也不用傳遞同樣的，減少了大量的 I/O，因?yàn)?，?PageR計(jì)算是，類似這樣的數(shù)據(jù)才是主要數(shù)據(jù)。簡單地來講，Pregel 將 PageR處理對象看成是連通圖，而 MapReduce 則將其看成是Key-Value 對。Pregel 將計(jì)算細(xì)

24、化到頂點(diǎn) vertex，同時(shí)在 vertex 內(nèi)控制循環(huán)迭代次數(shù)，而MapReduce 則將計(jì)算批量化處理，按任務(wù)進(jìn)行循環(huán)迭代控制。圖算法如果用 MapReduce 實(shí)現(xiàn)，需要一系列的 MapReduce 的調(diào)用。從一個(gè)階段到下一個(gè)階段，它需要傳遞整個(gè)圖的狀態(tài)，會(huì)產(chǎn)生大量不必要的序列化和反序列化開銷。而Pregel 使用超步簡化了這個(gè)過程。9.3Pregel 圖計(jì)算模型9.3.1 Pregel 的消息傳遞模型Pregel 選擇了一種純消息傳遞模型（如圖 9-4 所示），忽略數(shù)據(jù)和其他共享內(nèi)存的方式。這樣做的原因有兩個(gè)：第一，消息傳遞有足夠的表達(dá)能力，沒必要使用，還沒有發(fā)現(xiàn)哪種算法是消息傳遞所不

25、能表達(dá)的；第二是出于性能的考慮，在一個(gè)集群環(huán)境中，機(jī)器上一個(gè)值是會(huì)有很高的延遲的，這種情況很難避免，而Pregel 的消息傳遞從模式通過異步和批量的方式傳遞消息，可以緩解這種的延遲。12 / 33Reduce 函數(shù)的偽碼： input mappers output beginRK := 0;for each inlink PageNiRK += RNi/Nni * beta/ output the PageK and its new Rfor the next iteration output - end圖 9-2 純消息傳遞模型圖9.3.2 Pregel 的計(jì)算過程Pregel 的計(jì)算過程是

26、由一系列被稱為超步(superstep)的迭代(iterations)組成的。在每一個(gè)超步中，計(jì)算框架都會(huì)針對每個(gè)頂點(diǎn)調(diào)用用戶自定義的函數(shù)，這個(gè)過程是并行的。該函數(shù)描述的是一個(gè)頂點(diǎn) V 在一個(gè)超步 S 中需要執(zhí)行的操作。該函數(shù)可以前一個(gè)超步(S-1)中發(fā)送給 V 的消息，并發(fā)送消息給其他頂點(diǎn)，這些消息將會(huì)在下一個(gè)超步(S+1)中被接收，并且在此過程中修改頂點(diǎn) V 及其出射邊的狀態(tài)。消息通常沿著頂點(diǎn)的出射邊發(fā)送，但一個(gè)消息可能會(huì)被發(fā)送到任意已知 ID 的頂點(diǎn)上去。9.3.3 Pregel 計(jì)算模型的實(shí)體在 Pregel 計(jì)算模型中，輸入是一個(gè)有向圖，該有向圖的每一個(gè)頂點(diǎn)都有一個(gè)相應(yīng)的由Strin

27、g 描述的頂點(diǎn)標(biāo)識(shí)符。每一個(gè)頂點(diǎn)都有一個(gè)與之對應(yīng)的可修改的用戶自定義值。每一條有向邊都和其源頂點(diǎn)關(guān)聯(lián)，并且也擁有一個(gè)可修改的用戶自定義值，并同時(shí)了其目標(biāo)頂點(diǎn)的標(biāo)識(shí)符。在每個(gè)超步中，頂點(diǎn)的計(jì)算都是并行的，每個(gè)頂點(diǎn)執(zhí)行相同的用于表達(dá)給定算法邏輯的用戶自定義函數(shù)。每個(gè)頂點(diǎn)可以修改其自身及其出射邊的狀態(tài)，接收前一個(gè)超步(S-1)中發(fā)送給它的消息，并發(fā)送消息給其他頂點(diǎn)(這些消息將會(huì)在下一個(gè)超步中被接收)，甚至是修改整個(gè)圖的拓?fù)浣Y(jié)構(gòu)。在這種計(jì)算模式中，“邊”并不是對象，沒有相應(yīng)的計(jì)算運(yùn)行在其上。13 / 339.3.4 Pregel 計(jì)算模型的進(jìn)程算法是否能夠結(jié)束，取決于是否所有的頂點(diǎn)都已經(jīng)“vote”標(biāo)

28、識(shí)其自身已經(jīng)達(dá)到“halt”狀態(tài)了。在第 0 個(gè)超步，所有頂點(diǎn)都處于 active 狀態(tài)，所有的 active 頂點(diǎn)都會(huì)參與所有對應(yīng)超步中的計(jì)算。頂點(diǎn)通過將其自身的 sus 設(shè)置成“halt”來表示它已經(jīng)不再active。這就表示該頂點(diǎn)沒有進(jìn)一步的計(jì)算需要去執(zhí)行，除非再次被外部觸發(fā)，而 Pregel 框架將不會(huì)在接下來的超步中執(zhí)行該頂點(diǎn)，除非該頂點(diǎn)收到其它頂點(diǎn)傳送的消息。如果頂點(diǎn)接收到消息被喚醒進(jìn)入 active 狀態(tài)，那么在隨后的計(jì)算中該頂點(diǎn)必須顯式地“deactive”。整個(gè)計(jì)算在所有頂點(diǎn)都達(dá)到“inactive”狀態(tài)，并且沒有 message 在傳送的時(shí)候才結(jié)束。這種簡單的狀態(tài)機(jī)如圖 9

29、-5 所示。圖 9-5 一個(gè)簡單的狀態(tài)機(jī)圖下面通過一個(gè)簡單的例子來說明這些基本概念。給定一個(gè)強(qiáng)連通圖（如圖 9-6 所示），圖中每個(gè)頂點(diǎn)都包含一個(gè)值，它會(huì)將最大值到每個(gè)頂點(diǎn)。在每個(gè)超步中，頂點(diǎn)會(huì)從接收到的消息中選出一個(gè)最大值，并將這個(gè)值傳送給其所有的相鄰頂點(diǎn)。當(dāng)某個(gè)超步中已經(jīng)沒有頂點(diǎn)更新其值，那么算法就結(jié)束。圖 9-6 一個(gè)求最大值的步驟圖14 / 339.4 Pregel 的 C+ API編寫一個(gè)Pregel 程序需要繼承Pregel 中已預(yù)定義好的一個(gè)基類Vertex 類，如下所示：該類的模板參數(shù)中定義了三個(gè)值類型參數(shù)，分別表示頂點(diǎn)、邊和消息。每一個(gè)頂點(diǎn)都有一個(gè)對應(yīng)的給定類型的值。這種形式

30、可能看上有很多限制，但用戶可以用protocol buffer 來管理增加的其他定義和屬性。而邊和消息類型的行為比較類似。用戶覆寫 Vertex 類的虛函數(shù) Compute()，該函數(shù)會(huì)在每一個(gè)超步中對每一個(gè)頂點(diǎn)進(jìn)行調(diào)用。預(yù)定義的 Vertex 類方法允許 Compute()方法查詢當(dāng)前頂點(diǎn)及其邊的信息，以及發(fā)送消息到其他的頂點(diǎn)。Compute()方法可以通過調(diào)用 GetValue()方法來得到當(dāng)前頂點(diǎn)的值，或者通過調(diào)用 MutableValue()方法來修改當(dāng)前頂點(diǎn)的值。同時(shí)還可以通過由出射邊的迭代器提供的方法來查看修改出射邊對應(yīng)的值。這種狀態(tài)的修改是立時(shí)可見的。由于這種可見性僅限于被修改的

31、那個(gè)頂點(diǎn)，所以，不同頂點(diǎn)并發(fā)進(jìn)行的數(shù)據(jù)是不存在競爭關(guān)系的。頂點(diǎn)和其對應(yīng)的邊所關(guān)聯(lián)的值，是唯一需要在超步之間持久化的頂點(diǎn)級狀態(tài)。將由計(jì)算框架管理的圖狀態(tài)限制在一個(gè)單一的頂點(diǎn)值或邊值的這種做法，簡化了主計(jì)算流程、圖的分布以及故障恢復(fù)。15 / 33template class Vertex public:virtual void Compute(MessageIterator* msgs) = 0; const string& vertex_id() const;64 superstep() const;const VertexValue& GetValue(); VertexValue* Mut

32、ableValue(); OutEdgeIteratetOutEdgeIterator();void SendMessageTo(const string& dest_vertex, const MessageValue& message);void VoteToHalt();9.4.1 消息傳遞機(jī)制頂點(diǎn)之間的通信是直接通過發(fā)送消息來實(shí)現(xiàn)的，每條消息都包含了消息值和目標(biāo)頂點(diǎn)的名稱。消息值的數(shù)據(jù)類型是由用戶通過 Vertex 類的模版參數(shù)來指定。在一個(gè)超步中，一個(gè)頂點(diǎn)可以發(fā)送任意多的消息。當(dāng)頂點(diǎn) V 的Compute()方法在 S+1 超步中被調(diào)用時(shí)，所有在 S 超步中發(fā)送給頂點(diǎn) V 的消息都可

33、以通過一個(gè)迭代器來到。在該迭代器中并不保證消息的順序，但是，可以保證消息一定會(huì)被傳送并且不會(huì)重復(fù)。一種通用的使用方式為：對一個(gè)頂點(diǎn) V，遍歷其自身的出射邊，向每條出射邊發(fā)送消息到該邊的目標(biāo)頂點(diǎn)，如 PageR算法所示的那樣。但是，消息要發(fā)送到的目標(biāo)頂點(diǎn)dest_vertex，并不一定是頂點(diǎn) V 的相鄰頂點(diǎn)。一個(gè)頂點(diǎn)可以從之前收到的消息中獲取到其非相鄰頂點(diǎn)的標(biāo)識(shí)符，或者頂點(diǎn)標(biāo)識(shí)符可以隱式地得到。比如，圖可能是一個(gè) clique(一個(gè)圖中兩兩相鄰的一個(gè)點(diǎn)集，或是一個(gè)完全子圖)，頂點(diǎn)名規(guī)則都是已知的(從 V1 到 Vn)，在這種情況下甚至都不需要顯式地保存邊的信息。當(dāng)任意一個(gè)消息的目標(biāo)頂點(diǎn)不存在時(shí)，

34、便執(zhí)行用戶自定義的 handlers。比如在這種情況下，一個(gè) handler 可以創(chuàng)建該不存在的頂點(diǎn)或從源頂點(diǎn)中刪除這條邊。找最大值的代碼實(shí)現(xiàn)如下：16 / 33Class MaxFindVertex: public Vertex public:virtual void Compute(MessageIterator* msgs) currMax = GetValue();SendMessageToAllNeighbors(currMax); for ( ; !msgs-Done(); msgs-Next() if (msgs-Value() currMax)currMax = msgs-Va

35、lue();if (currMax GetValue()*MutableValue() = currMax; else VoteToHalt();9.4.2 Combiner發(fā)送消息，尤其是當(dāng)目標(biāo)頂點(diǎn)在另外一臺(tái)機(jī)器上時(shí)，會(huì)產(chǎn)生一些開銷。某些情況可以在用戶的協(xié)助下降低這種開銷。比方說，假如 Compute() 收到許多的值消息，而它僅僅關(guān)心的是這些值的和，而不是每一個(gè)的值，這種情況下，系統(tǒng)可以將發(fā)往同一個(gè)頂點(diǎn)的多個(gè)消息合并成一個(gè)消息，該消息中僅包含它們的“和值”，這樣就可以減少傳輸和緩存的開銷。Combiners 在默認(rèn)情況下并沒有被開啟，這是因?yàn)橐业揭环N對所有頂點(diǎn)的 Compute()函數(shù)都

36、合適的 Combiner 是不可能的。而用戶如果想要開啟 Combiner 的功能，需要繼承Combiner 類，覆寫其 virtual 函數(shù) Combine()。框架并不會(huì)確保哪些消息會(huì)被 Combine 而哪些不會(huì)，也不會(huì)確保傳送給 Combine()的值和 Combining 操作的執(zhí)行順序。所以，Combiner只應(yīng)該對那些滿換律和結(jié)合律的操作才給予打開。觀察到通過使用 Combiner 可以把流量降對于某些算法來說，比如單源最短路徑，低 4 倍多。下面是有關(guān) combiner 應(yīng)用的例子。假設(shè)想統(tǒng)計(jì)在一組相關(guān)聯(lián)的頁面中所有頁面的數(shù)。在第一個(gè)迭代中，對從每一個(gè)頂點(diǎn)（頁面）出發(fā)的，會(huì)向目

37、標(biāo)頁面發(fā)送一個(gè)消息。這里輸入消息上的 count 函數(shù)可以通過一個(gè) combiner 來優(yōu)化性能。在上面求最大值的例子中，一個(gè) Max combiner 可以減少通信負(fù)荷（如圖 9-7 所示），比如，假設(shè)頂點(diǎn) 1 和 6在一臺(tái)機(jī)器上，頂點(diǎn) 2 和 3 在另一臺(tái)機(jī)器上，頂點(diǎn) 3 向頂點(diǎn) 6 傳遞的值是 3，頂點(diǎn) 2 向頂點(diǎn)6 傳遞的值是 2，頂點(diǎn) 2 和頂點(diǎn) 3 都需要把消息傳遞到目標(biāo)頂點(diǎn) 6，因此，可以采用 Combine()函數(shù)把這兩個(gè)消息進(jìn)行合并后再發(fā)送給目標(biāo)頂點(diǎn) 6，這樣就減少了網(wǎng)絡(luò)通信負(fù)責(zé)。17 / 33;圖 9-7 combiner 應(yīng)用的例子9.4.3 AggregatorPreg

38、el 的Aggregators 是一種提供全局通信、和數(shù)據(jù)查看的機(jī)制。在一個(gè)超步 S 中，每一個(gè)頂點(diǎn)都可以向一個(gè) Aggregator 提供一個(gè)數(shù)據(jù)，系統(tǒng)會(huì)使用一種 reduce 操作來負(fù)責(zé)聚合這些值，而產(chǎn)生的值將會(huì)對所有的頂點(diǎn)在超步 S+1 中可見。Pregel 包含了一些預(yù)定義的aggregators，比如可以在各種整數(shù)和 string 類型上執(zhí)行的 max、sum 操作。Aggregators 可以用來做統(tǒng)計(jì)。例如，一個(gè) sum aggregator 可以用來統(tǒng)計(jì)每個(gè)頂點(diǎn)的出度，最后相加就是整個(gè)圖的邊的條數(shù)。更復(fù)雜的一些 reduce 操作還可以產(chǎn)生統(tǒng)計(jì)直方圖。Aggregators 也

39、可以用來做全局協(xié)同。例如，Compute()函數(shù)的一些邏輯分支可能在某些超步中執(zhí)行，直到當(dāng)“and” aggregator 表明所有頂點(diǎn)都滿足了某條件時(shí)，會(huì)去執(zhí)行另外的邏輯分支直到結(jié)束。又比如一個(gè)作用在頂點(diǎn) ID 之上的 min 和 max aggregator，可以用來選定某頂點(diǎn)在整個(gè)計(jì)算過程中扮演某種角色等。要定義一個(gè)新的aggregator，用戶需要繼承預(yù)定義的Aggregator 類，并定義在第一次接收到輸入值后如何初始化，以及如何將接收到的多個(gè)值最后 reduce 成一個(gè)值。Aggregator操作也應(yīng)該滿換律和結(jié)合律。默認(rèn)情況下，一個(gè) aggregator 僅僅會(huì)對來自同一個(gè)超步的

40、輸入進(jìn)行聚合，但是，有時(shí)也可能需要定義一個(gè) sticky aggregator，它可以從所有的超步中接收數(shù)據(jù)。這是非常有用的，比如要全局的邊條數(shù)，那么就僅僅在增加和刪除邊的時(shí)候才調(diào)整這個(gè)值了。還可以有更高級的用法。比如，可以用來實(shí)現(xiàn)一個(gè)-step18 / 33最短路徑算法所需要的分布式優(yōu)先隊(duì)列。每個(gè)頂點(diǎn)會(huì)根據(jù)它的當(dāng)前距離分配一個(gè)優(yōu)先級 bucket。在每個(gè)超步中，頂點(diǎn)將它們的 indi匯報(bào)給 min aggregator。在下一個(gè)超步中，將最小值廣播給所有 worker，然后讓在最小 index 的 bucket 中的頂點(diǎn)放松它們的邊。一個(gè)有關(guān) aggregator 應(yīng)用的例子：Sum 運(yùn)算符

41、應(yīng)用于每個(gè)頂點(diǎn)的出射邊數(shù)，可以用來生成圖中邊的總數(shù)并使它能與所有的頂點(diǎn)相通信。更復(fù)雜的規(guī)約運(yùn)算符甚至可以產(chǎn)生直方圖。在求最大值得例子中，可以通過運(yùn)用一個(gè) Max aggregator 在一個(gè)超步中完成整個(gè)程序。9.4.4 Topology muion有一些圖算法可能需要改變圖的整個(gè)拓?fù)浣Y(jié)構(gòu)。比如一個(gè)聚類算法，可能會(huì)將每個(gè)聚類替換成一個(gè)單一頂點(diǎn)，又比如一個(gè)最小生成樹算刪除所有除了組成樹的邊之外的其他邊。正如用戶可以在自定義的Compute()函數(shù)能發(fā)送消息，同樣可以產(chǎn)生在圖中增添和刪除邊或頂點(diǎn)的請求。多個(gè)頂點(diǎn)有可能會(huì)在同一個(gè)超步中產(chǎn)生的請求(比如兩個(gè)請求都要增加一個(gè)頂點(diǎn)V，但初始值不一樣)。Pr

42、egel 中用兩種機(jī)制來決定如何調(diào)用：局部有序和 handlers。由于是通過消息發(fā)送的，拓?fù)涓淖冊谡埱蟀l(fā)出以后，在超步中可以高效地執(zhí)行。在該超步中，刪除會(huì)首先被執(zhí)行，先刪除邊，后刪除頂點(diǎn)，因?yàn)轫旤c(diǎn)的刪除通常也意味著刪除其所有的出射邊。然后執(zhí)行添加操作，先增加頂點(diǎn)，后增加邊，并且所有的拓?fù)涓淖兌紩?huì)在Compute()函數(shù)調(diào)用前完成。這種局部有序保證了大多數(shù)的結(jié)果的確定性。剩余的就需要通過用戶自定義的 handlers 來解決。如果在一個(gè)超步中有多個(gè)請求需要?jiǎng)?chuàng)建一個(gè)相同的頂點(diǎn)，在默認(rèn)情況下系統(tǒng)會(huì)隨便挑選一個(gè)請求，但有特殊需求的用戶可以定義一個(gè)更好的解決策略，用戶可以在 Vertex 類中通過定義

43、一個(gè)適當(dāng)?shù)?handler 函數(shù)來解決。同一種 handler 機(jī)制將被用于解決由于多個(gè)頂點(diǎn)刪除請求或多個(gè)邊增加請求或刪除請求而造成的。Pregel 委托 handler 來解決這種類型的，從而使得 Compute()函數(shù)變得簡單，而這樣同時(shí)也會(huì)限制 handler 和 Compute()的交互，但這在應(yīng)用中還沒有遇到什么問題。Pregel 的協(xié)同機(jī)制比較懶，全局的拓?fù)涓淖冊诒?apply 之前不需要進(jìn)行協(xié)調(diào)，即在變更請求的發(fā)出端不會(huì)進(jìn)行任何的控制協(xié)調(diào)，只有在它被接收到然后 apply 時(shí)才進(jìn)行控制，這樣就簡化了流程，同時(shí)能讓發(fā)送更快。這種設(shè)計(jì)的選擇是為了優(yōu)化流式處理。直觀來講，就是19 /

44、33對頂點(diǎn) V 的修改的由V 自己來處理。Pregel 同樣也支持純 local 的拓?fù)涓淖?，例如，一個(gè)頂點(diǎn)添加或刪除其自身的出射邊或刪除其自己。Local 的拓?fù)涓淖儾粫?huì)，并且頂點(diǎn)或邊的本地增減能夠立即生效，很大程度上簡化了分布式編程。9.4.5 Input and OutputPregel 可以采用多種文件格式進(jìn)行圖的保存，比如可以用 text 文件、關(guān)系數(shù)據(jù)庫或者Bigtable 中的行。為了避免規(guī)定死一種特定文件格式，Pregel 將“從輸入中出圖結(jié)構(gòu)”這個(gè)任務(wù)從圖的計(jì)算過程中進(jìn)行了分離。類似地，結(jié)果可以以任何一種格式輸出，并根據(jù)應(yīng)用程序選擇最適合的方式。Pregel library

45、本身提供了很多常用文件格式的 readers 和writers，但是，用戶可以通過繼承 Reader 和 Writer 類來定義他們自己的讀寫方式。9.5 Pregel 的基本體系結(jié)構(gòu)Pregel 是為的集群架構(gòu)而設(shè)計(jì)的。每一個(gè)集群都包含了上千臺(tái)機(jī)器，這些機(jī)器都分列在許多機(jī)架上，機(jī)架之間有著非常高的通信帶寬。集群之間是互聯(lián)的，但地理上是分布在不同地方的。應(yīng)用程序通常通過一個(gè)集群管理系統(tǒng)來執(zhí)行，該管理系統(tǒng)會(huì)通過調(diào)度作業(yè)來優(yōu)化集群資源的使用率，有時(shí)候會(huì)殺掉一些任務(wù)或?qū)⑷蝿?wù)遷移到其他機(jī)器上去。該系統(tǒng)中提供了一個(gè)名字服務(wù)系統(tǒng)，所以，各任務(wù)間可以通過與物理地址無關(guān)的邏輯名稱來各自標(biāo)識(shí)自己。持久化的數(shù)據(jù)被

46、在 GFS 或 Bigtable 中，而臨時(shí)文件比如緩存的消息則在本地磁盤中。9.5.1 Pregel 的執(zhí)行過程Pregel library 將一張圖劃分成許多的 partitions（如圖 9-8 所示），每一個(gè) partition 包含了一些頂點(diǎn)和以這些頂點(diǎn)為起點(diǎn)的邊。將一個(gè)頂點(diǎn)分配到某個(gè) partition 上去，取決于該頂點(diǎn)的ID，這意味著，即使在別的機(jī)器上，也是可以通過頂點(diǎn)的 ID 來知道該頂點(diǎn)是屬于哪個(gè)partition，即使該頂點(diǎn)已經(jīng)不存在了。默認(rèn)的 partition 函數(shù)為 hash(ID) mod N，N 為所有partition 總數(shù)，但是，用戶可以替換掉它。20 /

47、33圖 9-8 圖的劃分圖將一個(gè)頂點(diǎn)分配給哪個(gè) worker 機(jī)器，是整個(gè) Pregel 中對分布式不透明的主要地方。有些應(yīng)用程序使用默認(rèn)的分配策略就可以工作得很好，但是，有些應(yīng)用可以通過定義一些可以更好地利用圖本身的locality 的分配函數(shù)而從中獲益。比如，一種典型的可以用于 Web graph的啟發(fā)式方法是，將來自同一個(gè)站點(diǎn)的網(wǎng)頁數(shù)據(jù)分配到同一臺(tái)機(jī)器上進(jìn)行計(jì)算。在不考慮出錯(cuò)的情況下，一個(gè) Pregel 程序的執(zhí)行過程（如圖 9-9 和圖 9-10 所示）分為如下幾個(gè)步驟：1. 用戶程序的多個(gè) copy 開始在集群中的機(jī)器上執(zhí)行。其中，有一個(gè) copy 將會(huì)作為master，其他的作為

48、worker，master 不會(huì)被分配圖的任何一部分，而只是負(fù)責(zé)協(xié)調(diào) worker間的工作。worker 利用集群管理系統(tǒng)中提供的名字服務(wù)，來定位 master 位置，并發(fā)送信息給 master。2. Master 決定對這個(gè)圖需要多少個(gè) partition，并分配一個(gè)或多個(gè) partitions 到 worker 所在的機(jī)器上，如圖 9-8 所示。這個(gè)數(shù)字也可能由用戶進(jìn)行控制。一個(gè) worker 上有多個(gè) partition的情況下，可以提高 partitions 間的并行度，實(shí)現(xiàn)更好的負(fù)載平衡，通常都可以提高性能。每一個(gè) worker 負(fù)責(zé)在其之上的圖的那一部分的狀態(tài)(頂點(diǎn)及邊的增刪)，對

49、該部分中的頂點(diǎn)執(zhí)行 Compute()函數(shù)，并管理發(fā)送出去的以及接收到的消息。每一個(gè) worker 都知道該圖的計(jì)算在所有worker 中的分配情況。3. Master 進(jìn)程為每個(gè)worker 分配用戶輸入中的一部分，這些輸入被看作是一系列的集合，每一條都包含任意數(shù)目的頂點(diǎn)和邊。對輸入的劃分和對整個(gè)圖的劃分是正交的，通常都是基于文件邊界進(jìn)行劃分。如果一個(gè) worker 加載的頂點(diǎn)剛好是這個(gè)worker 所分配到21 / 33的那一部分，那么相應(yīng)的數(shù)據(jù)結(jié)構(gòu)就會(huì)被立即更新。否則，該 worker 就需要將它發(fā)送到它所應(yīng)屬于的那個(gè) worker 上。當(dāng)所有的輸入都被 load 完成后，所有的頂點(diǎn)將被

50、標(biāo)記為 active狀態(tài)。4. Master 給每個(gè) worker 發(fā)指令，讓其運(yùn)行一個(gè)超步，worker 輪詢在其之上的頂點(diǎn)，會(huì)為每個(gè) partition 啟動(dòng)一個(gè)線程。調(diào)用每個(gè) active 頂點(diǎn)的 Compute()函數(shù)，傳遞給它從上一次超步發(fā)送來的消息。消息是被異步發(fā)送的，這是為了使得計(jì)算和通信可以并行，以及進(jìn)行batching，但是，消息的發(fā)送會(huì)在本超步結(jié)束前完成。當(dāng)一個(gè) worker 完成了其所有的工作后，會(huì)通知 master，并告知當(dāng)前該 worker 上在下一個(gè)超步中將還有多少 active 節(jié)點(diǎn)。不斷重復(fù)該步驟，只要有頂點(diǎn)還處在 active 狀態(tài)，或者還有消息在傳輸。5.計(jì)

51、算結(jié)束后，master 會(huì)給所有的 worker 發(fā)指令，讓它保存它那一部分的計(jì)算結(jié)果。圖 9-9 Pregel 的執(zhí)行過程圖22 / 33圖 9-10 Worker 結(jié)構(gòu)圖9.5.2 容錯(cuò)性容錯(cuò)是通過 checkpoing 來實(shí)現(xiàn)的。在每個(gè)超步的開始階段，master 命令 worker 讓它保存它上面的 partitions 的狀態(tài)到持久設(shè)備，包括頂點(diǎn)值、邊值以及接收到的消息。Master自己也會(huì)保存 aggregator 的值。worker 的失效是通過 master 發(fā)給它的周期性的消息來檢測的。如果一個(gè) worker 在特定的時(shí)間間隔內(nèi)沒有收到消息，該 worker 進(jìn)程會(huì)終止。如果

52、 master 在一定時(shí)間內(nèi)沒有收到 worker 的反饋，就會(huì)將該 worker 進(jìn)程標(biāo)記為失敗。當(dāng)一個(gè)或多個(gè) worker 發(fā)生故障，被分配到這些 worker 的 partitions 的當(dāng)前狀態(tài)信息就丟失了。Master 重新分配圖的 partition 到當(dāng)前可用的 worker 集合上，所有的 partition 會(huì)從最近的某超步S 開始時(shí)寫出的checkpo中，重新加載狀態(tài)信息。該超步可能比在失敗的worker上最后運(yùn)行的超步 S要早好幾個(gè)階段，此時(shí)，失去的幾個(gè)超步將需要被重新執(zhí)行。Pregel對 checkpo頻率的選擇，是基于某個(gè)故障模型的平均時(shí)間的，從而平衡 checkpo

53、的開銷和恢復(fù)執(zhí)行的開銷。為了改進(jìn)恢復(fù)執(zhí)行的開銷和延遲， Confined recovery 已經(jīng)在開發(fā)中。它們會(huì)首先通過checkpo進(jìn)行恢復(fù)，然后，系統(tǒng)會(huì)通過回放來自正常的 partitions 的記入日志的消息以及恢23 / 33復(fù)過來的 partitions 重新生成的消息，更新狀態(tài)到 S階段。這種方式通過只對丟失的 partitions進(jìn)行重新計(jì)算，節(jié)省了在恢復(fù)時(shí)消耗的計(jì)算資源，同時(shí)，由于每個(gè) worker 只需要恢復(fù)很少的 partitions，減少了恢復(fù)時(shí)的延遲。對發(fā)送出去的消息進(jìn)行保存，會(huì)產(chǎn)生一定的開銷，但是，通常機(jī)器上的磁盤帶寬不會(huì)讓這種 IO 操作成為瓶頸。Confined

54、recovery 要求用戶算法是確定性的，以避免原始執(zhí)行過程中所保存下的消息與恢復(fù)時(shí)產(chǎn)生的新消息并存情況下帶來的不一致。隨機(jī)化算法可以通過基于超步和 partition 產(chǎn)生一個(gè)偽隨機(jī)數(shù)來使之確定化。非確定性算法需要關(guān)閉 Confined recovery 而使用老的恢復(fù)機(jī)制。9.5.3 Worker一個(gè) worker 機(jī)器會(huì)在內(nèi)存中分配到其之上的 graartition 的狀態(tài)。從概念上來講，可以簡單地看作是一個(gè)從頂點(diǎn) ID 到頂點(diǎn)狀態(tài)的 Map，其中，頂點(diǎn)狀態(tài)包括如下信息：該頂點(diǎn)的當(dāng)前值，一個(gè)以該頂點(diǎn)為起點(diǎn)的出射邊(包括目標(biāo)頂點(diǎn) ID，邊本身的值)列表，一個(gè)保存了接收到的消息的隊(duì)列，以及一

55、個(gè)當(dāng)前是否 active 的標(biāo)志位。該 worker 在每個(gè)超步中，會(huì)循環(huán)遍歷所有頂點(diǎn)，并調(diào)用每個(gè)頂點(diǎn)的 Compute()函數(shù)，傳給該函數(shù)頂點(diǎn)的當(dāng)前值，一個(gè)接收到的消息的迭代器和一個(gè)出射邊的迭代器。這里沒有對入射邊的，原因是每一條入射邊其實(shí)都是其源頂點(diǎn)的所有出射邊的一部分，通常在另外的機(jī)器上。出于性能的考慮，標(biāo)志頂點(diǎn)是否為 active 的標(biāo)志位，是和輸入消息隊(duì)列分開保存的。另外，只保存了一份頂點(diǎn)值和邊值，但有兩份頂點(diǎn) active flag 和輸入消息隊(duì)列存在，一份是用于當(dāng)前超步，另一個(gè)用于下一個(gè)超步。當(dāng)一個(gè) worker 在進(jìn)行超步 S 的頂點(diǎn)處理時(shí)，同時(shí)還會(huì)有另外一個(gè)線程負(fù)責(zé)接收來自同

56、一個(gè)超步的其他 worker 的消息。由于頂點(diǎn)當(dāng)前需要的是S-1 超步的消息，那么對超步 S 和超步 S+1 的消息就必須分開保存。類似地，頂點(diǎn) V 接收到了消息，表示 V 將會(huì)在下一個(gè)超步中處于 active，而不是當(dāng)前這一次。當(dāng) Compute()請求發(fā)送一個(gè)消息到其他頂點(diǎn)時(shí)，worker 首先確認(rèn)目標(biāo)頂點(diǎn)是屬于的worker 機(jī)器，還是當(dāng)前 worker。如果是在的 worker 機(jī)器上，那么消息就會(huì)被緩存，當(dāng)緩存大小達(dá)到一個(gè)閾值，最大的那些緩存數(shù)據(jù)將會(huì)被異步地 flush 出去，作為單獨(dú)的一個(gè)網(wǎng)絡(luò)消息傳輸?shù)侥繕?biāo) worker。如果是在當(dāng)前 worker，那么就可以做相應(yīng)的優(yōu)化：消息就會(huì)

57、直接被放到目標(biāo)頂點(diǎn)的輸入消息隊(duì)列中。如果用戶提供了 Combiner，那么在消息被加入到輸出隊(duì)列或者到達(dá)輸入隊(duì)列時(shí)，會(huì)執(zhí)24 / 33行 combiner 函數(shù)。后一種情況并不會(huì)節(jié)省網(wǎng)絡(luò)開銷，但是會(huì)節(jié)省用于消息的空間。9.5.4 MasterMaster 主要負(fù)責(zé) worker 之間的工作協(xié)調(diào)，每一個(gè) worker 在其到 master 的時(shí)候會(huì)被分配一個(gè)唯一的 ID。Master著一個(gè)當(dāng)前活動(dòng)的 worker 列表，該列表中就包括每個(gè)worker 的 ID 和地址信息，以及哪些 worker 被分配到了整個(gè)圖的哪一部分。Master 中保存這些信息的數(shù)據(jù)結(jié)構(gòu)大小，與 partitions 的

58、個(gè)數(shù)相關(guān)，與圖中的頂點(diǎn)和邊的數(shù)目無關(guān)。因此，雖然只有一臺(tái) master，也足夠用來協(xié)調(diào)對一個(gè)非常大的圖的計(jì)算工作。絕大部分的 master 的工作，包括輸入 、輸出、計(jì)算、保存以及從 checkpo中恢復(fù)，都將會(huì)在一個(gè)叫做 barriers 的地方終止。Master 在每一次操作時(shí)都會(huì)發(fā)送相同的指令到所有的活著的 worker，然后等待來自每個(gè) worker 的響應(yīng)。如果任何一個(gè) worker 失敗了，master便進(jìn)入恢復(fù)模式。如果 barrier 同步成功，master 便會(huì)進(jìn)入下一個(gè)處理階段，例如 master 增加超步的 index，并進(jìn)入下一個(gè)超步的執(zhí)行。Master 同時(shí)還保存著整

59、個(gè)計(jì)算過程以及整個(gè) graph 的狀態(tài)的統(tǒng)計(jì)數(shù)據(jù)，如圖的總大小，關(guān)于出度分布的柱狀圖，處于 active 狀態(tài)的頂點(diǎn)個(gè)數(shù)，在當(dāng)前超步的時(shí)間信息和消息流量，以及所有用戶自定義 aggregators 的值等。為方便用戶，Master 在運(yùn)行了一個(gè) HTTP服務(wù)器來顯示這些信息。9.5.5 Aggregators每個(gè)Aggregator 會(huì)通過對一組 value 值集合應(yīng)用 aggregation 函數(shù)計(jì)算出一個(gè)全局值。每一個(gè) worker 都保存了一個(gè) aggregators 的實(shí)例集，由 type name 和實(shí)例名稱來標(biāo)識(shí)。當(dāng)一個(gè)worker 對 graph 的某一個(gè) partition 執(zhí)

60、行一個(gè)超步時(shí)，worker 會(huì) combine 所有的提供給本地的那個(gè) aggregator 實(shí)例的值到一個(gè) local value：即利用一個(gè) aggregator 對當(dāng)前 partition 中包含的超步結(jié)束時(shí)，所有 workers 會(huì)將所有包含局部歸約值的所有頂點(diǎn)值進(jìn)行局部歸aggregators 的值進(jìn)行最后的匯總，并匯報(bào)給 master。這個(gè)過程是由所有 worker 構(gòu)造出一棵歸約樹而不是順序地通過流水線的方式來歸約，這樣做的原因是為了并行化歸約時(shí) CPU 的使用。在下一個(gè)超步開始時(shí)，master 就會(huì)將aggregators 的全局值發(fā)送給每一個(gè) worker。25 / 339.