【文章內(nèi)容簡介】 ster把一個(gè)圖分成多個(gè)分區(qū)，并把分區(qū)分配到多個(gè) Worker （ 3） Master會(huì)把用戶輸入劃分成多個(gè)部分，通常是基于文件邊界進(jìn)行劃分 （ 4） Master向每個(gè) Worker發(fā)送指令，Worker收到指令后，開始運(yùn)行一個(gè)超步 。當(dāng)完成以后， Worker會(huì)通知Master，并把自己在下一個(gè)超步還處于“活躍”狀態(tài)的頂點(diǎn)的數(shù)量報(bào)告給Master。上述步驟會(huì)被不斷重復(fù)，直到所有頂點(diǎn)都不再活躍并且系統(tǒng)中不會(huì)有任何消息在傳輸，這時(shí)，執(zhí)行過程才會(huì)結(jié)束 （ 5）計(jì)算過程結(jié)束后， Master會(huì)給所有的 Worker發(fā)送指令，通知每個(gè)Worker對自己的計(jì)算結(jié)果進(jìn)行持久化存儲(chǔ) 《 大數(shù)據(jù)技術(shù)原理與應(yīng)用 》 廈門大學(xué)計(jì)算機(jī)科學(xué)系 林子雨 容錯(cuò)性 ? Pregel采用檢查點(diǎn)機(jī)制來實(shí)現(xiàn)容錯(cuò)。在每個(gè)超步的開始， Master會(huì)通知所有的 Worker把自己管轄的分區(qū)的狀態(tài)（包括頂點(diǎn)值、邊值以及接收到的消息），寫入到持久化存儲(chǔ)設(shè)備 ? Master會(huì)周期性地向每個(gè) Worker發(fā)送 ping消息， Worker收到 ping消息后會(huì)給 Master發(fā)送反饋消息。如果 Master在指定時(shí)間間隔內(nèi)沒有收到某個(gè) Worker的反饋消息，就會(huì)把該 Worker標(biāo)記為“失效”。同樣地，如果一個(gè) Worker在指定的時(shí)間間隔內(nèi)沒有收到來自 Master的 ping消息，該 Worker也會(huì)停止工作 ? 每個(gè) Worker上都保存了一個(gè)或多個(gè)分區(qū)的狀態(tài)信息，當(dāng)一個(gè) Worker發(fā)生故障時(shí)，它所負(fù)責(zé)維護(hù)的分區(qū)的當(dāng)前狀態(tài)信息就會(huì)丟失。 Master監(jiān)測到一個(gè) Worker發(fā)生故障“失效”后，會(huì)把失效 Worker所分配到的分區(qū)，重新分配到其他處于正常工作狀態(tài)的 Worker集合上，然后，所有這些分區(qū)會(huì)從最近的某超步 S開始時(shí)寫出的檢查點(diǎn)中，重新加載狀態(tài)信息。很顯然，這個(gè)超步 S可能會(huì)比失效 Worker上最后運(yùn)行的超步 S1要早好幾個(gè)階段，因此，為了恢復(fù)到最新的正確狀態(tài)，需要重新執(zhí)行從超步 S到超步 S1的所有操作 《 大數(shù)據(jù)技術(shù)原理與應(yīng)用 》 廈門大學(xué)計(jì)算機(jī)科學(xué)系 林子雨 Worker 在一個(gè) Worker中，它所管轄的分區(qū)的狀態(tài)信息是保存在內(nèi)存中的。分區(qū)中的頂點(diǎn)的狀態(tài)信息包括： ?頂點(diǎn)的當(dāng)前值 ?以該頂點(diǎn)為起點(diǎn)的出射邊列表，每條出射邊包含了目標(biāo)頂點(diǎn) ID和邊的值 ?消息隊(duì)列，包含了所有接收到的、發(fā)送給該頂點(diǎn)的消息 ?標(biāo)志位，用來標(biāo)記頂點(diǎn)是否處于活躍狀態(tài) 在每個(gè)超步中， Worker會(huì)對自己所管轄的分區(qū)中的每個(gè)頂點(diǎn)進(jìn)行遍歷，并調(diào)用頂點(diǎn)上的 Compute()函數(shù)，在調(diào)用時(shí)，會(huì)把以下三個(gè)參數(shù)傳遞進(jìn)去： ?該頂點(diǎn)的當(dāng)前值 ?一個(gè)接收到的消息的迭代器 ?一個(gè)出射邊的迭代器 《 大數(shù)據(jù)技術(shù)原理與應(yīng)用 》 廈門大學(xué)計(jì)算機(jī)科學(xué)系 林子雨 Master ?Master主要負(fù)責(zé)協(xié)調(diào)各個(gè) Worker執(zhí)行任務(wù)，每個(gè) Worker會(huì)借助于名稱服務(wù)系統(tǒng)定位到 Master的位置，并向 Master發(fā)送自己的注冊信息，Master會(huì)為每個(gè) Worker分配一個(gè)唯一的 ID ?Master維護(hù)著關(guān)于當(dāng)前處于“有效”狀態(tài)的所有 Worker的各種信息，包括每個(gè) Worker的 ID和地址信息，以及每個(gè) Worker被分配到的分區(qū)信息 ?一個(gè)大規(guī)模圖計(jì)算任務(wù)會(huì)被 Master分解到多個(gè) Worker去執(zhí)行，如果參與任務(wù)執(zhí)行的多個(gè) Worker中的任意一個(gè)發(fā)生了故障失效， Master就會(huì)進(jìn)入恢復(fù)模式 ?Master在內(nèi)部運(yùn)行了一個(gè) HTTP服務(wù)器來顯示圖計(jì)算過程的各種信息，用戶可以通過網(wǎng)頁隨時(shí)監(jiān)控圖計(jì)算執(zhí)行過程各個(gè)細(xì)節(jié) 《 大數(shù)據(jù)技術(shù)原理與應(yīng)用 》 廈門大學(xué)計(jì)算機(jī)科學(xué)系 林子雨 Aggregator ? 每個(gè)用戶自定義的 Aggregator都會(huì)采用聚合函數(shù)對一個(gè)值集合進(jìn)行聚合計(jì)算得到一個(gè)全局值 ? 每個(gè) Worker都保存了一個(gè) Aggregator的實(shí)例集，其中的每個(gè)實(shí)例都是由類型名稱和實(shí)例名稱來標(biāo)識的 ? 在執(zhí)行圖計(jì)算過程的某個(gè)超步 S中，每個(gè) Worker會(huì)利用一個(gè)Aggregator對當(dāng)前本地分區(qū)中包含的所有頂點(diǎn)的值進(jìn)行歸約，得到一個(gè)本地的局部歸約值 ? 在超步 S結(jié)束時(shí)，所有 Worker會(huì)將所有包含局部歸約值的 Aggregator的值進(jìn)行最后的匯總，得到全局值，然后提交給 Master ? 在下一個(gè)超步 S+1開始時(shí)， Master就會(huì)將 Aggregator的全局值發(fā)送給每個(gè) Worker 《 大數(shù)據(jù)技術(shù)原理與應(yīng)用 》 廈門大學(xué)計(jì)算機(jī)科學(xué)系 林子雨 Pregel的應(yīng)用實(shí)例 ? 單源最短路徑 ? 二分匹配 《 大數(shù)據(jù)技術(shù)原理與應(yīng)用 》 廈門大學(xué)計(jì)算機(jī)科學(xué)系 林子雨 單源最短路徑 Pregel非常適合用來解決單源最短路徑問題，實(shí)現(xiàn)代碼如下： class ShortestPathVertex : public Vertexint, int, int { void Compute(MessageIterator* msgs) { int mindist = IsSource(vertex_id()) ? 0 : INF。 for (。 !msgsDone()。 msgsNext()) mindist = min(mindist, msgsValue())。 if (mindist GetValue()) { *MutableValue() = mindist。 OutEdgeIterator iter = GetOutEdgeIterator()。 for (。 !()。 ()) SendMessageTo((), mindist + ())。 } VoteToHalt()。 } }。 《 大數(shù)據(jù)技術(shù)原理與應(yīng)用 》 廈門大學(xué)計(jì)算機(jī)科學(xué)系 林子雨 二分匹配 程序的執(zhí)行過程是由四個(gè)階段組成的多個(gè)循環(huán)組成的，當(dāng)程序執(zhí)行到超步 S時(shí)， S mod 4就可以得到當(dāng)前超步處于循環(huán)的哪個(gè)階段。每個(gè)循環(huán)的四個(gè)階段如下： （ 1）階段 0：對于左集合中的任意頂點(diǎn) V，如果 V還沒有被匹配，就發(fā)送消息給它的每個(gè)鄰居頂點(diǎn)請求匹配，然后，頂點(diǎn) V會(huì)調(diào)用 VoteToHalt()進(jìn)入“非活躍”狀態(tài)。如果頂點(diǎn) V已經(jīng)找到了匹配，或者 V沒有找到匹配但是沒有出射邊，那么，頂點(diǎn) V就不會(huì)發(fā)送消息。當(dāng)頂點(diǎn) V沒有發(fā)送消息，或者頂點(diǎn) V發(fā)送了消息但是所有的消息接收者都已經(jīng)被匹配，那么，該頂點(diǎn)就不會(huì)再變?yōu)椤盎钴S（ active）”狀態(tài) （ 2）階段 1：對于右集合中的任意頂點(diǎn) U，如果它還沒有被匹配，則會(huì)隨機(jī)選擇它接收到的消息中的其中一個(gè)，并向左集合中的消息發(fā)送者發(fā)送消息表示接受該匹配請求，然后給左集合中的其他請求者發(fā)送拒絕消息；然后，頂點(diǎn) U會(huì)調(diào)用VoteToHalt()進(jìn)入“非活躍”狀態(tài) （ 3）階段 2：左集合中那些還未被匹配的頂點(diǎn)，會(huì)從它所收到的、右集合發(fā)送過來的接受請求中，選擇其中一個(gè)給予確認(rèn)，并發(fā)送一個(gè)確認(rèn)消息。對于左集合中已經(jīng)匹配的頂點(diǎn)而言，因?yàn)樗鼈冊陔A段 0不會(huì)向右集合發(fā)送任何匹配請求消息，因而也不會(huì)接收到任何來自右集合的匹配接受消息，因此，是不會(huì)執(zhí)行階段 2的 （ 4）階段 3：右集合中還未被匹配的