面向數據高交互任務的分布式圖計算方案的設計與實現

俞山青，王甬琪，崔文豪，孟櫟均，李 冰，傅晨波

1(浙江工業大學 信息工程學院，杭州 310000) 2(杭州中奧科技有限公司，杭州 310000)

1 引 言

許多領域中數據間的復雜關系，例如金融交易、社交網絡、電子商務等等，均可以使用圖這種數據結構進行抽象化簡，進而為相關問題的研究提供巨大的便利.例如Google通過網頁間超鏈接關系設計了PageRank[1]算法對網頁進行排序，優化搜索引擎；Facebook通過對用戶及其社交關系進行建模研究，提供更好的社區服務；Amazon通過用戶及其購物行為關系等信息構建推薦系統，引導用戶購物行為[2]等.在相關問題的研究過程中，圖計算往往是必不可少的，而對于其中較大規模圖數據的復雜計算任務，集中式計算方式難以達到預期效果，而現有的分布式計算方法又會由于過多的數據傳輸產生大量時耗，計算效率較低.

基于上述情況，本文提出了一種面向數據高交互任務的分布式圖計算方案.該方案中系統由任務管理中心、數據中心與計算節點組成.首先，任務管理中心分割計算任務并創建任務狀態表，將子任務分配給計算節點執行，同時將狀態表上傳至數據中心.其次，計算節點從數據中心獲取計算所需數據后執行計算任務，并將計算結果上傳至數據中心.在此過程中，任務管理中心通過任務狀態表定時檢測所有子任務是否已完成.最后，當檢測到所有任務完成后，任務管理中心將獲取所有子任務中間計算結果，將其進行整合匯總后上傳至數據中心存儲.在上述過程中，任務分割過程采用均勻分割策略，使每一子任務計算用時趨于一致以避免各計算節點負載不均衡；任務分發過程使用現有的任務分發框架實現，并由此實現任務的分布式計算；數據中心功能由數據庫實現，該方式可以在任務異常中斷后保留已完成的任務計算結果及任務狀態，且可以通過使用特定的文件存儲機制降低數據庫的讀表開銷，通過文件壓縮降低數據傳輸量；各計算節點執行任務時可獲取完整圖數據，從而避免了現有分布式圖計算方案中各節點間數據頻繁傳輸帶來的時耗.

此外，本文選取節點中心性作為計算目標，實現了上述分布式圖計算方案并進行相關實驗.節點中心性具體包括度中心性、接近中心性與介數中心性.選擇節點中心性作為計算目標的原因在于：1)節點中心性可以表示圖模型中各節點的影響力、控制力、流行性等，是一個重要的計算指標；2)該計算任務屬于我們的目標任務類型：中心性計算，尤其是介數中心性的計算，計算復雜度較高且需要使用到全圖數據.介數中心性計算的一般方法計算復雜度為O(n3)，其中n為節點數，當圖數據較大時，集中式計算方法難以實現高效運算.而該計算過程又需要頻繁使用全圖數據，現有分布式圖計算方案會產生大量的數據交互時耗.本文主要貢獻點如下：

1)提出了一種面向數據高交互任務的分布式圖計算方案，為此類圖計算任務提供一種可選的高效計算途徑；

2)以節點中心性為計算目標實現了本文所提出的分布式圖計算方案，并通過實驗驗證本文設計方案在目標任務類型上具有優良的計算性能.

2 相關工作

2.1 分布式圖計算系統

當前大規模分布式圖計算系統按照其計算模型主要可分為基于MapReduce模型的計算系統、基于BSP模型的計算系統、基于GAS模型的計算系統等幾大類[3].本小節將簡要介紹其中最具代表性的幾種計算模型.

MapReduce[4]是并行處理大規模數據的分布式計算平臺，最早由Google于2004年提出.該模型首先對數據進行切片，將計算任務分配到每一個切片上.切片后的處理過程分為Map與Reduce兩個階段.MapReduce模型的優勢在于使用簡單，只需定義Map與Reduce函數，且其對大數據的處理能力較強.但由于該模型切片一般較大，處理過程又分為兩個階段，對算法的迭代運算帶來較大困難，計算過程中頻繁讀寫磁盤數據又降低了計算效率.研究人員提出了一些基于MapReduce的優化模型，例如Haloop[5]、Twister[6]等，用于彌補其不足，但仍然難以勝任大規模圖計算任務.

BSP(Bulk Synchronous Parallel)[7]整體同步并行模型是一種適用于迭代計算的基于消息通信的并行計算模型，由Valiant于1990年提出.該模型將計算任務分解為一系列的迭代運算，每次迭代運算又分為本地計算、消息通信和路障同步三個階段.在BSP基礎上，Google于2010年提出“像節點一樣思考”的分布式圖計算模型Pregel[8].此類模型解決了MapReduce不善于做迭代計算的問題，使得計算效能得到提升.但由于存在著路障同步，各節點需要等待所有節點完成當前階段后才能進入下一階段.Giraph[9]，GPS[10]，Hama[11]等在Pregel的基礎上實現了一定的改進，用以提升計算性能.

GAS(Gather-Apply-Scatter)模型由GraphLab[12]項目組提出.該模型處理流程類似于BSP模型，又將節點更新函數細分為收集、應用和分散三個階段.GAS模型通過將節點更新函數細分化，使得計算節點間在運算時不需要時刻保持路障同步，大大減少了各節點的閑置時間，提升了系統的計算性能.但當各節點由于軟硬件原因等使得迭代輪數相差較大時，最終計算結果可能會不正確，需要引入額外的機制保證系統運算的準確性.在基于GAS模型的計算系統中，PowerGraph[13]首次使用節點分割策略替代邊分割策略，提升了系統的并發處理能力.PowerSwitch[14]、BiGraph[15]等均為在GraphLab與PowerGraph之上改進的計算模型，在不同層面上得到一定的提升.

除此之外，GraphX[16]是基于BSP與GAS的混合計算模型，可理解為Pregel與GraphLab的重寫與優化，它結合了兩者的優點，提升了計算性能.GraphX底層基于分布式計算引擎Spark，可部署于Hadoop分布式大數據處理平臺，為圖處理提供了巨大的便利.基于上述原因，本文選用GraphX作為實驗的參照對象.

2.2 中心性計算

本文選取了中心性指標中最具代表性的三種作為計算目標測試本文所提出的系統，分別為度中心性、接近中心性與介數中心性.本小節簡要介紹其概念及計算方式.

度中心性(Degree Centrality)是度量網絡中節點中心性最直接的指標，其值直接決定于與其相鄰的節點數.度中心性越大，表示相鄰節點越多，該節點越重要.其歸一化后計算公式為：

(1)

其中cD(i)為節點i的度，n為網絡的總節點數.

接近中心性(Closeness Centrality)是用于度量節點與其它節點接近程度的指標.接近中心性越大，表示該節點與其它節點的交互越迅速.其歸一化后計算公式為：

(2)

其中d(i,j)為節點i與節點j之間的最短距離，n為網絡的總節點數.

介數中心性(Betweenness Centrality)取決于一個節點擔任其它任意兩節點間最短路徑橋梁的次數.介數中心性越大，表示該節點在節點信息傳遞過程中的中介承載程度越高，起重要橋梁作用.其標準化后計算公式為：

(3)

其中V為所有節點集，s與t為節點集中與i相異的任意兩節點，σst為節點s與節點t之間的最短路徑的條數，σst(i)為節點i經過的節點s與節點t間最短路徑的條數，n為網絡的總節點數.

上述中心性計算公式可同時實現對有向圖、無向圖、有權圖與無權圖的中心性計算，而在各公式符號的計算中存在著一些差異：有向圖的度中心性需要依據邊的方向，分別基于出度與入度計算；有向圖接近中心性、介數中心性中最短距離、最短路徑的計算應考慮邊的方向，即路徑必須沿邊的正向；有權圖的中心性計算公式中的度、最短距離、最短路徑均應結合邊的權重進行計算等.

3 方案設計

本文提出的分布式圖計算方案整體設計如圖1所示.該設計分為三個部分，分別為任務管理中心、數據中心與計算節點，其中計算節點可按需求進行擴展.在執行計算任務時，首先由任務管理中心對任務進行分割，并同時創建任務狀態表傳輸至數據中心存儲.而后任務管理中心將分割后子任務分發給計算節點執行計算，更新狀態表中任務狀態為執行中.計算節點完成計算任務后將結果傳輸至數據中心存儲，更新狀態表中任務狀態為已完成.在上述過程中，任務管理中心通過任務狀態表定時檢測任務完成情況.當檢測到所有子任務均執行完成后，任務管理中心從數據中心獲取所有計算子任務的中間計算結果，并對其進行整合匯總后傳輸至數據中心存儲，完成本次計算任務.

圖1 方案整體設計Fig.1 Overall design of the scheme

具體地，在任務分割過程中，本文方案采用均勻分割策略.該策略可使得各計算節點執行子任務時用時趨于一致，從而實現負載均衡，避免任務將完成時有計算節點處于閑置狀態而降低計算效率.在任務分發過程中，本文方案使用現有的任務分發系統框架，由此實現任務的分布式計算.

方案中選用數據庫作為數據中心，其優點在于：將數據存儲于數據庫中可以有效避免任務非正常中斷時計算結果及任務狀態表的丟失；通過數據庫存儲的方式，極大地方便了各計算節點對數據的存取，也方便了計算結果的獲取與管理.考慮到使用數據庫存儲管理數據，會產生額外的讀寫開銷，本文方案對數據進行了整體存儲、整體讀寫的設計，相較于數據庫中數據記錄的逐條讀寫方式，具有更低的讀寫開銷；同時通過對整體存儲的數據進行壓縮，降低了數據的傳輸量.

計算節點在執行計算子任務時，從數據中心獲取全圖數據.采用數據分布式存儲的分布式圖計算方法在計算數據高交互計算任務時，會在各計算節點間產生大量的數據傳輸時耗，本文方案采取計算節點保留整體圖數據的設計，提升了計算性能.

4 具體實現

本節將基于節點中心性計算任務，具體實現本文所設計的方案，并將其用于測試本方案的有效性.

4.1 任務管理中心具體實現

本文選取任務分發框架Gearman(1)https：//gearman.org/實現任務的分發功能，其原因在于：1)Gearman可以有效實現任務的分布式并行計算；2)Gearman具有一定負載均衡的能力；3)Gearman可以實現多種程序語言之間的溝通，方便程序的實現.

任務管理中心主要負責任務的分割、分發、定時檢測及最終計算結果的匯總，其具體流程如圖2所示.值得注意的是，在任務分割的過程中，由于不同節點的中心性計算時耗不一且難以估計，具體實現過程中基于節點數進行均勻分割，使得各計算子任務時耗盡可能趨于一致，避免負載不均衡造成時間損耗.任務分割完成后創建任務狀態表上傳至數據中心，用于后續的定時檢測功能的實現.任務分發過程通過Gearman實現.定時檢測過程中，查看任務狀態表中任務狀態是否全部為已完成，若均已完成則進入最終計算結果的匯總過程.由于三種中心性中，度中心性及接近中心性計算結果已為最終計算結果，不需要進行整合匯總，該過程僅對介數中心性實行，即從數據中心中下載中間計算結果至任務管理中心實現整合匯總后，重新上傳數據中心存儲.

圖2 任務管理中心流程圖Fig.2 Flow chart on tasks management center

此外，由于度中心性的計算較為簡單，對其使用分布式計算方案會由于數據傳輸的時耗而降低整體計算效率.本文直接在任務管理中心中完成其計算，即任務中心也可直接完成簡單計算任務以提升整體計算效率.

4.2 數據中心具體實現

本文選取MongoDB(2)https://www.mongodb.com/數據庫作為數據中心，其原因在于：

1)MongoDB數據庫為眾多程序語言提供了接口，使用方便；2)MongoDB數據庫中可使用GirdFS文件存儲機制，可以實現數據以文檔形式進行整體傳輸，降低數據庫的讀寫開銷.

數據中心主要負責數據的存儲與傳輸.在數據存儲功能中，MongoDB數據庫主要實現了圖數據、任務狀態表、子任務計算結果及最終任務結果的存儲.其中圖數據用于各計算節點執行任務的過程；任務狀態表用于任務管理中心的定時檢測；子任務計算結果來自于各計算節點，并用于計算結果的整合匯總，即最終任務結果.具體地，數據庫中設計了四張數據表，分別為任務狀態表、度中心性表、接近中心性表與介數中心性表.其中任務狀態表中具體字段及相關信息如表1所示；而度中心性表、接近中心性表與介數中心性表由于字段及相關信息基本一致，其主要設置可同時參照表2.此外，針對有向圖的中心性計算，在度中心性表中存在出度、入度兩個字段未在表中給出.

表1 數據中心任務狀態表Table 1 Task status table in data center

表2 數據中心中心性表Table 2 Centrality tables in data cente

在數據傳輸功能上，方案實例使用GirdFS文件存儲機制：相較于對數據庫中的記錄進行逐條讀寫，基于文件存儲機制的數據整體傳輸方式可以在一定程度上減少讀寫開銷.此外，使用Google開發的壓縮庫Snappy(3)https：//github.com/google/snappy對文件進行壓縮，以減少數據傳輸量：Snappy壓縮庫特點在于能提供快速解壓縮的同時具有較低的壓縮率，性能均衡.官方文檔顯示，該壓縮庫在64位Core i7單核模式下，具有超過250MB/s的壓縮速率與超過500MB/s的解壓縮速率.同時，使用Snappy對實驗中原數據大小為91.3MB的web-Google數據集進行壓縮后文件大小降為35.5MB.本文通過上述設計優化數據傳輸過程.

4.3 計算節點具體實現

本文方案中計算節點主要負責計算子任務的執行及結果的上傳.具體地，計算節點連接到任務分發框架Gearman后，從中獲取計算任務及MongoDB數據庫端口，并由此獲取計算數據.計算節點執行接近中心性及介數中心性分量的計算程序.待執行完成后，將計算結果上傳至數據庫，等待最終的結果匯總.其具體流程如圖3所示.

圖3 計算節點執行流程圖Fig.3 Flow chart on computing nodes

5 實驗結果

5.1 實驗環境

本文方案實例部署在由5臺相同配置服務器組成的內網集群，其CPU型號為Intel(R)Xeon(R)E5-2650 v4，主頻2.20GHz，核數8個，內存8G.

實驗中五臺服務器均作為計算節點.在此基礎上，選取其中兩臺服務器分別部署任務管理中心與數據中心.同一臺服務器同時作為計算節點與任務管理中心或數據中心可能會影響程序的運行性能，但這種方式計算性能高于僅將3臺服務器作為計算節點.

此外，實驗選取幾種單機下集中式圖計算工具與分布式圖計算工具GraphX作為本文設計系統的參照對象.實驗環境分別為單臺服務器與五臺服務器組成的Spark內網集群.

5.2 數據集

實驗中使用的數據集來自美國斯坦福大學和德國科布倫茨-蘭道大學的開源網絡數據集網站.數據集的說明如表3所示.

表3的數據集源數據存儲格式不一，實驗前需進行預處理，使其具有相同格式.此外，實驗中通過兩條有向邊來表示無向邊；通過對無權圖數據集上對每條邊做1-10整數的隨機賦權，作為有權圖進行測試實驗，并在相應文件名后添加“-w”區分于無權圖數據集.

表3 實驗數據集說明Table 3 Description on experimental data sets

5.3 參照方案

在本文實驗中，選擇幾種常用的集中式圖計算工具igraph(4)https：//igraph.org/、Gephi(5)https：//gephi.org/、NetworkX(6)http：//networkx.github.io/及Spark平臺下的分布式圖計算工具GraphX作為參照對象.各方案所使用的程序語言及介數中心性計算的近似算法復雜度等相關信息如表4所示.其中m為邊數，n為節點數.

表4 各方案相關信息Table 4 Related information of each scheme

表4中除NetworkX外的參照方案的程序語言效率及算法復雜度均與本方案相近或是優于本方案.其中NetworkX方案主要用于保證本文方案計算結果與直接通過定義計算的方案結果一致.GraphX方案采用基于DFS加速的分布式介數中心性算法[17]實現優化.

5.4 實驗相關信息與結果分析

本小節首先介紹具體實驗過程中計算任務分割及任務分發情況，其次通過計算本文方案計算結果與集中式工具結果之間的相關系數及各自L2范數，驗證本方案及方案實例計算結果的正確性，最后通過比較各種方案對相同計算任務的計算用時，驗證本文方案對此類任務具有優良的計算效率.

在實驗過程中，任務分割采用基于節點數的均勻分割策略，分割后任務由Gearman任務分發系統實現分配.在一次Epinions數據集的中心性計算實驗中，分割后的子任務及分配情況如表5所示.任務數依據計算節點的整數倍被設置為10，每個任務所需計算的節點數均勻分配，計算節點完成當前任務的計算后發送請求并開始下一個任務的計算.

表5 任務分割及分發實例Table 5 Example of task split and tasks distribute

在計算結果正確性的驗證實驗中，本文以email-Enron數據集的介數中心性為例，計算本文提出方案與常用集中式計算工具計算結果之間的皮爾森(Person)、斯皮爾曼(Spearman)和肯德爾(Kendall)相關系數，用以說明兩類方法變化趨勢一致性，實驗中數據集又分為無權圖數據與有權圖數據.此外，還對各種工具及分布式計算方案的實驗結果計算各自的L2范數，用以確保計算結果數值一致性.實驗結果如表6-表8所示.其中表6為本方案與集中式圖計算方案在無權圖上計算結果的相關系數，表7為兩者在有權圖上的相關系數，表8為各方法計算結果的L2范數.表中“/”表示不支持計算.

表6 本文方案與各方案在無權圖計算中的相關系數Table 6 Correlation coefficient on unweighted graphcalculation between our scheme and otrher schemes

表7 本文方案與各方案在有權圖計算中的相關系數Table 7 Correlation coefficient on weighted graphcalculation between our scheme and otrher schemes

表8 各方案計算結果L2范數Table 8 L2 norm on calculation results of each scheme

由表中數據可以得知，本文方案的運算結果與常用集中式計算工具的運算結果之間相關系數約等于1，即兩者的變化趨勢基本一致.此外，各方案計算結果的L2范數基本一致，說明了本文方案計算數值能與常用集中式計算工具保持一致.本方案的計算正確性基本可以保證.在實際結果中，本文方案計算結果在小數點后十一位均可與常用集中式計算工具保持一致.

上述實驗結果已驗證本文提出方案計算結果的正確性，在此基礎上進行各方案計算效率的比較實驗.實驗中使用選取的3種集中式計算工具、分布式圖計算系統Graph X與本文設計方案分別在7個數據集上計算3種節點中心性，數據集又各自分為有權圖數據與無權圖數據.實驗結果各方案時耗如表9與表10所示.Gephi工具不支持有權圖的計算，故表10中不含該列.表中“-”表示計算用時超過12小時.

表9 無權圖中各方案計算時耗Table 9 Time consumption on unweighted graph computation of each scheme

表10 有權圖中各方案計算時耗Table 10 Time consumption on weighted graph computation of each scheme

由表9和表10中各方案計算時耗數據中可以得知，本文提出的方案計算用時遠小于常用的集中式圖計算工具，而基于GraphX的分布式圖計算方案計算用時遠大于常用的集中式圖計算工具.本文方案相較于集中式圖計算工具與分布式圖計算系統GraphX，都具有更高的計算效率.

兩種分布式方案在計算效率上存在巨大差異，其原因在于它們并行計算的原理不同.基于GraphX的分布式計算方法對圖數據進行了圖分割，數據分布在各個計算節點上，當計算介數中心性這類需要全圖節點信息的目標時，需要在計算節點間頻繁傳遞數據，從而產生大量額外開銷，降低計算效率.該分布式方法適用于各計算節點數據間依賴程度不高的任務.而本文方案不對圖數據進行分割，而是通過直接對任務進行分割實現分布式計算，這種方式大大減少了數據交互時耗，適用于數據高交互的圖計算任務.

6 結 語

本文設計并實現了一種面向數據高交互任務的分布式圖計算方案.該方案由任務管理中心、數據中心及計算節點組成.其中心性計算實例驗證了本方案計算結果的正確性，并且在目標任務上具有優良性能，為此類計算任務提供了一種可選的計算方案.