肖榕 金甌 葉建鋒

摘 要： 船聯(lián)網(wǎng)項目涉及到全國范圍的內(nèi)河航運數(shù)據(jù)，需要連通所有省級航運數(shù)據(jù)中心并接入數(shù)據(jù)，因此需要具備分布式、可線性擴展的并行計算能力。針對船聯(lián)網(wǎng)項目中智能航運數(shù)據(jù)處理場景，參考國內(nèi)外成熟的理論基礎(chǔ)，提出了一種兩級分布式彈性計算技術(shù)，介紹了其計算模型和故障處理機制，并描述了該計算框架的應(yīng)用場景。這種計算技術(shù)可以完成海量航運數(shù)據(jù)的處理任務(wù)，滿足上層航運信息服務(wù)的需求，并適應(yīng)今后不斷增長的數(shù)據(jù)量和計算規(guī)模。

關(guān)鍵詞： 內(nèi)河航運； 彈性計算技術(shù)； 兩級分布式； 數(shù)據(jù)處理

中圖分類號：TP301 文獻標志碼：A 文章編號：1006-8228（2014）06-34-04

0 引言

內(nèi)河航運是我國綜合運輸體系的重要組成部分，在我國經(jīng)濟發(fā)展中起著非常重要的作用。針對航運市場發(fā)展的總體態(tài)勢，發(fā)揮信息化對水路運輸行業(yè)現(xiàn)代化發(fā)展的支撐和引領(lǐng)作用，以信息資源共享為基礎(chǔ)，以信息服務(wù)為核心，以現(xiàn)代物聯(lián)網(wǎng)技術(shù)、通信技術(shù)和信息技術(shù)為支撐，以促進水路交通運輸產(chǎn)業(yè)和物聯(lián)網(wǎng)產(chǎn)業(yè)發(fā)展為目標，進行智能航運信息服務(wù)應(yīng)用的研究。

物聯(lián)網(wǎng)是通過基礎(chǔ)設(shè)施層的各種感知設(shè)備感知物體信息，然后利用網(wǎng)絡(luò)技術(shù)將感知獲取的海量數(shù)據(jù)傳輸至物聯(lián)網(wǎng)數(shù)據(jù)中心，物聯(lián)網(wǎng)信息服務(wù)系統(tǒng)整合了海量數(shù)據(jù)，為各種物聯(lián)網(wǎng)應(yīng)用提供信息服務(wù)并接收反饋控制[4]。而船聯(lián)網(wǎng)是物聯(lián)網(wǎng)的一個典型的應(yīng)用，船聯(lián)網(wǎng)以船舶、航道、陸岸設(shè)施為基本節(jié)點和信息源，結(jié)合具有衛(wèi)星定位系統(tǒng)、無線通信技術(shù)的船載智能信息服務(wù)，利用船載電子傳感裝置，通過網(wǎng)絡(luò)完成信息交換，在網(wǎng)絡(luò)平臺上完成各節(jié)點的屬性和動/靜態(tài)信息的進行提取、監(jiān)管和利用[5]。

1 平臺架構(gòu)

船聯(lián)網(wǎng)項目采用一個中心、多個省級分節(jié)點的部署架構(gòu)，借助目前成熟的MapReduce計算模型，結(jié)合智能航運數(shù)據(jù)處理的實際需要，本文提出一種兩級分布式計算平臺：中心和各省分節(jié)點之間組成一個船聯(lián)網(wǎng)全局性的分布式平臺；中心節(jié)點自身建設(shè)為一個局域分布式平臺。這里稱大平臺為全局分布式框架，中心節(jié)點的局域平臺為中心分布式框架，示意如圖1。

船聯(lián)網(wǎng)全局分布式框架廣域部署，通過中心節(jié)點調(diào)度各省資源，支撐船聯(lián)網(wǎng)省級航運業(yè)務(wù)應(yīng)用，如區(qū)域船舶監(jiān)控、應(yīng)急指揮等。中心分布式框架在中心節(jié)點局域部署，通過中心主節(jié)點調(diào)度中心分節(jié)點資源，支撐項目全局性的業(yè)務(wù)應(yīng)用，如綜合監(jiān)管、統(tǒng)計分析等。

這兩級分布式計算框架都需要能夠處理航運海量的數(shù)據(jù)，比如船舶船員基本信息、船舶簽證、船舶動態(tài)感知數(shù)據(jù)、航運地理信息數(shù)據(jù)等。從區(qū)別上講，全局分布式框架側(cè)重各省級分節(jié)點與中心節(jié)點、分節(jié)點之間的互聯(lián)互通和數(shù)據(jù)共享；中心分布式框架用于完成中心節(jié)點各種數(shù)據(jù)計算任務(wù)，并為全局分布式框架的計算調(diào)度提供支持。

2 計算模型

2.1 全局分布式框架計算模型

船聯(lián)網(wǎng)全局分布式框架的計算模型如下。

⑴ 各省級分節(jié)點處理各省航運數(shù)據(jù)，包括數(shù)據(jù)的整合、清洗、轉(zhuǎn)換等，即分節(jié)點可以處理的計算任務(wù)直接在分節(jié)點完成。

⑵ 中心節(jié)點整合各個分節(jié)點上報的數(shù)據(jù)，包括船舶船員基本信息、船舶簽證信息、動態(tài)感知信息等。

⑶ 船舶船員基本信息、編碼等靜態(tài)數(shù)據(jù)由中心節(jié)點輪詢各分節(jié)點采集；船舶簽證信息、感知信息等動態(tài)數(shù)據(jù)由分節(jié)點主動推送。

⑷ 各分節(jié)點只與中心節(jié)點聯(lián)系，分節(jié)點之間不直接交換數(shù)據(jù)。

2.1.1 省級分節(jié)點的數(shù)據(jù)處理

省級分節(jié)點負責從各省航運業(yè)務(wù)系統(tǒng)采集數(shù)據(jù)，完成省一級的數(shù)據(jù)整合和清洗，同時可以支撐省一級的航運應(yīng)用。

例如，A省有一GPS系統(tǒng)采集船舶位置信息，則A省分節(jié)點負責從該GPS系統(tǒng)采集船舶位置信息并存儲，然后完成數(shù)據(jù)的校驗清洗，再按要求將該數(shù)據(jù)上傳中心節(jié)點。A省如有另一業(yè)務(wù)系統(tǒng)只需要本省船舶GPS信息，則直接從A省分節(jié)點獲取，不需要向船聯(lián)網(wǎng)中心節(jié)點發(fā)起請求。

2.1.2 中心節(jié)點的數(shù)據(jù)處理

船聯(lián)網(wǎng)中心節(jié)點負責接收各個省級分節(jié)點上傳的已初步清洗的數(shù)據(jù)，然后再作最終的數(shù)據(jù)整合。由于在各分節(jié)點已清洗過，已屏蔽掉大量數(shù)據(jù)問題，所以中心節(jié)點只需要處理少量計算（主要是再一次檢驗的計算量）即可完成數(shù)據(jù)整合。

完成整合后的數(shù)據(jù)在中心節(jié)點落地，交由中心節(jié)點作進一步的分析處理。船聯(lián)網(wǎng)在整合數(shù)據(jù)的基礎(chǔ)上提供全局性的數(shù)據(jù)服務(wù)支持，比如跨區(qū)域的數(shù)據(jù)服務(wù)。

2.1.3 節(jié)點間的數(shù)據(jù)傳輸機制

船聯(lián)網(wǎng)節(jié)點間的數(shù)據(jù)傳輸將只在中心節(jié)點和分節(jié)點之間進行，分節(jié)點之間相互隔離。中心節(jié)點定期輪詢各個省級分節(jié)點，將船舶基本信息等主數(shù)據(jù)采集上來（分節(jié)點提供相應(yīng)的數(shù)據(jù)庫接口），并監(jiān)控各分節(jié)點的運行狀態(tài)。各個省級分節(jié)點通過消息服務(wù)，定期將簽證、感知信息等動態(tài)實時數(shù)據(jù)推送給中心節(jié)點。

2.2 中心分布式框架計算模型

船聯(lián)網(wǎng)中心節(jié)點分布式框架計算模型使用目前較為成熟的MapReduce計算模型。目前世界上最快的1TB排序記錄就是由基于MapReduce實現(xiàn)的。

MapReduce將計算任務(wù)劃分為map和reduce兩個階段。map階段負責“分”，即把復(fù)雜的任務(wù)分解為若干個“簡單的任務(wù)”執(zhí)行?！昂唵蔚娜蝿?wù)”有以下幾個含義：

⑴ 數(shù)據(jù)或計算規(guī)模相對于原任務(wù)要大大縮??；

⑵ 就近計算，即任務(wù)會被分配到存放了所需數(shù)據(jù)的節(jié)點進行計算；

⑶ 這些小任務(wù)可以并行計算，彼此間幾乎沒有依賴關(guān)系。

reduce階段負責對map階段輸出的結(jié)果進行匯總，即將分割開的任務(wù)合并，將與一個key關(guān)聯(lián)的一組中間數(shù)值集歸約為一個更小的數(shù)值集，輸出最終的計算結(jié)果。

2.2.1 MapReduce執(zhí)行機制[1-3]

通過將Map調(diào)用的輸入數(shù)據(jù)自動分割為M個數(shù)據(jù)片段的集合，Map被分布到多臺機器上調(diào)用執(zhí)行。輸入的數(shù)據(jù)片段能夠在不同的機器上并行處理。使用分區(qū)函數(shù)將Map調(diào)用產(chǎn)生的中間key值分成R個不同分區(qū)，Reduce調(diào)用也被分布到多臺機器上執(zhí)行。分區(qū)數(shù)量和分區(qū)函數(shù)由用戶來指定。

⑴ 計算程序首先調(diào)用的MapReduce庫將輸入文件分成M個數(shù)據(jù)片段，每個數(shù)據(jù)片段的大小一般從 16MB到64MB（可以通過備選參數(shù)來控制每個數(shù)據(jù)片段的大?。Ｈ缓笥嬎愠绦蛟跈C群中創(chuàng)建大量的程序副本。

⑵ 這些程序副本中有一個特殊的程序-master。副本中其他程序都是worker程序，由master分配任務(wù)。有M個map任務(wù)和R個reduce任務(wù)將被分配，master將一個map任務(wù)或reduce任務(wù)分配給一個空閑的worker。

⑶ 被分配了map任務(wù)的worker程序讀取相關(guān)的輸入數(shù)據(jù)片段，從輸入的數(shù)據(jù)片段中解析出key/value對，然后把key/value對傳遞給計算程序自定義的map函數(shù)，由map函數(shù)生成并輸出的中間key/value對，并緩存在內(nèi)存中。

⑷ 緩存中的key/value對通過分區(qū)函數(shù)分成R個區(qū)域，之后周期性的寫入到本地磁盤上。緩存的key/value對在本地磁盤上的存儲位置將被回傳給master，由master負責把這些存儲位置再傳送給reduce worker。

⑸ 當reduce worker程序接收到master程序發(fā)來的數(shù)據(jù)存儲位置信息后，使用RPC從map worker所在主機的磁盤上讀取這些緩存數(shù)據(jù)。當reduce worker讀取了所有的中間數(shù)據(jù)后，通過對key進行排序后使得具有相同key值的數(shù)據(jù)聚合在一起。由于許多不同的key值會映射到相同的reduce任務(wù)上，因此必須進行排序。如果中間數(shù)據(jù)太大無法在內(nèi)存中完成排序，那么就要在外部進行排序。

⑹ reduce worker程序遍歷排序后的中間數(shù)據(jù)，對于每一個惟一的中間key值，reduce worker程序?qū)⑦@個key值和它相關(guān)的中間value值的集合傳遞給用戶自定義的reduce函數(shù)。reduce函數(shù)的輸出被追加到所屬分區(qū)的輸出文件。

⑺ 當所有的map和reduce任務(wù)都完成之后，master喚醒計算程序。在這個時候，在計算程序里的對MapReduce調(diào)用才返回。

2.2.2 結(jié)果數(shù)據(jù)的處理

在成功完成任務(wù)之后，MapReduce的輸出存放在R個輸出文件中（對應(yīng)每個Reduce任務(wù)產(chǎn)生一個輸出文件，文件名由用戶指定）。如果這些輸出不是最終的業(yè)務(wù)計算結(jié)果，則不需要將這R個輸出文件合并成一個文件，而是把這些文件作為另外一個MapReduce的輸入，或者在另外一個可以處理多個分割文件的分布式應(yīng)用中使用。

當?shù)贸鲎罱K的業(yè)務(wù)計算結(jié)果，可能需要將數(shù)據(jù)遷移到適合業(yè)務(wù)應(yīng)用訪問的存儲中，比如關(guān)系型數(shù)據(jù)庫或支持高并發(fā)、低響應(yīng)延遲的NOSQL數(shù)據(jù)庫中。

3 節(jié)點故障處理機制

作為一個分布式計算平臺必須要能很好地處理節(jié)點故障，不能因為某一個節(jié)點的故障而導(dǎo)致整個集群的計算任務(wù)失敗。對于任務(wù)本身因為代碼缺陷造成的執(zhí)行失敗，當任務(wù)執(zhí)行次數(shù)超過一定閾值后便不再執(zhí)行，不列為節(jié)點故障問題。

3.1 中心節(jié)點故障

中心節(jié)點分布式框架中存在worker和master兩類節(jié)點角色。

3.1.1 worker故障

master周期性地ping每個worker。如果在一個約定的時間范圍內(nèi)沒有收到worker返回的信息，master將把這個worker標記為失效。所有由這個失效的worker完成的map任務(wù)被重設(shè)為初始的空閑狀態(tài)，之后這些任務(wù)就可以被安排給其他的worker。同樣，worker失效時正在運行的map或reduce任務(wù)也將被重新置為空閑狀態(tài)，等待重新調(diào)度[2]。

當worker故障時，由于已經(jīng)完成的map任務(wù)的輸出存儲在這臺機器上，map任務(wù)的輸出已不可訪問了，因此必須重新執(zhí)行。而已經(jīng)完成的reduce任務(wù)的輸出存儲在全局文件系統(tǒng)上，因此不需要再次執(zhí)行。

當一個map任務(wù)首先被worker A執(zhí)行，之后由于worker A失效了，又被調(diào)度到worker B執(zhí)行，這個“重新執(zhí)行”的動作會通知給所有執(zhí)行reduce任務(wù)的worker。任何還沒有從worker A讀取數(shù)據(jù)的reduce任務(wù)將從worker B讀取數(shù)據(jù)。

MapReduce可以處理大規(guī)模worker失效的情況，MapReduce master只需要簡單地再次執(zhí)行那些不可訪問的worker的工作，之后繼續(xù)執(zhí)行未完成的任務(wù)，直到最終完成這個MapReduce操作。

3.1.2 mastre故障

一個簡單的解決辦法是，讓master周期性的將描述集群計算任務(wù)的數(shù)據(jù)結(jié)構(gòu)的寫入磁盤（位于集群以外的位置），即檢查點（checkpoint）。如果這個master任務(wù)失效了，可以從最后一個檢查點（checkpoint）開始啟動另一個master進程。然而，由于只有一個master進程，master失效后再恢復(fù)是比較麻煩的，因此我們現(xiàn)在的實現(xiàn)是，如果master失效就中止MapReduce運算。客戶可以檢查到這個狀態(tài)，并且可以根據(jù)需要重新執(zhí)行MapReduce操作[2]。

當用戶提供的map和reduce操作是輸入確定性函數(shù)（即相同的輸入產(chǎn)生相同的輸出）時，我們的分布式計算任務(wù)在任何情況下的輸出都和所有程序以正常的順序執(zhí)行所產(chǎn)生的輸出是一樣的。

3.2 省級分節(jié)點故障

基于全局分布式框架的計算模型，如果省級分節(jié)點發(fā)生故障，則該省的數(shù)據(jù)將缺失（主要是影響動態(tài)感知數(shù)據(jù)），和該省相關(guān)的數(shù)據(jù)計算任務(wù)都將失敗。但不會影響其他省與中心節(jié)點之間的業(yè)務(wù)，中心節(jié)點仍能保持絕大部分的業(yè)務(wù)服務(wù)能力。

省級分節(jié)點需要具備一定的冗災(zāi)機制，比如數(shù)據(jù)庫HA、數(shù)據(jù)備份，以應(yīng)對分節(jié)點故障，保證分節(jié)點的不間斷運行。同時在省級分節(jié)點和中心節(jié)點之間的網(wǎng)絡(luò)設(shè)施上也應(yīng)該有一定的冗余，以應(yīng)對突發(fā)事件。

4 應(yīng)用場景

以下論述船聯(lián)網(wǎng)兩級分布式計算框架的應(yīng)用場景，以及在各個場景下集群的計算擴展能力。

4.1 跨省船舶監(jiān)控

船聯(lián)網(wǎng)要求實現(xiàn)跨省船舶實現(xiàn)聯(lián)網(wǎng)監(jiān)控，比如A省船舶進入B省區(qū)域，B省應(yīng)能夠立即獲取到該船舶的基本信息和動態(tài)感知數(shù)據(jù)（如GPS）。

在船聯(lián)網(wǎng)分布式計算框架下，當有船舶跨省行駛，船舶所屬省的省級分節(jié)點將率先監(jiān)控到這一行為，分節(jié)點會查詢該船只的基本信息和簽證信息，然后通知中心節(jié)點，并開始向中心節(jié)點持續(xù)發(fā)送該船只的動態(tài)感知數(shù)據(jù)。中心節(jié)點得到信息后，更新該船只的狀態(tài)信息，然后聯(lián)系船舶進入的目的省所在的省級分節(jié)點，將船舶基本信息、動態(tài)數(shù)據(jù)也發(fā)送給目的省分節(jié)點。當船只返回原籍或駛?cè)肫渌》?，則中心節(jié)點變更推送的目的分節(jié)點。如此一來，跨省船舶的信息在中心節(jié)點和目的省都可以查詢和監(jiān)控，在保證集群線性擴展能力的前提下完成了跨省船舶的監(jiān)控跟蹤。

4.2 航運數(shù)據(jù)質(zhì)量分布式管控

智能航運首先需要按照一個統(tǒng)一的數(shù)據(jù)標準校驗其數(shù)據(jù)質(zhì)量，通過校驗的數(shù)據(jù)才能應(yīng)用于應(yīng)用服務(wù)。同時，船聯(lián)網(wǎng)也需要一套持續(xù)可行的數(shù)據(jù)質(zhì)量管理機制，保證后續(xù)數(shù)據(jù)處理的規(guī)范性。

通過智能航運數(shù)據(jù)處理的分布式彈性計算框架，可以采取分而治之的方式達到數(shù)據(jù)質(zhì)量的管控，同時也能保證整體架構(gòu)的擴展性。中心節(jié)點制定數(shù)據(jù)質(zhì)量管控的規(guī)范，下發(fā)數(shù)據(jù)標準到各個省級分節(jié)點。省級分節(jié)點按照數(shù)據(jù)標準校驗并清洗自己負責的部分，轉(zhuǎn)換為符合要求的數(shù)據(jù)結(jié)構(gòu)，然后將結(jié)果數(shù)據(jù)和數(shù)據(jù)質(zhì)量處理報告上報給中心節(jié)點。如果省級分節(jié)點持續(xù)增加，數(shù)據(jù)在達到一定規(guī)模限制后中心節(jié)點也可以借助自身的中心分布式框架進行計算擴展。

4.3 海量航運數(shù)據(jù)挖掘分析

隨著海量航運數(shù)據(jù)的集中，為挖掘數(shù)據(jù)的內(nèi)在價值，需要對這些業(yè)務(wù)數(shù)據(jù)進行統(tǒng)計分析和數(shù)據(jù)挖掘，如預(yù)測建模、關(guān)聯(lián)分析、數(shù)據(jù)聚類、異常檢測等。進行海量航運數(shù)據(jù)挖掘分析時，兩級分布式計算框架通過全局分布式保證了挖掘模型樣本的數(shù)據(jù)質(zhì)量，然后在中心分布式框架上完成挖掘分析計算。在中心分布式框架上可以結(jié)合mahout、R等挖掘分析軟件，更快速地完成分析模型的構(gòu)建。

5 結(jié)束語

兩級分布式計算平臺的設(shè)計體系可以在滿足當前智能航運數(shù)據(jù)處理要求的情況下，充分考慮到今后航運信息化發(fā)展中數(shù)據(jù)爆炸性增長時的計算擴展要求。雖然在初期，整個分布式計算平臺的建設(shè)需要一定的投入（涉及多個省市以及中心節(jié)點的建設(shè)），但由于具備良好的線性擴展能力，今后的集群擴展成本將非常可控：平臺不需要更換升級，只要不斷增加節(jié)點，即可以提供不斷增長的計算能力。隨著具體計算需求的落地和實現(xiàn)，智能航運分布式計算平臺將充分挖掘航運數(shù)據(jù)中的價值，為航運業(yè)務(wù)應(yīng)用提供更好的數(shù)據(jù)支撐和運算服務(wù)。

參考文獻：

[1] L?mmel R. Google's MapReduce programming model—Revisited[J].Science of computer programming，2008.70（1）： 1-30

[2] Dean J， Ghemawat S. MapReduce： simplified data processing on

large clusters[J]. Communications of the ACM，2008.51（1）：107-113

[3] Urbani J， Kotoulas S， Oren E， et al. Scalable distributed reasoning

using mapreduce[M]//The Semantic Web-ISWC 2009. Springer Berlin Heidelberg，2009：634-649

[4] 周開樂，丁帥，胡小建.面向海量數(shù)據(jù)應(yīng)用的物聯(lián)網(wǎng)信息服務(wù)系統(tǒng)研究綜述[J].計算機應(yīng)用研究，2012.29（1）.

[5] 趙學洋，李海紅，儲凌劍.基于船聯(lián)網(wǎng)的內(nèi)河智能航行體系探討研究[J].新技術(shù)新工藝，2013.6：117-121