許杰：騰訊Angel Graph大規(guī)模圖計(jì)算平臺(tái)｜DataFunTalk

導(dǎo)讀：本次分享主要從四個(gè)方面來(lái)介紹：

• Angel Graph起源和發(fā)展
• Angel Graph框架
• 通信和計(jì)算優(yōu)化
• Angel Graph應(yīng)用

01

Angel Graph起源和發(fā)展

首先，簡(jiǎn)單介紹一下Angel Graph的起源和發(fā)展，方便大家理解我們的演進(jìn)方向。

Angel在2017年開源了1.0版本，當(dāng)時(shí)它還只是一個(gè)支持高維稀疏大模型的平臺(tái)，只包含LR，LDA，GBDT這樣一些機(jī)器學(xué)習(xí)算法。到18年發(fā)布了1.5版本，有兩個(gè)非常重要的feature奠定了系統(tǒng)的高性能和易用性。一是支持了自定義的PS function，它使得PS可以承擔(dān)部分計(jì)算功能；二是引入了spark生態(tài)，整個(gè)平臺(tái)的應(yīng)用性得到了很大的提升。

從18年下旬開始，業(yè)界和學(xué)術(shù)界圖數(shù)據(jù)增多，挖掘價(jià)值很大，但當(dāng)時(shí)業(yè)界的一些圖平臺(tái)難以做到高性能、高可用和應(yīng)用性的統(tǒng)一。GraphX作為流行的圖計(jì)算框架不能很好的支持百億級(jí)規(guī)模，因此，很有必要基于自己的平臺(tái)打造一個(gè)高性能、高可用的圖平臺(tái)。因?yàn)閳D算法本質(zhì)是一個(gè)高維稀疏的模型，與Angel PS的能力完美契合，所以我們就基于Angel打造了一個(gè)圖平臺(tái)。v3.0，v3.1和v3.2這些版本側(cè)重點(diǎn)集中在圖算法的完善和優(yōu)化上。3.2版本算是一個(gè)比較完備的大規(guī)模圖計(jì)算平臺(tái)。這就是我們的框架從機(jī)器學(xué)習(xí)到圖學(xué)習(xí)的平臺(tái)演進(jìn)。

02

Angel Graph框架

接下來(lái)給大家詳細(xì)介紹Angel Graph框架。

1. Spark on Angel

Angel Graph可以分為幾個(gè)要素來(lái)理解：第一個(gè)就是Angel PS，從這個(gè)角度可以將Angel視作一個(gè)高性能的參數(shù)服務(wù)器。第二個(gè)要素就是Spark，Spark是大數(shù)據(jù)生態(tài)中一個(gè)很完善高效的ETL處理平臺(tái)。

為什么要將angel和spark結(jié)合起來(lái)做圖計(jì)算，主要基于兩點(diǎn)考慮。

• 其一，算法和模型的本質(zhì)是高維稀疏的。高維體現(xiàn)在工業(yè)界中的圖數(shù)據(jù)輕而易舉就達(dá)到十億、百億，甚至千億級(jí)別。稀疏性主要體現(xiàn)在節(jié)點(diǎn)之間的連接本身就是非常稀疏的。類似PageRank，k-core等圖算法，在迭代過(guò)程中會(huì)有一些節(jié)點(diǎn)慢慢變得不活躍?；钴S的節(jié)點(diǎn)越來(lái)越少，所以計(jì)算過(guò)程也是偏稀疏化的。高維、稀疏這種特征與Angel PS完美契合，因此可以高效的地支持大圖、超大圖的規(guī)模。
• 其二，圖計(jì)算任務(wù)本身需要一個(gè)復(fù)雜的數(shù)據(jù)預(yù)處理過(guò)程，比如數(shù)據(jù)切分，生成表示做采樣，或者統(tǒng)計(jì)圖上的一些節(jié)點(diǎn)信息如出度入度之類。引入Spark生態(tài)，可以很好的實(shí)現(xiàn)端到端的數(shù)據(jù)處理。數(shù)據(jù)從HDFS中讀入，預(yù)處理后直接給圖算子計(jì)算，最后將結(jié)果存儲(chǔ)在HDFS中，實(shí)現(xiàn)端到端的過(guò)程。

Sparkon Angel是一個(gè)基于PS結(jié)構(gòu)的高性能圖計(jì)算引擎。其架構(gòu)如上圖所示，上面是angel，下面是spark。Spark負(fù)責(zé)分布式計(jì)算，Angel負(fù)責(zé)存儲(chǔ)圖模型，主要做內(nèi)容的隨機(jī)讀取和訪問(wèn)。

下面以一個(gè)求共同鄰居的例子詳細(xì)介紹圖計(jì)算任務(wù)在Spark on Angel框架上的實(shí)現(xiàn)。假設(shè)在如上圖所示的graph上，需要計(jì)算節(jié)點(diǎn)5和節(jié)點(diǎn)7的共同鄰居。一個(gè)簡(jiǎn)單的計(jì)算邏輯是先拿到節(jié)點(diǎn)5的鄰居{1，2，6，7}和節(jié)點(diǎn)7的鄰居{3，4，5，6}，把這兩個(gè)序列做交集得到節(jié)點(diǎn)5和7的共同鄰居6。

在Spark on Angel架構(gòu)上，Angel PS用于管理頻繁更新和隨機(jī)訪問(wèn)的數(shù)據(jù)，這里主要指節(jié)點(diǎn)的鄰居，也就是我們常說(shuō)的鄰接表。Spark從HDFS中讀原始的輸入邊，形成一個(gè)edge RDD。在具體計(jì)算時(shí)，每一個(gè)executor對(duì)每一條邊從PS中拉取原節(jié)點(diǎn)和目標(biāo)節(jié)點(diǎn)的鄰居，之后做交集。對(duì)于某一些算法，可能需要迭代更新PS上的模型，此時(shí)需要用一個(gè)push接口將需要更新的數(shù)據(jù)push到PS上。整個(gè)計(jì)算過(guò)程邏輯上非常簡(jiǎn)單。在實(shí)現(xiàn)上，這里簡(jiǎn)單列舉了比較粗略的代碼?？梢园l(fā)現(xiàn)主要都是Spark的接口，中間會(huì)穿插一些對(duì)PS的push接口，具體主要取決于用戶如何處理規(guī)劃自己的數(shù)據(jù)。

2. PyTorch on Angel

Sparkon angel主要是針對(duì)比較簡(jiǎn)單的圖挖掘算法，但是對(duì)于圖神經(jīng)網(wǎng)絡(luò)，它的能力是不夠的。主要的原因是：

• 圖神經(jīng)網(wǎng)絡(luò)需要自動(dòng)求導(dǎo)工具，但是Scala的數(shù)據(jù)結(jié)構(gòu)和運(yùn)算是沒辦法實(shí)現(xiàn)的，angel自身的計(jì)算圖框架也無(wú)法滿足。
• 之前的模式是CPU共享集群，當(dāng)引入PyTorch后可以方便的擴(kuò)展到GPU的計(jì)算上。
• 引入PyTorch是為了擁抱深度學(xué)習(xí)的業(yè)界生態(tài)。由于學(xué)術(shù)界和業(yè)界基于PyTorch和TensorFlow的模型非常多，不需要重復(fù)造輪子。引入PyTorch就可以直接復(fù)用這些模型了。

基于以上三點(diǎn)，pytorch被引入進(jìn)來(lái)。對(duì)計(jì)算邏輯相對(duì)簡(jiǎn)單的圖挖掘算法，計(jì)算部分可以用一些簡(jiǎn)單的scala運(yùn)算符實(shí)現(xiàn)，而在PyTorch on Angel框架下，Torch主要作為一個(gè)單機(jī)的runtime核心，原來(lái)的計(jì)算核心被替換成一個(gè)單機(jī)的PyTorch model，除此之外，外層的分布式框架幾乎是一樣的。PS負(fù)責(zé)存儲(chǔ)模型的參數(shù)，單機(jī)的PyTorch model負(fù)責(zé)接收數(shù)據(jù)，然后做前向計(jì)算和反向的梯度計(jì)算，梯度被傳遞給PS做更新。對(duì)于用戶來(lái)說(shuō)，不需要關(guān)注外面分布式的殼子是怎么實(shí)現(xiàn)的，只需要去定義自己的PyTorch模型就可以了。

3. 圖數(shù)據(jù)切分

所有的分布式系統(tǒng)都會(huì)涉及到數(shù)據(jù)切分的問(wèn)題。我們這里有兩個(gè)分區(qū)，一個(gè)是Spark端的data分區(qū)，還有一個(gè)就是AngelPS端的模型分區(qū)。對(duì)于Spark端的圖數(shù)據(jù)切分，本平臺(tái)支持了多種切分方式，包括邊切、點(diǎn)切、混切，還有塊切。其它比較復(fù)雜的情況，需要去做預(yù)處理。其中點(diǎn)切、邊切的方式都是非常簡(jiǎn)單的。比如，直接從HDFS讀邊，天然就是一個(gè)點(diǎn)切的方式。如果是對(duì)邊的RDD做group by key操作，形成的就是一個(gè)邊切的模式，操作起來(lái)非常簡(jiǎn)單。用戶可以根據(jù)自己數(shù)據(jù)分布特點(diǎn)和計(jì)算邏輯，去選擇需要的切分方式。

4. 模型切分

對(duì)于PS上面的模型切分，我們同時(shí)支持Range和Hash兩種切分方式。

• Range切分。假設(shè)現(xiàn)有0到20這些節(jié)點(diǎn)，Range切分是按照它的最小ID和最大ID做一個(gè)均勻的切開，這樣節(jié)點(diǎn)ID分布在連續(xù)的編碼空間內(nèi)，內(nèi)存占用也比較少。但在節(jié)點(diǎn)ID不是連續(xù)分布的情況下，容易造成負(fù)載均衡的問(wèn)題。
• Hash分區(qū)。Hash切分通過(guò)將ID打散，盡量的分布到不同的PS分區(qū)內(nèi)，很好的緩解了負(fù)載不均衡的問(wèn)題。而且Hash可以支持任意的節(jié)點(diǎn)類型，不需要提前將其他數(shù)據(jù)形式的節(jié)點(diǎn)ID編碼成Long類型。其缺點(diǎn)是內(nèi)存占用比較多，在實(shí)際使用的過(guò)程中，還是要根據(jù)具體的數(shù)據(jù)分布去選擇合適的切分方式。

5. 穩(wěn)定性

在共享集群的環(huán)境中單機(jī)發(fā)生宕機(jī)的概率是非常高的，而且圖計(jì)算對(duì)錯(cuò)誤的容忍度非常低，對(duì)于圖挖掘算法，如果PS宕機(jī)，那么所獲取到的消息是空的，此時(shí)整個(gè)計(jì)算都會(huì)發(fā)生錯(cuò)誤，所以必須要考慮整個(gè)平臺(tái)的穩(wěn)定性。要保證魯棒性的前提下，再去兼顧高性能和應(yīng)用性。

系統(tǒng)層容災(zāi)：類似RDDcache，Angel PS端提供了checkpoint功能。用戶可以選擇讓每個(gè)PS每隔一段時(shí)間checkpoint一下數(shù)據(jù)。當(dāng)PS發(fā)生宕機(jī)時(shí)，重啟的PS可以去檢查是否有之前PS的備份，若有則直接加載進(jìn)來(lái)，繼續(xù)運(yùn)算。

算法層容災(zāi)：根據(jù)算法的錯(cuò)誤容忍度選擇合適的checkpoint策略。① 圖挖掘算法不能容忍消息的丟失，所以ps的容錯(cuò)策略也非常嚴(yán)格，需要同步做全局的checkpoint，可以保證全局正確性，但耗時(shí)較長(zhǎng)；② 圖表示學(xué)習(xí)類似于機(jī)器學(xué)習(xí)，可以容忍部分模型權(quán)重的丟失，因此不需要全局同步checkpoint，可以設(shè)置每個(gè)PS隔一段時(shí)間自動(dòng)單獨(dú)去checkpoint；③ 圖神經(jīng)網(wǎng)絡(luò)算法在ps上既存有圖模型權(quán)重又存有圖結(jié)構(gòu)和屬性，同時(shí)兼具前兩種類型的特點(diǎn)，因此建議單點(diǎn)和全局checkpoint結(jié)合起來(lái)使用，對(duì)不同數(shù)據(jù)類型采取不同容災(zāi)策略。

6. 易用性

Angel graph的易用性體現(xiàn)在融合了大數(shù)據(jù)生態(tài)，是一個(gè)非常容易上手的平臺(tái)。對(duì)于直接使用的情況，系統(tǒng)提供了非常豐富的圖算法庫(kù)，不僅支持了圖挖掘、圖表示還有圖神經(jīng)網(wǎng)絡(luò)；對(duì)于需要二次開發(fā)的用戶，系統(tǒng)抽象了許多通用的圖算子，涵蓋了從數(shù)據(jù)的加載，到預(yù)處理，再到一些常用的圖計(jì)算算子的抽象。

03

通信和計(jì)算優(yōu)化

因?yàn)樵诜植际较到y(tǒng)上主要會(huì)涉及到通信和計(jì)算上面的開銷，所以著重從這兩個(gè)方面來(lái)介紹。

1. 通信優(yōu)化

Angle是一個(gè)PS架構(gòu)，計(jì)算和存儲(chǔ)是分離的，因此在通信上會(huì)有兩個(gè)問(wèn)題。

• 連接數(shù)過(guò)多。在極端的情況下，一個(gè)worker可能要與所有的PS去做通信。
• 單次通信的量級(jí)較大。每次push或者要pull很大的數(shù)據(jù)，會(huì)導(dǎo)致性能問(wèn)題。

(1) 通信鏈路過(guò)多

通信鏈路過(guò)多主要是會(huì)有一個(gè)短板效應(yīng)。一次pull或push的時(shí)間是由最長(zhǎng)的RPC決定的。因此可以進(jìn)行RPC的剪枝，讓一個(gè)計(jì)算節(jié)點(diǎn)只與少數(shù)幾個(gè)PS去進(jìn)行通信。

具體做法分為兩種情況：

第一種情況，當(dāng)只有源節(jié)點(diǎn)（或者key值）需要跟PS通信的時(shí)候，完全可以把data分區(qū)按照PS分區(qū)的方式進(jìn)行分區(qū)，這樣完全可以做到executor和PS一對(duì)一的關(guān)系；

另一種情況，當(dāng)源節(jié)點(diǎn)和目標(biāo)節(jié)點(diǎn)同時(shí)需要去跟PS進(jìn)行通信時(shí)，我們基于這種情況針對(duì)性地提出了Square Partitioner分區(qū)方式，其原理如上圖所示。假如PS有8個(gè)分區(qū)，data分區(qū)就有64個(gè)。每個(gè)data分區(qū)內(nèi)獲得的邊都是根據(jù)PS分區(qū)來(lái)決定的，如圖中第0個(gè)data分區(qū)中的所有邊的源節(jié)點(diǎn)和目標(biāo)節(jié)點(diǎn)都屬于ps的第0個(gè)分區(qū)，而第1個(gè)data分區(qū)中的所有邊的源節(jié)點(diǎn)和目標(biāo)節(jié)點(diǎn)分別屬于ps的第0個(gè)和第1個(gè)分區(qū)，依次類推，可以判定每個(gè)data分區(qū)最多只需要與兩個(gè)ps分區(qū)進(jìn)行交互。這種方法也會(huì)存在當(dāng)PS分區(qū)值較大時(shí)導(dǎo)致造成data分區(qū)數(shù)暴增的問(wèn)題。此時(shí)我們引入kernel的概念，如當(dāng)kernel等于2時(shí)，可以讓2*2=4個(gè)小分區(qū)合并為一個(gè)大的data分區(qū)，這樣每個(gè)data分區(qū)最多只與kernel * kernel個(gè)PS分區(qū)進(jìn)行通信，但data分區(qū)的量級(jí)迅速減少。這種分區(qū)方式在node2vec算法的測(cè)試可以讓通信時(shí)間減少一半以上。

(2) 通信量較大

另外一個(gè)問(wèn)題是通信的量比較大，通訊的數(shù)據(jù)比較多。Angel PS支持自定義PS function功能。Angel PS與傳統(tǒng)PS的區(qū)別在于它可以做計(jì)算，計(jì)算的邏輯是由用戶自定義的。對(duì)于經(jīng)常要做鄰居節(jié)點(diǎn)采樣的情況，傳統(tǒng)PS計(jì)算與存儲(chǔ)完全分離，在對(duì)一個(gè)節(jié)點(diǎn)做采樣時(shí)，例如上圖左邊的情況，worker端從PS把全部鄰居拉下來(lái)(拉的是1234所有的鄰居)，然后在下邊做采樣(得到2)。但是如果PS可以承擔(dān)部分計(jì)算，把采樣的功能放在PS上面去做，這樣拉的數(shù)據(jù)就只有2了。整體來(lái)看，把通訊的數(shù)據(jù)大小從邊的量級(jí)降到了點(diǎn)的量級(jí)。

2. 計(jì)算優(yōu)化

(1) 超級(jí)頂點(diǎn)打散計(jì)算

前面的是通信上的優(yōu)化，計(jì)算也是較為關(guān)注的一個(gè)點(diǎn)。做圖計(jì)算，超級(jí)節(jié)點(diǎn)是不可避免的。超級(jí)節(jié)點(diǎn)會(huì)引起負(fù)載不均衡的問(wèn)題，通信會(huì)有，計(jì)算也會(huì)有。在通訊中，負(fù)載均衡和一些通訊的策略都可以去緩解。而在計(jì)算上有很多常用的trick。

舉例來(lái)說(shuō)，要計(jì)算任意兩個(gè)商品的相似度，需要找到同時(shí)購(gòu)買了這兩個(gè)商品的所有用戶。比如說(shuō)100萬(wàn)個(gè)用戶同時(shí)購(gòu)買了A，B兩種商品。這100萬(wàn)個(gè)用戶各自有一條購(gòu)買序列，計(jì)算A、B兩種商品的相似度時(shí)有個(gè)操作是要去計(jì)算這100萬(wàn)個(gè)序列的兩兩交叉項(xiàng)，就形成了一個(gè)類似笛卡爾積的運(yùn)算。100萬(wàn)×100萬(wàn)就達(dá)到了一萬(wàn)億的維度，如果把這種計(jì)算放在一個(gè)worker上面是完全不可接受的。打散計(jì)算的策略是將把這些超級(jí)商品對(duì)單獨(dú)提出來(lái)，將單個(gè)笛卡爾積的多次交叉項(xiàng)計(jì)算打散到各個(gè)worker上去。

總而言之，原來(lái)計(jì)算的力度是單個(gè)的商品對(duì)計(jì)算，現(xiàn)在是把計(jì)算力度打散，降低到單個(gè)交叉項(xiàng)上計(jì)算，這樣就可以將已經(jīng)空閑的資源全部利用起來(lái)，時(shí)間上也得到了非常顯著的提升。

(2) 動(dòng)態(tài)壓縮鄰接表

另一個(gè)計(jì)算優(yōu)化是動(dòng)態(tài)壓縮鄰接表，多適用于千億邊的大圖。因?yàn)楹芏鄨D算法都是需要先生成節(jié)點(diǎn)的鄰接表，然后推送到PS上面去做一些采樣或聚合操作。一般我們從HDFS讀進(jìn)來(lái)的直接是一個(gè)邊表，要得到鄰接表，就可以簡(jiǎn)單做一個(gè)RDD的groupByKey操作。對(duì)于這種千億邊的大圖，這種操作會(huì)產(chǎn)生非常大的shuffle量。如果集群條件不是特別穩(wěn)定的話，很容易在這一步直接宕機(jī)。

動(dòng)態(tài)壓縮鄰接表讀進(jìn)來(lái)的也是邊表的結(jié)構(gòu)，但與之前不同的是不需要用groupByKey事先生成鄰接表。當(dāng)向PS push鄰居的時(shí)候，只是push當(dāng)前分區(qū)里能拿到的所有鄰居。在Angel PS端會(huì)自動(dòng)去做動(dòng)態(tài)接收和拼接鄰居的操作，并且當(dāng)鄰居個(gè)數(shù)達(dá)到一定的閾值之后會(huì)做壓縮，整個(gè)過(guò)程是沒有RDD shuffle的，對(duì)于超級(jí)節(jié)點(diǎn)也非常友好，push過(guò)程中可以按照mini-batch向PS做鄰居推送，在資源受限的時(shí)候也可以穩(wěn)定運(yùn)行。

04

Angel Graph應(yīng)用

在公司內(nèi)部，本平臺(tái)在推薦、風(fēng)控、社交、游戲等多個(gè)領(lǐng)域都有比較多的應(yīng)用。數(shù)據(jù)和場(chǎng)景支持同構(gòu)、異構(gòu)、有向、無(wú)向。節(jié)點(diǎn)和邊有無(wú)屬性，有監(jiān)督還是無(wú)監(jiān)督都是支持的，涵蓋了圖挖掘、圖表示學(xué)習(xí)、圖神經(jīng)網(wǎng)絡(luò)等多種算子。