基于分布式并行計算的鐵路電子支付平臺對賬業(yè)務數(shù)據(jù)處理方案研究

王 寧，郝雅青，甘人才

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

鐵路電子支付平臺（簡稱：支付平臺）承載著鐵路客、貨運輸?shù)碾娮又Ц逗蛯~結(jié)算業(yè)務。自2010年投入運行以來，隨著客、貨運業(yè)務量的增長和業(yè)務種類的增多，交易量從最初日均100萬筆提升至日均千萬筆，對賬處理效率較低的問題日漸凸顯。同時，隨著二維碼掃碼支付、鐵路e卡通等多種新型支付的增加，對賬業(yè)務和資金核算邏輯更為復雜，對支付平臺對賬業(yè)務與資金核算的靈活性和擴展性也提出更高要求。

為此，研究提出基于分布式并行計算架構(gòu)的對賬業(yè)務數(shù)據(jù)處理方案，對支付平臺的對賬與資金核算業(yè)務數(shù)據(jù)處理過程進行技術(shù)升級改造，將各接入渠道業(yè)務系統(tǒng)、支付機構(gòu)、支付平臺三方與對賬相關(guān)的數(shù)據(jù)均匯集到Hadoop平臺中，采用Spark、Kafka等組件搭建分布式并行計算環(huán)境，用于完成對賬和資金核算的數(shù)據(jù)處理任務；通過敏捷查詢引擎（AQE，Agile Query Engine），Hadoop數(shù)據(jù)可以通過JDBC和REST等多種方式，提供給電子支付管理平臺，支持后續(xù)的交易查詢、報表統(tǒng)計、偏差處理等業(yè)務；并在實驗室搭建測試環(huán)境，對該數(shù)據(jù)處理方案進行測試驗證。

1 鐵路電子支付平臺對賬業(yè)務及數(shù)據(jù)處理要求

1.1 對賬和資金核算業(yè)務流程

支付平臺對賬和資金核算功能主要完成各類交易信息的匯總、核對、差異處理、資金報表生成等業(yè)務處理[1]，主要業(yè)務流程如圖1所示。

圖1 支付平臺對賬與資金核算業(yè)務流程示意

1.1.1 接入渠道

接入渠道是指為鐵路客戶提供客貨業(yè)務服務的各類業(yè)務系統(tǒng)，如鐵路12306互聯(lián)網(wǎng)售票系統(tǒng)（簡稱：12306）、鐵路貨運電子商務系統(tǒng)（簡稱：貨運系統(tǒng)）；各接入渠道業(yè)務系統(tǒng)在執(zhí)行交易時，可通過接口將交易信息發(fā)送至支付平臺。

1.1.2 支付機構(gòu)

鐵路客戶在辦理鐵路客、貨運業(yè)務時，可選用不同支付機構(gòu)的支付服務，各個支付機構(gòu)通過接口向支付平臺提供交易接入服務。

1.1.3 支付平臺

支付平臺主要完成交易處理、對賬、交易查詢、偏差處理、報表統(tǒng)計等功能，對賬業(yè)務主要流程如下：

（1）對賬文件入庫：各接入渠道業(yè)務系統(tǒng)在約定的時間將固定格式的對賬文件傳輸至支付平臺的文件服務器，文件服務器上的程序自動檢查對賬文件格式和數(shù)據(jù)的正確性，將正確的對賬文件數(shù)據(jù)寫入數(shù)據(jù)庫；若有對賬文件存在異常，則給出提示，以轉(zhuǎn)由相關(guān)業(yè)務人員進行人工處理；

（2）對賬處理：支付平臺每日自動執(zhí)行對賬處理，按照設定的規(guī)則，將各接入渠道業(yè)務系統(tǒng)的數(shù)據(jù)與支付平臺的數(shù)據(jù)、各支付機構(gòu)的數(shù)據(jù)與支付平臺的數(shù)據(jù)，分別進行逐筆交易兩兩對比核查；核對的數(shù)據(jù)項主要包括交易流水號、交易金額、交易狀態(tài)、交易時間等；

（3）對賬結(jié)果處理：對賬結(jié)果分為對賬成功、單邊交易賬、金額不一致等類型；對于單邊交易賬和金額不一致的交易數(shù)據(jù)，按系統(tǒng)設置的審核規(guī)則進行數(shù)據(jù)偏差審核，判定是否需要給客戶退款；對于無法自動完成審核的交易數(shù)據(jù)，由業(yè)務人員手工處理[2]；

（4）資金核算：資金核算處理程序每月對各接入渠道業(yè)務系統(tǒng)的當月結(jié)賬資金、支付平臺記錄的當月資金、支付機構(gòu)的當月資金進行核對，資金核算結(jié)果包括月切交易數(shù)據(jù)、單邊交易數(shù)據(jù)、交易類型對比核查；通過資金核對，找出存在資金差異的交易數(shù)不一致數(shù)據(jù)、金額不一致數(shù)據(jù)等類型；業(yè)務人員將依據(jù)資金核算結(jié)果，進行欠款追款、調(diào)賬等多種處理；

（5）報表統(tǒng)計與資金上繳：根據(jù)交易處理、對賬、資金核算分類處理，支付平臺統(tǒng)計生成業(yè)務數(shù)據(jù)匯總、銀行數(shù)據(jù)匯總、平臺數(shù)據(jù)匯總、電子支付日報、資金差異明細表等多種業(yè)務報表和資金報表，業(yè)務人員依據(jù)相關(guān)報表進行資金上繳[3]。

1.2 對賬業(yè)務數(shù)據(jù)處理要求

（1）數(shù)據(jù)處理量及性能要求

支付平臺需要對賬的交易數(shù)據(jù)來自多個接入渠道業(yè)務系統(tǒng)和10多家支付機構(gòu)。對賬業(yè)務處理是逐日滾動執(zhí)行的，考慮到接入支付平臺的不同來源交易數(shù)據(jù)可能存在一定時間延遲，為保證每日數(shù)據(jù)核對的有效性，每日執(zhí)行對賬處理時一般對連續(xù)5天（即T-2、T-1、T、T+1、T+2）的交易數(shù)據(jù)進行核對；目前，每日對賬處理的交易數(shù)據(jù)記錄約為2億條，對賬處理全流程耗時應少于2 h。

（2）可擴展性

隨著支付平臺可支持的支付方式日益多樣化，以及所接入的支付機構(gòu)和各接入渠道業(yè)務系統(tǒng)的不斷增加，對賬處理應能快速、靈活地適應數(shù)據(jù)源增加和接口類型增多。

2 基于分布式并行計算架構(gòu)的對賬業(yè)務數(shù)據(jù)處理方案

基于分布式并行計算架構(gòu)的對賬業(yè)務數(shù)據(jù)處理方案的邏輯框架如圖2所示。

圖2 對賬業(yè)務數(shù)據(jù)處理方案的邏輯框架示意

支付平臺對賬業(yè)務的數(shù)據(jù)處理主要包括：數(shù)據(jù)采集模塊、對賬模塊、數(shù)據(jù)存儲模塊、數(shù)據(jù)查詢模塊等功能模塊，以及運行監(jiān)控和任務調(diào)度等輔助管理模塊。

2.1 數(shù)據(jù)采集模塊

各接入渠道業(yè)務系統(tǒng)的對賬文件，經(jīng)解析代碼處理為結(jié)構(gòu)化數(shù)據(jù)，并存儲到高性能消息中間件Kafka；采用Spark Streaming組件定期從Kafka拉取數(shù)據(jù)，然后存儲在分布式文件系統(tǒng)HDFS中，這些數(shù)據(jù)為待對賬數(shù)據(jù)；增量數(shù)據(jù)采用Hudi組件，實現(xiàn)數(shù)據(jù)增量更新、數(shù)據(jù)版本管理和數(shù)據(jù)痕跡追蹤。

2.2 數(shù)據(jù)存儲模塊

待對賬數(shù)據(jù)和對賬結(jié)果數(shù)據(jù)均存儲在分布式文件系統(tǒng)HDFS中，采用Hudi組件可構(gòu)建和管理PB級數(shù)據(jù)，為各類業(yè)務提供高效和低延遲的數(shù)據(jù)連接，支持文件級、記錄級的插入、刪除、更新操作，可按時間版本查詢數(shù)據(jù)，有效地改善存儲管理和查詢性能。

對賬業(yè)務數(shù)據(jù)處理按日進行，各接入渠道業(yè)務數(shù)據(jù)、支付平臺交易數(shù)據(jù)、各支付機構(gòu)數(shù)據(jù)均按日期分區(qū)存儲，并進行數(shù)據(jù)分片。

2.3 對賬模塊

對賬邏輯程序采用Spark 批處理完成，Spark 運行在Hadoop Yarn上，由Yarn管理Spark集群，負責資源統(tǒng)一管理，任務調(diào)度與監(jiān)控。

為充分利用Spark的并行計算能力，按照處理數(shù)據(jù)分片規(guī)則，將當天對賬數(shù)據(jù)進行分類后，再行執(zhí)行對賬邏輯計算，可顯著提高計算效率。

2.4 數(shù)據(jù)查詢模塊

AQE作為數(shù)據(jù)查詢核心組件，將存儲在HDFS中的數(shù)據(jù)暴露為JDBC接口或者REST接口，為其它業(yè)務系統(tǒng)提供數(shù)據(jù)查詢服務。

2.5 任務調(diào)度

為了便于日常運行維護，采用Airflow平臺完成任務調(diào)度和監(jiān)控；每日定時運行對賬處理邏輯代碼，可查看任務執(zhí)行歷史，對任務執(zhí)行異常進行提示與警告。

2.6 監(jiān)控模塊

監(jiān)控模塊主要有3類監(jiān)控對象：服務器和組件、對賬任務、對賬數(shù)據(jù)。

（1）服務器和組件：實時監(jiān)控虛擬機的CPU、內(nèi)存等使用情況；采集服務器級別的告警信息；實時監(jiān)控Kafka、Spark、Hadoop、Airflow 等組件的運行情況。

（2）對賬任務監(jiān)控：監(jiān)控Kafka寫入Hadoop的實時流任務和對賬任務的運行情況；監(jiān)控定時任務是否成功執(zhí)行。

（3）對賬數(shù)據(jù)監(jiān)控：監(jiān)控Kafka寫入Hadoop的數(shù)據(jù)量、監(jiān)控所采集的數(shù)據(jù)量；監(jiān)控當日各接入渠道數(shù)據(jù)同步的完成情況；監(jiān)控對賬處理完成情況、對賬相符數(shù)據(jù)量、對賬差異數(shù)據(jù)量、退款對賬數(shù)據(jù)量等業(yè)務數(shù)據(jù)。

3 關(guān)鍵技術(shù)

3.1 分布式數(shù)據(jù)計算

分布式通用數(shù)據(jù)計算引擎Spark是專為大規(guī)模數(shù)據(jù)快速處理而設計的通用計算引擎[4]，基于類Hadoop MapReduce的開源通用并行框架，但不同于MapReduce的是，計算任務的中間結(jié)果可保存于內(nèi)存中，無需讀寫HDFS。Spark能夠提供交互式數(shù)據(jù)處理，還可以優(yōu)化迭代工作負載，適用于數(shù)據(jù)挖掘與機器學習等需要迭代的MapReduce算法，常用于構(gòu)建大型的低延遲數(shù)據(jù)處理應用[5]。

支付平臺對賬處理采用Spark計算框架，將對賬數(shù)據(jù)加載到內(nèi)存中，并采用Spark分片和多任務并行計算方式，能夠極大地提升對賬處理效率。

3.2 分布式文件系統(tǒng)HDFS

分布式文件系統(tǒng)HDFS基于流數(shù)據(jù)模式，可運行于廉價服務器[6]，具有高容錯、高可靠性、高可擴展性、高可用性、高吞吐率等特性[7]，且安裝和維護簡單。采用分布式文件系統(tǒng)HDFS存儲和管理海量結(jié)構(gòu)化分析型數(shù)據(jù)，能夠以較低的成本實現(xiàn)安全、可靠的數(shù)據(jù)存儲，并保證數(shù)據(jù)存儲規(guī)模具有良好的持續(xù)擴展性。

采用分布式文件系統(tǒng)HDFS存儲支付平臺的對賬業(yè)務數(shù)據(jù)，解決了關(guān)系型數(shù)據(jù)庫因讀寫磁盤速率低造成的數(shù)據(jù)讀寫速度慢的問題，提高了數(shù)據(jù)處理吞吐率。HDFS提供的數(shù)據(jù)分片存儲方式，也為支持未來業(yè)務數(shù)據(jù)量的持續(xù)增長提供保障。

3.3 分布式消息中間件

Kafka是一種高吞吐量的分布式發(fā)布訂閱消息系統(tǒng)，可提供高吞吐、低延遲的實時數(shù)據(jù)處理平臺[8]。其持久化層是一個按照分布式事務架構(gòu)的大規(guī)模發(fā)布/訂閱消息隊列，常作為企業(yè)消息總線、實時數(shù)據(jù)管道，主要起到削峰填谷、系統(tǒng)解耦以及冗余的作用。

采用Kafka消息中間件實現(xiàn)支付平臺對賬業(yè)務的數(shù)據(jù)采集，提升了數(shù)據(jù)采集環(huán)節(jié)的效率和彈性，可為數(shù)據(jù)處理提供緩沖池，以避免對賬業(yè)務數(shù)據(jù)處理程中各環(huán)節(jié)間互相影響。

3.4 分布式查詢引擎

AQE是基于Apache Arrows內(nèi)存列式存儲的大數(shù)據(jù)查詢引擎，采用MPP技術(shù)，能接入HDFS、Hive、Kafka、RDBMS等多種外部數(shù)據(jù)源，支持JDBC、ODBC、Rest的查詢接口和基于Arrow Flight的查詢API接口，支持異構(gòu)數(shù)據(jù)源的混合查詢計算[9]。AQE使用物化視圖和SQL重寫技術(shù)，實現(xiàn)查詢性能優(yōu)化和提升。

采用AQE查詢引擎實現(xiàn)支付平臺對賬業(yè)務數(shù)據(jù)處理的查詢功能，可將多種數(shù)據(jù)源的訪問進行統(tǒng)一封裝和優(yōu)化，為前端業(yè)務功能提供統(tǒng)一、高效的查詢接口。

4 測試及驗證

4.1 測試環(huán)境

為驗證基于分布式并行計算架構(gòu)的對賬業(yè)務數(shù)據(jù)處理方案的有效性，在實驗室搭建對賬業(yè)務數(shù)據(jù)處理的測試環(huán)境，如圖3所示。

圖3 對賬業(yè)務數(shù)據(jù)處理的測試環(huán)境構(gòu)成示意

組件監(jiān)控：由2臺服務器部署相應監(jiān)控程序組成，負責監(jiān)控測試環(huán)境中相關(guān)資源的運行情況。

數(shù)據(jù)采集和對賬處理：消息隊列Kafka功能相對獨立，由2臺服務器組成；并行計算Spark、系統(tǒng)管理Yarn、文件系統(tǒng)Hadoop、數(shù)據(jù)存儲Hive、任務調(diào)度Airflow 等組件部署在4臺服務器上，構(gòu)成對賬數(shù)據(jù)處理運行環(huán)境。

數(shù)據(jù)查詢：AQE部署在2臺服務器上，通過外部接口調(diào)用數(shù)據(jù)存儲完成數(shù)據(jù)查詢。

4.2 主要測試項目

（1）對賬數(shù)據(jù)采集：將現(xiàn)有數(shù)據(jù)轉(zhuǎn)儲至大數(shù)據(jù)平臺中，包括各接入渠道、支付機構(gòu)和支付平臺的交易數(shù)據(jù)，為后續(xù)對賬處理準備好測試數(shù)據(jù)集。

（2）對賬處理：采用Spark多任務并行計算進行對賬處理，將對賬結(jié)果存放在Hive庫中。

（3）對賬結(jié)果查詢：通過AQE提供的JDBC接口和REST接口，按多種條件快速、靈活的查詢對賬數(shù)據(jù)。

4.3 測試結(jié)果比對

采用分布式并行計算架構(gòu)后，相較于原來基于Oracle關(guān)系型數(shù)據(jù)庫存儲的運行環(huán)境，技術(shù)升級改造前后對賬業(yè)務數(shù)據(jù)處理測試項目指標對比見表1。

表1 技術(shù)升級改造前后測試項目指標對比

（1）數(shù)據(jù)采集：采用Spark和Kafka相結(jié)合的數(shù)據(jù)采集處理方式，速度提高近4.5倍。此外，相對于現(xiàn)有系統(tǒng)完全基于數(shù)據(jù)庫的處理方式，采用Spark和Kafka組件采集數(shù)據(jù)，還能減輕數(shù)據(jù)庫負載，測試過程中數(shù)據(jù)庫服務器CPU監(jiān)測指標表明，可釋放出約30%資源能力。

（2）對賬計算：采用Spark并行任務進行基于內(nèi)存的分布式計算，數(shù)據(jù)處理速度提升達十倍量級。

（3）對賬結(jié)果查詢：在現(xiàn)有Oracle關(guān)系型數(shù)據(jù)庫中，采用按時間分區(qū)存儲方案，查詢跨分區(qū)數(shù)據(jù)時，需進行全表掃描，數(shù)據(jù)讀取耗時較長，一般查詢響應時間約15 s?；贏QE進行查詢時，可自動根據(jù)分區(qū)條件確定數(shù)據(jù)查詢范圍，無需全表掃描，平均查詢響應時間約300 ms；且數(shù)據(jù)量越大，查詢速度差距越明顯。另外，Oracle關(guān)系型數(shù)據(jù)庫僅提供基于JDBC接口的SQL查詢方式，而AQE查詢可提供多種類型的數(shù)據(jù)源訪問接口，還可提供REST接口，能夠支持更為靈活的查詢需求。

5 結(jié)束語

針對鐵路電子支付平臺對賬處理面臨的問題，分析現(xiàn)有對賬業(yè)務數(shù)據(jù)處理流程及性能要求，研究分布式并行計算相關(guān)的關(guān)鍵技術(shù)，提出基于Hadoop、Spark等技術(shù)的分布式并行計算方案；設計了數(shù)據(jù)采集、對賬、數(shù)據(jù)查詢等業(yè)務處理模塊，以及數(shù)據(jù)存儲、任務調(diào)度和運行監(jiān)控等輔助管理模塊；在實驗室搭建測試環(huán)境，對測試數(shù)據(jù)集進行處理。測試表明：該方案可顯著提高對賬核算業(yè)務處理效率，增強支付平臺對業(yè)務需求靈活支撐的能力。

該方案中所使用的相關(guān)組件多為開源技術(shù)，若要投入實際運用，如何有效確保其穩(wěn)定可靠運行，需要盡一步開展深入研究。