黃奇鵬 盧 山

（1.武漢郵電科學(xué)研究院 武漢 430074）（2.南京烽火軟件科技有限公司 南京 210019）

1 引言

2 傳統(tǒng)關(guān)系數(shù)據(jù)處理技術(shù)

2.1 關(guān)系數(shù)據(jù)去重簡(jiǎn)介

公共WiFi設(shè)備每天會(huì)收到大量的請(qǐng)求日志、上下線日志信息等數(shù)據(jù)。其中，WiFi登錄賬號(hào)產(chǎn)生的上下線日志便是一種典型的關(guān)系數(shù)據(jù)，它包含數(shù)據(jù)源與登錄賬號(hào)在線狀態(tài)的對(duì)應(yīng)關(guān)系，可用于資源調(diào)查、城市應(yīng)急等公共社會(huì)服務(wù)，因此具有較強(qiáng)的研究與應(yīng)用價(jià)值。WiFi賬號(hào)登錄上下線日志數(shù)據(jù)結(jié)構(gòu)定義為表1。

表1 WiFi賬號(hào)登陸上下線日志數(shù)據(jù)結(jié)構(gòu)

據(jù)調(diào)查統(tǒng)計(jì)，南京某城區(qū)的公共WiFi一天約有1.6億活躍的登錄賬號(hào)，平均每個(gè)賬號(hào)1小時(shí)會(huì)登錄2～3次。一個(gè)賬號(hào)會(huì)在多個(gè)區(qū)域范圍內(nèi)登錄，隨著區(qū)域范圍、時(shí)間范圍的擴(kuò)大，產(chǎn)生的上下線數(shù)據(jù)量將是海量的。現(xiàn)需要對(duì)登陸賬號(hào)在線時(shí)的區(qū)域（數(shù)據(jù)源）、終端類(lèi)型進(jìn)行關(guān)系分析，一天的數(shù)據(jù)內(nèi)約有30%重復(fù)上下線數(shù)據(jù)（賬號(hào)、數(shù)據(jù)源、終端類(lèi)型均相同的數(shù)據(jù)），龐大的數(shù)據(jù)會(huì)導(dǎo)致數(shù)據(jù)庫(kù)查詢(xún)速度逐步退化，磁盤(pán)的消耗空間也會(huì)因冗余而浪費(fèi)。把這些冗余重復(fù)的上下線信息去掉將是一項(xiàng)非常重要且有意義的工作。

2.2 傳統(tǒng)關(guān)系數(shù)據(jù)去重處理技術(shù)

面對(duì)海量的關(guān)系數(shù)據(jù)，傳統(tǒng)海量數(shù)據(jù)采集系統(tǒng)通常會(huì)把它存在數(shù)據(jù)服務(wù)器中，數(shù)據(jù)將會(huì)以文件、數(shù)據(jù)庫(kù)形式保存，且關(guān)系型數(shù)據(jù)庫(kù)應(yīng)用更多。傳統(tǒng)數(shù)據(jù)去重處理往往采用的是單線程或多線程的處理方式。由于業(yè)務(wù)需要獲得地區(qū)內(nèi)所有出現(xiàn)過(guò)的WiFi登錄賬號(hào)以及每個(gè)對(duì)應(yīng)賬號(hào)使用過(guò)的終端類(lèi)型來(lái)進(jìn)行關(guān)系分析，因此需要對(duì)登錄賬號(hào)的上下線日志數(shù)據(jù)進(jìn)行去重，來(lái)確保登錄賬號(hào)與所使用的終端類(lèi)型當(dāng)中不會(huì)有重復(fù)的值影響分析效率。以O(shè)r?acle10g對(duì)WiFi上下線日志數(shù)據(jù)去重統(tǒng)計(jì)為例對(duì)傳統(tǒng)去重技術(shù)進(jìn)行說(shuō)明，其流程為

1）初步去重階段會(huì)對(duì)每個(gè)WiFi登錄賬號(hào)與所對(duì)應(yīng)的數(shù)據(jù)源為key，通過(guò)時(shí)間字段，進(jìn)行分段去重，得到每個(gè)時(shí)間段的去重結(jié)果，并將去重的結(jié)果存入一張臨時(shí)的表當(dāng)中。得到的數(shù)據(jù)沒(méi)有重復(fù)的時(shí)間字段。

2）二次去重將對(duì)初步去重得到的臨時(shí)表數(shù)據(jù)進(jìn)行merge（合并）操作，記錄每一個(gè)登錄賬號(hào)數(shù)據(jù)源，并記錄歷史最晚發(fā)現(xiàn)時(shí)間與所有登錄使用過(guò)終端類(lèi)型，將其錄入到一張用于去重比對(duì)的表當(dāng)中，建立當(dāng)日去重?cái)?shù)據(jù)的庫(kù)，即完成當(dāng)日數(shù)據(jù)的去重。

3）比對(duì)統(tǒng)計(jì)階段會(huì)刪除初步去重階段產(chǎn)生的臨時(shí)表，通過(guò)對(duì)當(dāng)天產(chǎn)生的二次去重產(chǎn)生的比對(duì)表與歷史產(chǎn)生的去重比對(duì)表進(jìn)行merge操作，更新同一賬號(hào)相同數(shù)據(jù)源的最晚發(fā)現(xiàn)時(shí)間與所有登錄使用過(guò)的終端類(lèi)型，將結(jié)果錄入新的比對(duì)表。即完成與歷史數(shù)據(jù)的去重。同時(shí)當(dāng)天產(chǎn)生的最終比對(duì)表，也會(huì)在次日會(huì)作為歷史產(chǎn)生的去重比對(duì)表與次日二次去重產(chǎn)生的比對(duì)表進(jìn)行比對(duì)。

由于本業(yè)務(wù)的特殊性，每天將會(huì)對(duì)臨時(shí)表進(jìn)行頻繁操作，導(dǎo)致涉及的表會(huì)比較多，進(jìn)而使維護(hù)工作變得繁瑣。同時(shí)業(yè)務(wù)需求主要對(duì)登陸賬號(hào)和相對(duì)應(yīng)使用過(guò)的終端類(lèi)型這兩列進(jìn)行關(guān)系分析。Or?acle是行式存儲(chǔ)，為了讀取這兩列，必須把整個(gè)表的行全部讀完，才可得到這些列。雖能完成業(yè)務(wù)需求，但是卻具有一定的局限性。若采用列式存儲(chǔ)，同一個(gè)列會(huì)擠在一個(gè)塊，讀某些列，只需要把相關(guān)列讀到內(nèi)存中這樣可以減少I(mǎi)/O量。

通過(guò)圖12對(duì)比發(fā)現(xiàn)，灌漿套筒連接件破壞過(guò)程分為彈性、屈服、強(qiáng)化、頸縮4個(gè)階段，試件H500-14承載力高出H400-14承載力10%左右，表明鋼筋強(qiáng)度的變化對(duì)試件承載力影響明顯，鋼筋強(qiáng)度越大，試件承載力越高。

使用MapReduce的開(kāi)源實(shí)現(xiàn)Hadoop搭建的平臺(tái)，能夠有效降低并行編程難度，提高編程效率［2］。它對(duì)海量數(shù)據(jù)進(jìn)行列式分布存儲(chǔ)、采用并行計(jì)算處理數(shù)據(jù)，可以解決本業(yè)務(wù)用Oracle處理的局限性。

3 基于MapReduce的海量關(guān)系數(shù)據(jù)去重技術(shù)

3.1 MapReduce去重的基本思想與體系架構(gòu)

MapReduce是一種并行處理框架，采用分而治之的思想，將一個(gè)龐大的數(shù)據(jù)處理任務(wù)，劃分成若干個(gè)小的計(jì)算過(guò)程與歸并［3～4］。其運(yùn)行的環(huán)境由客戶(hù)端、主節(jié)點(diǎn)和工作節(jié)點(diǎn)組成?？蛻?hù)端將用戶(hù)的并行處理作業(yè)提交給主節(jié)點(diǎn)，主節(jié)點(diǎn)自動(dòng)將作業(yè)分解為Map（分區(qū)）任務(wù)和Reduce（歸并）任務(wù)，并將任務(wù)調(diào)度到工作節(jié)點(diǎn)執(zhí)行［5］。

現(xiàn)以南京某城區(qū)WiFi上下線日志進(jìn)行去重為例，說(shuō)明MapReduce對(duì)關(guān)系數(shù)據(jù)去重的思想。其操作流程圖如圖1。

圖1 MapReduce操作流程圖

MapReduce模型首先會(huì)讀取數(shù)據(jù)源傳送過(guò)來(lái)的原始文件，將原始數(shù)據(jù)按照WiFi登錄賬號(hào)與數(shù)據(jù)源作為key（鍵）值，發(fā)現(xiàn)時(shí)間與終端類(lèi)型作為value（值）形成哈希散列，交給多個(gè)Mapper進(jìn)程進(jìn)行計(jì)算，接著以＜key，value＞對(duì)形式對(duì)中間結(jié)果進(jìn)行排序、合并，根據(jù)哈希值將這些結(jié)果交給多個(gè)Re?ducer進(jìn)程按進(jìn)行歸并計(jì)算。最后通過(guò)計(jì)算對(duì)哈希相同的key進(jìn)行合并，遍歷value值將數(shù)據(jù)按照時(shí)間維度進(jìn)行劃分，對(duì)一個(gè)時(shí)間段時(shí)間進(jìn)行排序，以最大的時(shí)間作為最后發(fā)現(xiàn)時(shí)間，并將使用過(guò)的終端類(lèi)型進(jìn)行匯總，進(jìn)而完成數(shù)據(jù)的去重流程。整個(gè)階段key與value的設(shè)定如圖2。

圖2 MapReduce輸出＜key，value＞圖

現(xiàn)基于MapReduce的運(yùn)算框架與海量關(guān)系數(shù)據(jù)的特點(diǎn)，提出一套完整的海量關(guān)系數(shù)據(jù)處理平臺(tái)架構(gòu)。體系架構(gòu)主要包括終端數(shù)據(jù)層、接入層、數(shù)據(jù)存儲(chǔ)與處理層、數(shù)據(jù)訪問(wèn)接口層、高級(jí)應(yīng)用層。海量關(guān)系數(shù)據(jù)處理平臺(tái)體系構(gòu)架，如圖3。

圖3 海量關(guān)系數(shù)據(jù)處理平臺(tái)體系構(gòu)架

1）終端數(shù)據(jù)層。作為需要處理的數(shù)據(jù)來(lái)源，將數(shù)據(jù)以一定的數(shù)據(jù)流格式發(fā)送到接入層；另外還負(fù)責(zé)設(shè)備保護(hù)。

2）接入層。接入層會(huì)做一些文件格式轉(zhuǎn)換與數(shù)據(jù)的映射傳給數(shù)據(jù)存儲(chǔ)與處理層。并且將對(duì)數(shù)據(jù)源數(shù)據(jù)進(jìn)行有效字段的篩選，減少后續(xù)層次的處理負(fù)荷，減少傳輸所耗的資源。

3）數(shù)據(jù)存儲(chǔ)與處理層。HDFS與HBase用于海量關(guān)系數(shù)據(jù)的存儲(chǔ)，MapReduce階段對(duì)數(shù)據(jù)進(jìn)行去重，以及相應(yīng)字段的更新。

4）訪問(wèn)接口層。對(duì)處理層的數(shù)據(jù)進(jìn)行協(xié)議還原，將之前映射的數(shù)據(jù)還原成原始數(shù)據(jù)。向應(yīng)用層提供一系列的數(shù)據(jù)訪問(wèn)接口，便于高級(jí)應(yīng)用層的調(diào)用。

5）應(yīng)用層。此層次將處理后的關(guān)系數(shù)據(jù)進(jìn)行前端展示、交互設(shè)計(jì)、數(shù)據(jù)分析等應(yīng)用。

3.2 基于MapReduce的關(guān)系數(shù)據(jù)去重技術(shù)改進(jìn)

由于MapReduce是存在一定局限性的，它要求任務(wù)之間必須有良好的可并行性，多個(gè)任務(wù)之間不存在數(shù)據(jù)共享，那么它無(wú)法完成當(dāng)日數(shù)據(jù)相對(duì)于歷史數(shù)據(jù)的去重任務(wù)；正常情況下每天約處理1.6億條數(shù)據(jù)，在節(jié)假日的時(shí)候，會(huì)比平時(shí)的數(shù)量還要多，在Reduce階段若直接利用HashSet集合去重，在一個(gè)Set集合中加入312萬(wàn)條已經(jīng)去重?cái)?shù)據(jù)記錄后，繼續(xù)添加記錄會(huì)導(dǎo)致堆內(nèi)存溢出。所以需要采取分塊讀取到內(nèi)存進(jìn)行哈希散列來(lái)改進(jìn)。

為解決上述問(wèn)題，現(xiàn)對(duì)MapReduce模型去重流程進(jìn)行歸并改進(jìn)。在Map和Reduce階段產(chǎn)生哈希散列的字段上進(jìn)行了如圖4調(diào)整。

圖4 改進(jìn)的MapReduce輸出＜key，value＞圖

整個(gè)處理流程將進(jìn)行2次MR（分區(qū)歸并）運(yùn)算進(jìn)行去重。首次去重以WiFi登錄賬號(hào)與數(shù)據(jù)源做MD5運(yùn)算，產(chǎn)生的MD5值與對(duì)應(yīng)兩個(gè)字段作為key值，發(fā)現(xiàn)時(shí)間與終端類(lèi)型作為value值。產(chǎn)生的中間結(jié)果不執(zhí)行Reduce運(yùn)算，通過(guò)中間數(shù)據(jù)的保存實(shí)現(xiàn)數(shù)據(jù)共享，在集群各計(jì)算節(jié)點(diǎn)上分別去重，從而保證局部數(shù)據(jù)中，沒(méi)有重復(fù)的WiFi登錄賬號(hào)。流程如圖5。

圖5 登錄賬號(hào)去重第一次MR運(yùn)算圖

為了進(jìn)行歷史去重，所以得將當(dāng)日去重的結(jié)果輸出到中間文件。根據(jù)第一步產(chǎn)生的n個(gè)中間結(jié)果文件，進(jìn)行歸并運(yùn)算。對(duì)于結(jié)果文件（tempFile），建立后綴樹(shù)，并依次再輸入后續(xù)文件（HistoryFile）中內(nèi)容，刪除后綴樹(shù)中重復(fù)的元素，即保留下來(lái)的數(shù)據(jù)都是不重復(fù)的數(shù)據(jù)，完成與歷史數(shù)據(jù)比對(duì)去重。第二次MR運(yùn)算的整體流程如圖6。

圖6 登錄賬號(hào)去重第二次MR運(yùn)算圖

第二次MR的運(yùn)算中，會(huì)對(duì)第一次運(yùn)行的結(jié)果進(jìn)行歸并，對(duì)MD5值進(jìn)行比對(duì)去重即做到了相對(duì)于歷史數(shù)據(jù)的去重。在Map階段，無(wú)論是歷史數(shù)據(jù)還是中間數(shù)據(jù)都會(huì)根據(jù)數(shù)據(jù)源與WiFi登錄賬號(hào)產(chǎn)生MD5值作為key，其他信息作為value。利用MD5值抗碰撞性，保證哈希散列在海量數(shù)據(jù)條件下不會(huì)沖突。根據(jù)MD5作為key，在進(jìn)入Reduce階段之前，執(zhí)行重寫(xiě)的partition的步驟，對(duì)數(shù)據(jù)進(jìn)行全局分區(qū)，保證全局域內(nèi)，所有相同的key都能進(jìn)入同一個(gè)Reduce，再根據(jù)哈希值進(jìn)行比對(duì)即完成去重。此方法利用分布式處理的優(yōu)勢(shì)，將整個(gè)哈希過(guò)程分解成了很多個(gè)小的歸并任務(wù)執(zhí)行，能避免使用HashSet去重因數(shù)據(jù)量大，導(dǎo)致內(nèi)存溢出的問(wèn)題。

4 實(shí)證分析

4.1 實(shí)驗(yàn)環(huán)境

實(shí)驗(yàn)采用的集群環(huán)境由2臺(tái)內(nèi)存為DDR4 16GB*4、硬盤(pán)為600G 10K*4的I620-G20服務(wù)器組成，機(jī)器均運(yùn)行RedHat操作系統(tǒng)，裝有Ha?doop2.60集群環(huán)境。輸入數(shù)據(jù)集結(jié)構(gòu)如表1所示。

4.2 效率對(duì)比測(cè)試

為了客觀對(duì)比傳統(tǒng)去重技術(shù)與MapReduce去重能力，因此采用4.1節(jié)提出的環(huán)境，分別用Oracle10g和MapReduce（5個(gè)節(jié)點(diǎn)）對(duì)同一批數(shù)據(jù)進(jìn)行去重統(tǒng)計(jì)處理（因500G過(guò)后MapReduce會(huì)發(fā)生堆內(nèi)存溢出，故只測(cè)到500G），得到的結(jié)果如圖7所示。

圖7 測(cè)試結(jié)果對(duì)比圖

此組實(shí)驗(yàn)使用的MapReduce流程未改進(jìn)，沒(méi)有經(jīng)過(guò)partition重寫(xiě)階段，在Map階段切割的時(shí)候key用的是WiFi登錄賬號(hào)與數(shù)據(jù)源。Value字段使用的值為其他信息，Reduce階段使用HashSet進(jìn)行去重。

通過(guò)圖7可見(jiàn)，當(dāng)數(shù)據(jù)量不大的時(shí)候（低于2GB），MapReduce數(shù)據(jù)平臺(tái)處理效率沒(méi)有傳統(tǒng)平臺(tái)Oracle處理效率高，100G與200G的時(shí)候，Ma?pReduce的效率略高于Oracle，到了500G的時(shí)候，已經(jīng)明顯比Oracle效率高。

4.3 改進(jìn)方案測(cè)試

將改進(jìn)處理流程后的MapReduce與常規(guī)的MapReduce對(duì)相同數(shù)據(jù)進(jìn)行去重處理進(jìn)行對(duì)比，結(jié)果如圖8。

圖8 測(cè)試結(jié)果圖

對(duì)數(shù)據(jù)分析發(fā)現(xiàn)，在處理文件小于200G以前，改進(jìn)流程后的消耗時(shí)間多于沒(méi)有改進(jìn)的消耗時(shí)間，經(jīng)分析發(fā)現(xiàn)，是由于在數(shù)據(jù)處理過(guò)程中，文件拆分歸并等基本操作上所需的定長(zhǎng)時(shí)間量所占比例過(guò)大，以及產(chǎn)生MD5值的運(yùn)算上。500G的時(shí)候改進(jìn)后的時(shí)間會(huì)少于沒(méi)有改進(jìn)的情況，此時(shí)原始Ma?pReduce的處理內(nèi)存消耗快達(dá)到飽和，導(dǎo)致運(yùn)行效率變慢。到800G的時(shí)候沒(méi)改進(jìn)流程的方案直接報(bào)OutOfMemoryError異常，而導(dǎo)致任務(wù)運(yùn)行失敗。改進(jìn)流程后的程序重寫(xiě)了partition的方法，因此在re?duce階段，會(huì)根據(jù)哈希散列分區(qū)且不會(huì)產(chǎn)生哈希沖突，即達(dá)到去重的目的，同時(shí)內(nèi)存也不會(huì)溢出。

4.4 實(shí)驗(yàn)小結(jié)

本節(jié)對(duì)傳統(tǒng)數(shù)據(jù)去重方式與MapReduce數(shù)據(jù)去重的方式進(jìn)行了效率的對(duì)比，驗(yàn)證了在海量數(shù)據(jù)的條件下對(duì)關(guān)系數(shù)據(jù)去重處理，MapReduce具有高效性。但是在小范圍的數(shù)據(jù)處理上，Oracle要優(yōu)于MapReduce。最后對(duì)改進(jìn)流程的MapReduce進(jìn)行了測(cè)試。由于進(jìn)行了歸并，所以會(huì)增加部分內(nèi)存消耗，但是隨著數(shù)據(jù)量的增大，改進(jìn)流程后的MapRe?duce運(yùn)行良好，系統(tǒng)的運(yùn)行效率有著不錯(cuò)的表現(xiàn)，且避免了內(nèi)存的溢出問(wèn)題。

5 結(jié)語(yǔ)

本文以WiFi上下線日志為例，借鑒傳統(tǒng)海量關(guān)系數(shù)據(jù)去重處理技術(shù)的經(jīng)驗(yàn)，提出了基于Ma?pReduce的海量關(guān)系數(shù)據(jù)去重處理技術(shù)，并論述了去重的基本思想與平臺(tái)架構(gòu)體系。就常規(guī)MapRe?duce模型對(duì)海量數(shù)據(jù)去重會(huì)遇到的問(wèn)題，提出了處理流程上的改進(jìn)。最后通過(guò)實(shí)驗(yàn)，驗(yàn)證了在海量條件下MapReduce對(duì)關(guān)系數(shù)據(jù)去重的效率。且初步驗(yàn)證了改進(jìn)后的性能，為后續(xù)進(jìn)一步研究用于海量關(guān)系數(shù)據(jù)處理技術(shù)提供了很好的參考依據(jù)。