基于Spark的分布式并行推理算法①

葉怡新,汪璟玢

(福州大學 數學與計算機科學學院,福州 350108)

基于Spark的分布式并行推理算法①

葉怡新,汪璟玢

(福州大學 數學與計算機科學學院,福州 350108)

現有的RDF數據分布式并行推理算法大多需要啟動多個MapReduce任務,有些算法對于含有多個實例三元組前件的OWL規則的推理效率低下,使其整體的推理效率不高.針對這些問題,文中提出結合TREAT的基于Spark的分布式并行推理算法(DPRS).該算法首先結合RDF數據本體,構建模式三元組對應的alpha寄存器和規則標記模型;在OWL推理階段,結合MapReduce實現TREAT算法中的alpha階段;然后對推理結果進行去重處理,完成一次OWL全部規則推理.實驗表明DPRS算法能夠高效正確地實現大規模數據的并行推理.

RDF;OWL;分布式推理;TREAT;Spark

語義萬維網中的RDF和OWL標準已在各個領域有著廣泛的應用,如一般知識(DBpedia[1])、醫療生命科學(LODD[2])、生物信息學(UniProt[3])、地理信息系統(Linkedgeodata)和語義搜索引擎(Watson)等.隨著語義萬維網的應用,產生了海量的語義信息.由于數據的復雜性和大規模性,如何通過語義信息并行推理高效地發現其中隱藏的信息是一個亟待解決的問題.由于語義網數據的急速增長,集中式環境的內存限制,已不適用于大規模數據的推理.

研究RDFS/OWL分布式并行推理是目前較新的一個領域.J.Urbani[4-6]等人在RDFS/OWL規則集上采用WebPIE進行推理,能夠滿足大數據的并行推理;但該算法針對每一條規則啟用一個或者多個MapReduce任務進行推理,由于Job的啟動相對耗時,因此隨著RDFS/OWL推理規則的增加,整體推理的效率受到了限制.顧榮[7]等人提出了基于MapReduce的高效可擴展的語義推理引擎(YARM),使推理在一次MapReduce任務內即可完成RDFS規則的推理;但該算法并不適用于復雜的OWL規則的推理.此外,當某一規則產生的新三元組重復時,YARM會存在過多的冗余計算且產生無用數據.汪璟玢[8]等人提出結合Rete的RDF數據分布式并行推理算法,該算法結合RDF數據本體,構建模式三元組列表和規則標記模型;在RDFS/OWL推理階段,結合MapReduce實現Rete算法中的alpha階段和beta階段,從而實現Rete算法的分布式推理;但該算法在連接beta網絡推理時需要消耗較多的內存且進行多次迭代時效率低下,因而此算法受到集群內存和平臺的限制.顧榮[9]等人提出了一種基于Spark的高效并行推理引擎(Cichlid),結合RDD的編程模型,優化了并行推理算法;但該算法未考慮規則能否被激活,均需要進行推理,因而造成了推理性能的浪費和傳輸的冗余.

為了解決上述問題,本文針對OWL Horst規則,提出 了 DPRS算 法 (Distributed parallel reasoning algorithm based on Spark).該算法結合TREAT[10]算法和RDF數據本體構建模式三元組的alpha寄存器RDD,預先對規則能否被激活做出判斷并標記,僅對可激活的規則進行推理的處理,實現在一個MapReduce任務中完成OWL全部規則的一次推理.最后,實時地刪除重復的三元組數據和更新沖突集數據到相應的寄存器中,以進一步提高后續迭代推理的效率.實驗表明,該算法在數據量動態增加的情況下能夠高效地構建alpha網絡,并執行正確的推理.

1 基本定義

定義1.模式三元組(SchemaTriple),指三元組的主語謂語和賓語都在本體文件(OntologyFile)中有定義.即:

其中,n表示模式三元組的總數.若v∈{Si,Pj,Ok}, v∈OntologyFile,則:

定義2.實例三元組(InstanceTriple),指主語謂語和賓語至少有一個在本體文件(OntologyFile)中未定義,是具體的實例.即:

其中,n表示實例三元組的總數.若v∈{Si,Pj,Ok},?v?OntologyFile,則:

定義3.三元組類型標記(Flag_TripleType),用于標識模式三元組與實例三元組,結合定義1和定義2,三元組類型標記Flag_TripleType定義如下:

其中,n表示三元組的總數.則:

定義4.模式三元組列表(SchemaRDD).用于獲取相同謂語或者賓語的模式三元組集合.結合定義1,模式三元組列表SchemaRDD定義如下:

其中,n表示模式三元組的總數.則,

其中,Om_RDD表示滿足謂語Pj∈{rdf:type}且具有相同賓語的三元組集合,以該賓語命名;Pt_RDD表示滿足謂語Pj?{rdf:type}的所有具有相同謂語的三元組集合,以該謂語命名.具體定義如下:

定義 5.連接變量(LinkVar).連接變量為在RDFS/OWL規則中用于連接兩個前件的模式三元組項,根據規則描述,連接變量可以不止一個.本文將每一條規則的連接變量信息以的形式存儲在Rulem_RDD,其中key存儲該規則所有用于前件連接的模式三元組項,value存儲該規則結論部分的模式三元組項.

DPRS算法根據連接變量的類型,對OWL Horst規則進行分類.本文引用OWL Horst規則時采用OWL-規則編號的形式,例如OWL-4表示圖1中的第4條規則.同時,給每條規則分配一個規則名稱標記,規則名稱標記即為該規則所對應的名稱(例如,規則OWL-4的規則名稱標記為OWL-4).具體的規則分類如下:

1)類型 1:只包含一個前件的規則或SchemaTriple與InstanceTriple組合的規則,且只有一個InstanceTriple,可以在Map推理過程中直接輸出推理結果(圖1中規則OWL-3、OWL-5a、OWL-5b、OWL-6、OWL-8a、OWL-8b、OWL-9、OWL-12a、OWL-12b、OWL-12c、OWL-13a、OWL-13b、OWL-13c、OWL-14a、OWL-14b).

2)類型2:SchemaTriple與InstanceTriple組合的規則,且有多個InstanceTriple的,需要結合 Map和ReduceByKey兩個階段推理(圖1中規則OWL-1、OWL-2、OWL-4、OWL-7、OWL-15、OWL-16).

定義 6.設Cmn為第m條規則的第n個模式三元組前件,定義規則前件模式標記Indexmn, 用于標識是否有符合該前件的模式三元組存在,即以該模式三元組前件Cmn所命名的SchemaRDD是否為空.結合定義4,規則前件模式標記Indexmn定義如下:

定義7.規則標記Flag_Rule_m,用于標記該規則是否為不可能激活的規則.結合定義6進行定義規則標記Flag_Rulem如下:

Flag_Rulem={0,1,2}其中,規則不能激活時,Flag_Rulem=0;規則激活且為類型1時,Flag_Rulem=1;規則激活且為類型2時, Flag_Rulem=2.

由于圖1中OWL規則5a、5b不影響推理的并行化,因而,本文所述推理不考慮這兩條規則.

圖1 OWLHorst規則

2 DPRS算法

在Rete算法中,同一規則連接結點上的寄存器保留了大量的冗余結果.實際上,寄存器中大部分信息已經體現在沖突集的規則實例中.因此,如果在部分匹配過程中直接使用沖突集來限制模式之間的變量約束,不僅可以減少寄存器的數量,而且能夠加快匹配處理效率.這一思想稱為沖突集支撐策略.基于沖突集支撐策略,TREAT[10]算法放棄了Rete算法中利用β寄存器保存模式之間變量約束中間結果的思想.

DPRS算法根據Spark RDD的特點,結合TREAT算法的原理,首先根據RDF本體數據構建模式三元組對應的alpha寄存器Om_RDD或Pt_RDD并廣播,然后對每條規則的模式前件進行連接并生成對應的連接模式三元組集合Rulem_linkvar_RDD,從而加快推理過程中的匹配速度,能夠實現多條規則的分布式并行推理.DPRS算法主要包括以下幾個步驟:

Step1.加載模式三元組集合Pt_RDD、Om_RDD和Rulem_linkvar_RDD并廣播.

Step2.構建規則標記模型Flag_Rulem并廣播.

Step3.并行執行OWL Horst規則推理.

Step4.刪除重復三元組.

Step5.如果產生新的模式三元組數據,則跳至步驟Step2,如果產生新的實例三元組數據,則跳至步驟Step3,否則跳至步驟Step6.

Step6.算法結束.

DPRS算法的總體框架圖如圖2所示.

圖2 DPRS算法總體框架圖

2.1 加載模式三元組與構建規則標記模型

由于模式三元組的數量遠遠少于實例三元組, DPRS算法將SchemaTriple加載到SchemaRDD中并廣播.并構建每條規則中的模式三元組或模式三元組連接后的數據(Rulem_linkvar_RDD或 Om_RDD或Pt_RDD)為 alpha寄存器并廣播,保存對應的SchemaTriple.

為了盡早判斷出不可能被激活的規則,DPRS算法根據OWL規則構建每一條規則內SchemaTriple間的關系Om_RDD或Pt_RDD,并判斷SchemaRDD中是否存在規則前件中的SchemaTriple,生成對應規則的標記Flag_Rulem,構建所有規則的標記模型,將規則標記模型加載到Flag_Rulem并廣播.

通過SchemaRDD和構建規則標記模型能夠過濾大量InstanceTriple,減少Map階段鍵值對的輸出,從而減少了無效的網絡傳輸,提高整體推理效率.

2.2 Map階段

Map階段主要完成數據選擇過濾與類型1推理,將過濾的結果以鍵值對的形式輸出,本文提出的數據分配與過濾算法具體步驟如下:

Step1.獲取廣播變量中的Om_RDD、Pt_RDD和

Rulem_linkvar_RDD以及規則標記Flag_Rulem.

Step2.對于輸入的?(Si,Pj,Ok)∈InstanceTriple判斷所有Flag_Rulem的值.如果值為0,則跳至Step3,如果值為1,則跳至Step4,否則跳至Step5.

Step3.對(Si,Pj,Ok)不做任何處理.

Step4.結合Om_RDD或Pt_RDD執行類型1的規則推理,根據規則的結論直接輸出對應的三元組

Step5獲取對應規則中的模式三元組alpha寄存器Om_RDD、Pt_RDD或Rulem_linkvar_RDD,判斷當前的實例三元組是否滿足前件連接變量的條件;滿足,則構建對應的鍵值對輸出;不滿足,則不做處理.

以圖1中規則8a(inverseOf)為例,偽碼描述如下:

類似于規則8,推理可以在Map階段就得到規則產生的三元組結果,那么reduce階段就可以對規則8產生的三元組去重并輸出.

以圖1中規則9(type+sameAs)為例,偽碼描述如下:

以圖1中規則15(someValuesFrom)為例,偽碼描述如下:

如上所描述的規則9和15,以規則9為例,在Map階段需要對輸入的三元組進行處理,以“Rule9+連接變量”為key,如果謂語為type,那么value中標記為type且資源為連接變量;如果謂語為sameAs,那么value中標記為sameAs且資源為賓語.

2.3Reduce階段

Reduce階段主要完成連接推理.利用RDD的reduceByKey,結合OWL規則,根據SchemaRDD和alpha寄存器以及Map階段的InstanceTriple輸出結果完成連接推理,得到推理結果.本文提出的連接推理算法具體步驟如下:

Step1.獲取廣播變量中的Om_RDD、Pt_RDD和Rulem_linkvar_RDD以及規則標記Flag_Rulem.

Step2.獲取相同鍵對應的迭代器;如果key為Rulem,則表示為類型1,直接將value的三元組輸出;如果key為Rulem_linkvar,則表示為類型2,則根據該key對應的OWL規則和連接變量,結合alpha寄存器Rulem_linkvar_RDD與value迭代器完成連接推理,得到推理結果并輸出連接后的三元組在執行連接推理過程中,因為符合條件的SchemaTriple已經在構建alpha寄存器時已連接完畢,所以只需要執行SchemaTriple與InstanceTriple或InstanceTriple與InstanceTriple間的連接即可.

為了更加明確Reduce階段的連接推理,以圖1中規則9(type+sameAs規則)為例,偽碼描述如下:

以圖1中規則15(someValuesFrom規則)為例,偽碼描述如下:

由上述的規則9和規則15的偽碼,以規則9為例,在Reduce階段,根據輸入的key和values,我們通過values中的flag值來進行區分并構建輸出的三元組.

2.4 刪除重復三元組和沖突集更新策略

在執行算法推理的過程中會產生大量重復的三元組數據到沖突集中,如不刪除沖突集中的重復三元組,則更新alpha寄存器時將會產生重復三元組數據,浪費系統資源,降低推理效率.如果每次推理后都能夠及時刪除沖突集中的重復三元組,那將會減少很大的網絡傳輸開銷.本文借助RDD的distinct和subtract完成刪除重復三元組算法.

通過上述的刪除重復三元組后,沖突集中的模式三元組分別更新到對應的alpha寄存器中,實例三元組合則并到實例文件中.

2.5 算法的復雜度與完備性

復雜性分析是算法分析的核心,DPRS算法的復雜性與集中式算法復雜性的分析不太相同,將DPRS算法的最壞情況下的時間復雜性分為Map階段的時間復雜性和Reduce階段的時間復雜性.假設數據集的規模大小為N個三元組,其中模式三元組為n個,在MapReduce中Map階段的并行數為k,Reduce階段傳入的實例三元組個數為m,Reduce階段的并行數為t.

由于DPRS算法在Map階段對每個輸入的三元組,結合SchemaList、Flag_Rulem掃描一次,即可判斷該三元組是該舍棄或是能參與某些規則推理,如能參與后續規則推理,則以該規則名稱為key結合此三元組輸出.因此,Map階段的時間復雜性為:O(n*N/k).

由于圖1 OWL規則中,規則1、2、3、4、15、16都含有兩個實例三元組前件,將上述規則稱作多實例變量規則,多實例變量規則的Reduce階段則需要遍歷兩次輸入的實例三元組與模式三元組連接,才能得到推理結果.因此在Reduce階段的時間復雜性分為單變量和多變量進行分析.

Reduce階段多變量的時間復雜性為:O(n*m/t).由于n的數目非常少,可以認為其量級為常數.

DPRS算法首先將數據集中的模式三元組載入內存并廣播,根據定義7和OWL規則的描述構建各個規則的Flag_Rulem,從而過濾掉不可能激活的規則.在能被激活的規則并行推理過程中的Map階段,對于輸入的一個三元組,DPRS判斷其是否滿足某個規則前件,只要滿足,就將此規則名稱作為鍵(key),值(value)為該三元組輸出;若一個三元組數據滿足多個規則前件,我們也將據此方法產生多個不同鍵(key)的輸出,以保障Reduce階段推理連接的正確性和數據完整性.如果Reduce階段產生的三元組去重后,有產生新的模式三元組,那么 DPRS算法將重新計算各個規則的Flag_Rulem,再執行規則的并行推理迭代;如果Reduce階段產生的三元組去重后產生的是實例三元組,那么DPRS算法直接執行規則的并行推理迭代,直到沒有新的三元組數據產生為止.因而DPRS算法所得到的推理結果是完備的.

3 實驗與結果分析

實驗所使用的軟件環境為操作系統Linux Ubuntu,采用scala作為編程語言,開發環境為IntelliJIDEA.在實驗環境中,用表1所示配置作為本系統Spark集群的配置,共計8臺,其Hadoop集群中1臺作為HDFS的名稱節點,1臺作為JobTracker節點,6臺為HDFS的數據節點和TaskTracker節點,Spark集群中1臺作為Master兼Worker節點,7臺作為Worker節點.集群工作站的基本配置如表1所列.

表1 Hadoop集群工作站的基本配置

本文將DPRS算法與DRRM[4]和Cichlid-OWL[9]在相同的實驗環境下針對不同的數據集進行對比實驗.本實驗采用LUBM[11](Lehigh University Benchmark)數據集和DBpedia[1]數據集進行測試.數據集的基本參數說明如表2所列.

表2 數據集的基本參數說明

我們將實驗數據集中的模式三元組數進行統計如表2所示,與整個數據集的大小相比,模式三元組的數量非常少,在所測試的數據集范圍內,模式三元組數目最高僅僅達到了整個數據集的0.04%.

表3 不同數據集三種算法在OWL推理產生的三元組數對比

表4 不同數據集三種算法在OWL推理時間對比

表5 DPRS算法在不同數據集上執行推理的數據

從表3和表4可知,在OWL規則推理結果一致的情況下,DPRS比Cichlid-OWL具有優勢.其中,由于LUBM數據集本體比較簡單,OWL Horst中的許多規則無法被激活,所以DPRS相比Cichlid-OWL的優勢比較微弱;對于比較復雜的DBpedia本體而言,OWL的大部分規則都可被激活,由于本文使用了alpha寄存器廣播、連接變量、規則標記和沖突集更新策略,使得DPRS算法的推理時間相對Cichlid-OWL算法最大縮短了21%的時間.

另外,DPRS與DRRM相比均有較大的優勢.首先, DPRS使用Spark平臺比DRRM使用的Hadoop具有迭代性能優勢;再者,DPRS采用沖突集更新策略,避免了beta網絡的開銷,大大減少了傳輸冗余造成的浪費.使得DPRS算法的推理時間相對DRRM算法最大縮短了73.8%的時間.

根據2.5節的復雜度分析,其中k和t為常數,所以推理時間的復雜度與N和m成線性關系.結合表4和表5,考察數據集LUBM50和LUBM200,實例三元組個數N的比例為1:4.01,傳入Reduce的實例三元組數m的比例為1:3.89,推理時間的比例為1:4.20;考察數據集DBpedia3.7和DBpedia3.9,實例三元組個數N的比例為1:1.53,傳入Reduce的實例三元組數m的比例為1:1.41,推理時間的比例為1:1.35.可以發現,我們的推理時間基本是與N和m成線性關系.從實驗結果上符合了理論的分析,證明了算法的正確性.

從圖3和圖4可知,在執行OWL規則推理時,雖然兩種算法都需要多次迭代才能使得推理最終停止,但是DPRS在推理前構建并廣播了模式三元組的alpha寄存器,并且在每次迭代中采用高效的過濾機制,過濾掉大量的實例三元組數據,減少了并行計算量和網絡傳輸的開銷,使得DPRS算法在最終的推理時間較Cichlid-OWL略占優勢,尤其是在DBpedia數據集下,從表2中可以看出,Dbpedia的模式三元組占比相對LUBM高,且數據集較為復雜,其優勢更加明顯.

圖3 采用未預處理與預處理算法在OWL推理的時間對比

圖4 采用過濾算法與未過濾算法產生的中間結果數目對比

由于在執行推理過程中會產生重復的三元組數據,重復三元組數據會造成系統資源無謂的浪費并增加網絡的開銷.文中3.4節提出的刪除重復三元組算法,能夠減少重復的三元組數據.為了評估算法的有效性,將刪除重復三元組前后的數據量進行對比如圖5所示.刪除重復三元組后的三元組數量少于推理三元組數量,在所測試的數據范圍內.

圖5 刪除重復三元組前后三元組數量對比

4 結語

本文提出的 DPRS算法能夠通過執行一次MapReduce任務就完成OWL所有規則的一次推理,彌補了現有方法大多需要啟動多個MapReduce任務以及在大規模數據下無法對OWL規則中含有實例三元組的規則進行推理的問題.DPRS算法能夠在MapReduce計算框架下高效地實現大規模數據的并行推理,但無法對流式數據進行推理.下一步將會在此方面進行改進,且研究更深一步的OWLDL推理.

1 Auer S,Bizer C,Kobilarov G,et al.Dbpedia:A nucleus for a web of open data.The Semantic Web.Springer Berlin Heidelberg.2007.722–735.

2 Jentzsch A,Zhao J,Hassanzadeh O,et al.Linking Open Drug Data.I–SEMANTICS.2009.

3 Apweiler R,Bairoch A,Wu CH,et al.UniProt:The universal protein knowledgebase.Nucleic AcidsResearch,2004, 32(s1):D115–D119.

4 Urbani J,Kotoulas S,Maassen J,et al.WebPIE:A web-scale parallel inference engine using MapReduce.Web Semantics: Science,Services and Agents on the World Wide Web,2012, (10):59–75.

5 Urbani J,Kotoulas S,Maassen J,et al.OWL reasoning with WebPIE:Calculating the closure of 100 billion triples. Extended Semantic Web Conference.SpringerBerlin Heidelberg.2010.213–227.

6 UrbaniJ.On web-scale reasoning[PhD.dissertation]. Amsterdam,Netherlands:Computer Science Department, Vrije Universiteit,2013.

7顧榮,王芳芳,袁春風,等.YARM:基于MapReduce的高效可擴展的語義推理引擎.計算機學報,2015,38(1):74–85.

8汪璟玢,鄭翠春.結合Rete的RDF數據分布式并行推理算法.模式識別與人工智能,2016,(5):5.

9 Gu R,Wang S,Wang F,et al.Cichlid:Efficient large scale RDFS/OWL reasoning with spark.Parallel and Distributed Processing Symposium(IPDPS),2015 IEEE International. IEEE.2015.700–709.

10 Miranker DP.TREAT:A new and efficient match algorithm forAI production system.Morgan Kaufmann,2014.

11 Guo Y,Pan Z,Heflin J.LUBM:A benchmark for OWL knowledge base systems.Web Semantics:Science,Services andAgents on the World Wide Web,2005,3(2):158–182.

Distributed Parallel ReasoningAlgorithm Based on Spark

YE Yi-Xin,WANG Jing-Bin

(College of Mathematics and Computer Science,Fuzhou University,Fuzhou 350108,China)

Multiple MapReduce tasks are needed for most of current distributed parallel reasoning algorithm for RDF data;moreover,the reasoning of instances of triple antecedents under OWL rules can’t be performed expeditiously by some of these algorithms during the processing of massive RDF data,and so the overall efficiency can’t be fulfilled in reasoning process.In order to solve the problems mentioned above,a method named distributed parallel reasoning algorithm based on Spark with TREAT for RDF data is proposed to perform reasoning on distributed systems.First step, alpha registers of schema triples and models for rule markup with the ontology of RDF data are built;then alpha stage of TREAT algorithm is implemented with MapReduce at the phase of OWL reasoning;at last,reasoning results are dereplicated and a whole reasoning procedure within all the OWL rules is executed.Experimental results show that through this algorithm,the results of parallel reasoning for large-scale data can be achieved efficiently and correctly.

RDF;OWL;distributed reasoning;TREAT;Spark

國家自然科學基金(61300104)

2016-09-21;收到修改稿時間:2016-10-31

10.15888/j.cnki.csa.005790