智能航運數據處理的分布式彈性計算技術研究與應用

肖榕 金甌 葉建鋒

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

關鍵詞： 內河航運； 彈性計算技術； 兩級分布式； 數據處理

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

0 引言

內河航運是我國綜合運輸體系的重要組成部分，在我國經濟發展中起著非常重要的作用。針對航運市場發展的總體態勢，發揮信息化對水路運輸行業現代化發展的支撐和引領作用，以信息資源共享為基礎，以信息服務為核心，以現代物聯網技術、通信技術和信息技術為支撐，以促進水路交通運輸產業和物聯網產業發展為目標，進行智能航運信息服務應用的研究。

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

1 平臺架構

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

船聯網全局分布式框架廣域部署，通過中心節點調度各省資源，支撐船聯網省級航運業務應用，如區域船舶監控、應急指揮等。中心分布式框架在中心節點局域部署，通過中心主節點調度中心分節點資源，支撐項目全局性的業務應用，如綜合監管、統計分析等。

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

2 計算模型

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

船聯網全局分布式框架的計算模型如下。

⑴ 各省級分節點處理各省航運數據，包括數據的整合、清洗、轉換等，即分節點可以處理的計算任務直接在分節點完成。

⑵ 中心節點整合各個分節點上報的數據，包括船舶船員基本信息、船舶簽證信息、動態感知信息等。

⑶ 船舶船員基本信息、編碼等靜態數據由中心節點輪詢各分節點采集；船舶簽證信息、感知信息等動態數據由分節點主動推送。

⑷ 各分節點只與中心節點聯系，分節點之間不直接交換數據。

2.1.1 省級分節點的數據處理

省級分節點負責從各省航運業務系統采集數據，完成省一級的數據整合和清洗，同時可以支撐省一級的航運應用。

例如，A省有一GPS系統采集船舶位置信息，則A省分節點負責從該GPS系統采集船舶位置信息并存儲，然后完成數據的校驗清洗，再按要求將該數據上傳中心節點。A省如有另一業務系統只需要本省船舶GPS信息，則直接從A省分節點獲取，不需要向船聯網中心節點發起請求。

2.1.2 中心節點的數據處理

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

完成整合后的數據在中心節點落地，交由中心節點作進一步的分析處理。船聯網在整合數據的基礎上提供全局性的數據服務支持，比如跨區域的數據服務。

2.1.3 節點間的數據傳輸機制

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

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

船聯網中心節點分布式框架計算模型使用目前較為成熟的MapReduce計算模型。目前世界上最快的1TB排序記錄就是由基于MapReduce實現的。

MapReduce將計算任務劃分為map和reduce兩個階段。map階段負責“分”，即把復雜的任務分解為若干個“簡單的任務”執行?！昂唵蔚娜蝿铡庇幸韵聨讉€含義：

⑴ 數據或計算規模相對于原任務要大大縮?。?/p>

⑵ 就近計算，即任務會被分配到存放了所需數據的節點進行計算；

⑶ 這些小任務可以并行計算，彼此間幾乎沒有依賴關系。

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

2.2.1 MapReduce執行機制[1-3]

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

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

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

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

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

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

⑹ reduce worker程序遍歷排序后的中間數據，對于每一個惟一的中間key值，reduce worker程序將這個key值和它相關的中間value值的集合傳遞給用戶自定義的reduce函數。reduce函數的輸出被追加到所屬分區的輸出文件。

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

2.2.2 結果數據的處理

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

當得出最終的業務計算結果，可能需要將數據遷移到適合業務應用訪問的存儲中，比如關系型數據庫或支持高并發、低響應延遲的NOSQL數據庫中。

3 節點故障處理機制

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

3.1 中心節點故障

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

3.1.1 worker故障

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

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

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

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

3.1.2 mastre故障

一個簡單的解決辦法是，讓master周期性的將描述集群計算任務的數據結構的寫入磁盤（位于集群以外的位置），即檢查點（checkpoint）。如果這個master任務失效了，可以從最后一個檢查點（checkpoint）開始啟動另一個master進程。然而，由于只有一個master進程，master失效后再恢復是比較麻煩的，因此我們現在的實現是，如果master失效就中止MapReduce運算?？蛻艨梢詸z查到這個狀態，并且可以根據需要重新執行MapReduce操作[2]。

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

3.2 省級分節點故障

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

省級分節點需要具備一定的冗災機制，比如數據庫HA、數據備份，以應對分節點故障，保證分節點的不間斷運行。同時在省級分節點和中心節點之間的網絡設施上也應該有一定的冗余，以應對突發事件。

4 應用場景

以下論述船聯網兩級分布式計算框架的應用場景，以及在各個場景下集群的計算擴展能力。

4.1 跨省船舶監控

船聯網要求實現跨省船舶實現聯網監控，比如A省船舶進入B省區域，B省應能夠立即獲取到該船舶的基本信息和動態感知數據（如GPS）。

在船聯網分布式計算框架下，當有船舶跨省行駛，船舶所屬省的省級分節點將率先監控到這一行為，分節點會查詢該船只的基本信息和簽證信息，然后通知中心節點，并開始向中心節點持續發送該船只的動態感知數據。中心節點得到信息后，更新該船只的狀態信息，然后聯系船舶進入的目的省所在的省級分節點，將船舶基本信息、動態數據也發送給目的省分節點。當船只返回原籍或駛入其他省份，則中心節點變更推送的目的分節點。如此一來，跨省船舶的信息在中心節點和目的省都可以查詢和監控，在保證集群線性擴展能力的前提下完成了跨省船舶的監控跟蹤。

4.2 航運數據質量分布式管控

智能航運首先需要按照一個統一的數據標準校驗其數據質量，通過校驗的數據才能應用于應用服務。同時，船聯網也需要一套持續可行的數據質量管理機制，保證后續數據處理的規范性。

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

4.3 海量航運數據挖掘分析

隨著海量航運數據的集中，為挖掘數據的內在價值，需要對這些業務數據進行統計分析和數據挖掘，如預測建模、關聯分析、數據聚類、異常檢測等。進行海量航運數據挖掘分析時，兩級分布式計算框架通過全局分布式保證了挖掘模型樣本的數據質量，然后在中心分布式框架上完成挖掘分析計算。在中心分布式框架上可以結合mahout、R等挖掘分析軟件，更快速地完成分析模型的構建。

5 結束語

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

參考文獻：

[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] 周開樂，丁帥，胡小建.面向海量數據應用的物聯網信息服務系統研究綜述[J].計算機應用研究，2012.29（1）.

[5] 趙學洋，李海紅，儲凌劍.基于船聯網的內河智能航行體系探討研究[J].新技術新工藝，2013.6：117-121