基于混合式架構城市軌道交通大數據平臺與運營決策應用

張 銘

(中國鐵道科學研究院集團有限公司電子計算技術研究所 北京 100081)

0 引 言

近年大城市軌道交通快速進入規模化和網絡化格局，為日常生產調度、運營、管理服務的各類信息系統快速積累了大量的數據資源，陸續建設了數據中心或信息中心，對大數據技術和數據資源效益的挖潛日益重視[1-4]。目前城市軌道交通部分企業采用了傳統數據倉庫一體機的模式，以北京、廣州、深圳為代表，圍繞線網運營業務建立數據結構和主題域，該模式以數據資源的存儲為目標。但隨著硬件設備、軟件版本的更新換代，須同步更新歷史數據管理方式，且難以適應于不斷變化和彈性擴展的應用需求，投資成本高，逐步顯現出它的劣勢。隨著大數據和云計算技術的探索應用，文獻[5-6]總結了基于Hadoop各項大數據技術及應用創新。池亞平等[7]研究了大數據平臺海量監控數據的存儲效率改進方法。李敏等[8]就物聯網環境下的數據時空特性提出了異構大數據處理的時效性改進模型，這些為大數據平臺的建設和應用優化提供了借鑒。一些新建和規劃數據中心的軌道交通企業，如上海、南京、重慶，采用了新型的大數據平臺技術，如基于MPP(Massively Parallel Processor)架構、基于虛擬化共享硬件的方式，基于Hadoop大數據平臺模式，均不同程度地兼顧了遠期硬件擴容的投入，以及數據挖掘、商務智能方面應用的靈活擴展性。

針對城市軌道交通數據資源在業務、生產、經營、管理、服務方面的差異性，將數據運用于不同應用場景的需求，本文提出建立基于混合式架構的大數據平臺模式，以可擴展、自適應、可定制為目標導向，深化平臺的系統架構、數據架構、分層管控機制，并在運營管理與決策支持中加以運用，為運營信息化基礎大數據平臺提供一套實用的綜合解決方案。

1 運營信息化業務數據特征分析

1.1 數據類型多樣與時空特性

城市軌道交通運營信息化類系統作為企業數據資產的基礎數據源，根據業務歸屬和關聯關系劃分為生產類、管理類、業務類、線網級管理與基礎平臺類四種類型。具體如下：

1) 生產信息化系統，包括信號ATS、通信、綜合監控ISCS、電力PSCADA、火災報警FAS、環控BAS、自動售檢票AFC、清分中心ACC等系統[9]，主要服務于日常的生產調度與安全保障。

2) 辦公管理信息化系統，包括辦公自動化、人力資源管理、財務管理、合同管理、檔案管理等系統。

3) 運營業務信息化系統，包括票務管理、站務管理、乘務管理、資產管理、運營計劃管理、施工調度、維修管理等系統。

4) 線網級管理與基礎平臺系統，包括線網指揮中心、線網應急處置、信息發布等，以及統一認證、門戶網站、BIM系統等。

產生的數據類型包括結構化數據、非結構化數據、半結構化數據，例如存儲于關系型數據庫的數據表，視頻監控的多媒體數據，接口傳輸的文件類數據等，需要配置不同的存儲模式和空間與之相適應。實時數據存儲時間要求不一，如歷史報警、突發事件等數據需要長期保存，而常規視頻數據一般存儲15～30日，接口文件一般存儲0.5～1年。數據的時空差異性決定了統一數據資源管理需要因地制宜，結合業務和數據本身的特性制定策略。

1.2 數據處理的復雜性

根據各類業務系統數據的時效性差異，數據采集時間和方式也不同，包括實時數據(采集周期1 s以內)、近線數據(5 min以內)、離線數據(小時、日、周不等)、隨機數據(一般不固定時間，數據產生時自行抽取或推送)。接口方式包括專用的接口協議或通信機制，如Modbus TCP/IP協議，文件消息隊列(Message Queue)方式、FTP文件傳輸、報文、Web service等，實現用于數據表交換的監控數據靜態點、狀態點、監測點、控制點，及客流文件、多媒體數據等的獲取，再根據業務特點統一清洗、抽取、加載為可共享的數據形式存儲。

1.3 應用面廣與數據多重交互

共享數據資源經過整合后為各種業務系統調用，同時各系統之間存在數據交互，如調度監控系統為運營管理和應急處置系統提供設備報警的數據，及時預防災害或處理大型故障；資產管理系統為物資管理、維修管理系統提供統一的編碼；統一認證和門戶系統為各類業務系統提供統一的企業用戶名錄和角色保障數據同步更新。分析性數據為指標統計、效益核算、成本分析、線路規劃及建設后評估等提供決策依據。

2 運營信息大數據平臺的構建

2.1 混合式架構的數據平臺框架

平臺定位于建立面向多用戶的信息集中共享、資源高效利用的軌道交通數據服務和綜合信息管理平臺，實現統一采集、長期存儲、統計分析、數據挖掘的功能。根據各種業務的數據共享和使用需求，首先搭建定制化的基礎數據平臺，基于數據倉庫MPP架構、大數據Hadoop架構及關系型數據結構混合的數據管理模式，實現分級分類的數據共享。構建總體規劃的分層體系，即數據采集層、數據融合層、技術組件層、分析模型層、應用訪問層的五層系統架構。如圖1所示。

圖1 運營信息大數據平臺的層次架構

(1) 數據采集與融合層 完成各類系統基礎數據和設備級系統數據的接入，包括與車站層、線路層、車輛段、控制中心等系統的接口，與外部信息的邊界安全防護，保障實時、非實時數據的采集和處理。按照實時庫、非實時的業務主題數據、公共數據對象抽取，并根據結構化、非結構化數據類型區分存儲周期和更新機制。

(2) 技術組件與分析模型層 將通用的組件和工具形成資源模型庫，將工作流、接口標準化、微服務、流計算、數據發布、單元建模、畫面組件等通用組件集以調用模式統一管理，并封裝便捷的調用接口。將單元業務應用包劃分為最小單元，進行多維度分析和業務流程建模，濃縮和匯聚主要的業務單元，成為信息化系統調用的最直接資源庫。

(3) 業務應用訪問層 作為日常生產、管理、經營、企業資源利用等方面數據服務的系統工具，實現信息流轉和辦公支撐，構建標準、安全、高效的信息共享模式。面向車站、線路、線網各級別的系統，覆蓋城市軌道交通數據資源需求的各個業務面。

2.2 數據流規劃與分級存儲

數據處理系統作為大數據平臺的核心，通過對業務數據的有效集成，滿足對海量數據的存儲、管理和使用需要，需根據業務類型進行數據流規劃與存儲機制的設計，如圖2所示。

圖2 混合式數據管理模式及數據流

數據資源歸納為以下四種類型：

(1) 實時數據庫 用于采集實時數據，設計快速提取和計算服務模塊，將獲取的實時數據標準化處理后轉入實時監控的可視化展示，如設備狀態實時監控。

(2) 關系型數據庫 設置一定存儲空間的關系型數據庫，用于多元業務流程及可擴展應用的數據管理和事務型分析的應用，如資產、維修、調度管理等。

(3) 基于MPP架構的數據倉庫 設置基于主題域和元數據管理的數據倉庫，圍繞定向型業務定義數據結構，如歷史報警綜合查詢、多維統計、客流預測、仿真回放，并統一進行數據質量、數據治理、數據模型、數據維護管理。利用數據倉庫的數據分析能力支持運營效果評估和運營仿真應用。

(4) 基于Hadoop架構的大數據應用庫 將所有采集的數據匯總存入，非結構化數據獨立管理，并定義索引用于智能檢索。采用分布式Kafka消息隊列的方式采集數據后，定制可擴展的存儲邏輯單元管理，為長期數據存儲擴容提供條件，運用處理、查詢、讀取的各類工具集管理數據，為業務系統提供數據服務，如客流預測、云管理等。

2.3 數據采集與計算服務標準化

傳統的數據共享方式，經過數據采集后直接存儲，為業務系統提供數據時呈現出松散、繁雜的原始特征。由于城市軌道交通專業眾多，數據內容和類型多樣，所以數據經采集后處理的方式對數據分發的時效性、根據時間片存儲數據的同步性和一致性等都會產生影響。而接口采集的數據一般都是現場設備狀態及系統運行數據，實際業務需要呈現的數據，往往要根據數據形式進行數學運算和邏輯判斷。未經過以業務需求為出發點處理的數據，使用的效率低，而為了單一業務需求涉及多個數據表的訪問和關聯，對系統的負載和數據讀取的效率均產生不利影響。

針對這一問題，為大數據平臺的數據采集模塊設計基于數據實時計算的快速標準化的數據處理方法。數據采集后即做快速處理，以最小通用單元的形式，為業務系統提供便捷、可直接應用的數據單元。通過設定統一的數據處理模板，將單元數據的業務需求歸納并加以屏蔽，經過實時計算和解析為標準化的基礎數據單元，可大幅減輕在大規模數據接入后處理過程的繁雜性，提高數據的應用效率。并進一步引入統一的數據模型、制定報警處理規則等方式，將不同專業系統的業務流程歸一化。從實時數據接入后進行標準化處理過程規劃、實時計算處理原則、功能、性能保障角度對數據處理服務設計，確保數據的業務適用性、實效性和可靠性。主要包括三個階段：數據提取，數據計算及服務、自動分析與報警服務。

2.3.1 數據提取服務

數據提取模塊包括：專業數據、時間數據提取。從消息中間件獲取數據采集層提供的實時數據，如設備狀態、客流監測數據，經數據提取過濾的時間數據，根據表達式進行實時計算，并將計算結果推送給報警管理業務功能和實時可視化監控模塊，嚴格區分各自接收的專項數據內容。

數據采集層定義Producer(接口甲方)，計算服務模塊定義為Consumer(接口乙方)。如果有請求信息，數據采集層接收則為Consumer(接口甲方)，計算服務模塊則為Producer(接口乙方)。例如監控系統的數據載入按每條地鐵線路創建二個通道(Topic)，為多副本單分區形式。以客流數據計算服務為例說明計算方法如下。

(1) 初始化定義。定義過程(DatalogicWorker)：周期性地向數據計算服務模塊提供客流數據，包括：進出站客流、斷面客流、各類客流數據排名、各類客流數據報警狀態。流程如下：

定義數據提取任務(Job)：每個數據提取任務實現不同數據的提取，定義過程(PassengerFlowLogicJob)，該任務用于提取進出站客流、區間客流。具體業務邏輯在private static void GetPassengerFlowData()方法中實現。計算邏輯如下：

var pfstmeList=PassengerFlowDataCollection.GetInstance().GetPassengerFlowStation();

foreach(var pfsm in pfstmeList)

{

PfstmeMaps.Add(ConvetTools.LongTryParse(pfsm.Stationid.Trim()+pfsm.Flowtype), pfsm);

//作為字典Key，標示客流狀態

PfstmeMaps.Add(ConvetTools.LongTryParse("10000"+pfsm.Stationid.Trim()+pfsm.Flowtype),pfsm);

}

(2) 將數據提取服務(ScheduleService)注冊為系統服務。

(3) 定義數據提取服務的啟動、停止方法，并在啟動方法中初始化所有數據提取任務。

定義過程(Periodic)：周期性的向數據計算服務提供所需的時間數據，包括：當前時間(到s級，每隔10 s更新一次)、當日運營時間(到min級，每日更新一次)。

首先定義時間數據提取與發送機制：

private void StartEtlAction(SqlEtlInfo sqlEtlInfo, string timeRequest)

{

var config=sqlEtlInfo.DatabaseTypeConnQuery;

if(config.Trim().Length<1) return;

var sqlconfig=JsonHelper.DeserializeJsonToList(config);

var sqlTimeRequest=sqlconfig.Where(s=>s.TimeRequest.ToLower()==timeRequest.ToLower());

foreach(var item in sqlTimeRequest)

{

var type=GetDbType(item.DatabaseType);

try

{

using(var db=SugarDao.GetInstance(type, item.DatabaseConn))

{

var datast=db.SqlQuery(item.DatabaseQuery).ToList();

//System.Diagnostics.Debug.WriteLine(string.Join(";",

//datast));

var sendData=string.Join(";", datast);

ServiceProvider.GetService().

ActiveMqWorker.Send(sendData);

log.Info(＄"Send to MQ:{sendData}");

}

}

catch(Exception ex)

{

ExceptionHelper.HandleException(ex);

}

}

}

再經初始化，將時間數據提取服務添加到服務容器。

2.3.2 數據計算服務

將各類數據調取需求歸納為統一、可識別的表達式模板，轉換為數據表定義，對應不同的列形成機器識別的數據類目，并進一步對應到數據計算過程中。

定義過程(Bridge)：接收數據提取服務模塊發送的數據，實時計算相關的表達式，并將計算結果推送至消息中間件或報警管理服務。流程如下：

(1) 服務啟動入口；

(2) 將服務添加到服務容器內；

(3) 啟動數據計算服務；

(4) 數據分發。

具體計算過程包括：

初始化：

public void InitialzeResultExpDict(DbType dbType, string connStr, BlockingCollection