面向外匯市場監測的分布式計算框架設計

程文亮，王志宏，周 虞，過 弋,3,4，趙俊鋒

(1.中匯信息技術(上海)有限公司 開發二部，上海 201203； 2.華東理工大學 信息科學與工程學院，上海 200237；3.大數據流通與交易技術國家工程實驗室(商業智能與可視化研究中心)，上海 200237；4. 上海大數據與互聯網受眾工程技術研究中心，上海 200072)

0 引言

外匯市場[1]作為全球最大、流動性最好的金融市場，在全球經濟活動中扮演著舉足輕重的地位，外匯市場通過供求關系形成實際匯率來影響國際進出口成本和商品價格，并影響世界經濟發展方向。隨著我國國內外經濟金融的全面深化改革，境內外市場主體較為活躍，外匯交易量增長迅猛。為保護市場參與者的合法權益，維護市場公平、效率、透明和穩定，建設一套合理、高效的外匯市場監測平臺刻不容緩，主要體現在[2]：1)傳統人工監管已無法滿足當前外匯市場發展形勢，需要更科學、有效的外匯市場監測平臺和監管手段；2)外匯管理方式不斷改革，從重視事前審批轉化為重視交易全過程、從宏觀市場監測轉化為微觀指標監測；3)隨著程序化交易的高速發展，市場上的數據不僅數據量暴增，參與交易決策的數據維度也暴增，機構通過大數據和人工智能等技術可能會發現市場上新型交易漏洞從而產生非法套利空間，這給外匯市場監測提出了更高層、更復雜的要求。

對外匯交易市場數據的深入挖掘和利用，實施精準的分析和監測[3-4]，可以提高市場風險預警水平，是外匯市場監測的重要工作方法和內容[5]。目前外匯市場監測主要集中在對外匯市場中的個人外匯業務監測[6-7]，較好地實現了對個人結售匯數據和分拆交易行為的監測[8]。新時期背景下，外匯監測面臨著更多新問題和新挑戰，如數據來源多、格式復雜、數據量大等，因此利用以大數據為基礎的監測手段，打造集數據采集、數據處理和數據挖掘于一體的外匯市場監測信息化平臺是時代發展的趨勢[9-10]，可以及時有效地發現金融市場杠桿和抑制非法操作，維護金融外匯市場穩定[11]。目前在大數據分析及處理業務范疇內，可大致分為兩類：離線計算框架(如Map Reduce、Spark等)和在線(實時)計算框架(如Storm、Spark Streaming等)。其中，基于Spark的分布式計算引擎在各行各業的應用場景都有所研究和落地，比如在電力行業[12]、新零售業[13]、銀行信貸業[14]、醫療服務業[15-16]等。同樣Spark也已經在眾多企業的實際生產中得到了推廣和應用，如Yahoo的Audience Expansion、百度的BMR(Baidu MapReduce)、騰訊的廣點通等[17]。Spark大數據架構在上述垂直領域的成功應用，給外匯市場監測計算框架的設計提供了重要的參考意義。

針對外匯市場交易數據的特性，本文采用Spark作為外匯市場監測指標計算的基礎大數據框架，提出了一種全新的面向外匯市場監測的分布式計算框架。實驗結果表明，本文提出的分布式計算框架能滿足海量數據背景下的成交、報價、交易行為相關指標的高效計算，有效地提高了我國外匯市場監測監管效率和服務水平[18]。

1 外匯市場監測的現狀和不足

1.1 外匯市場監測的特性分析

市場監測是保證金融市場穩定、健康發展的重要手段之一，目前各國外匯監管機構都在持續制定和完善相關監測管理辦法，本文以中國外匯市場為例，針對外匯市場監測業務的特性進行如下分析。

完備性和準確性 中國外匯交易中心作為我國銀行間外匯市場的組織者和承辦者，為保證市場穩定、健康運行，在外匯市場監測方面設計了完善的市場監測指標體系，涵蓋交易前、交易中和交易后整個交易過程，如圖1所示。指標體系中的每項指標都從不同的角度對市場參與者的交易行為進行量化分析，指標間相輔相成、互為補充，形成一個完整的市場監測體系閉環。

銀行間外匯市場作為大額批發市場，一旦發生系統性風險，給機構帶來的損失無法估量，因此作為市場監測系統務必要能準確、及時地識別市場上的潛在異常行為。要達到這一點，就要求系統不管是在宏觀上還是微觀上都需要準確地計算市場的總量、分布、偏離度、增長率等相關指標。

多樣性和復雜性 根據中國貨幣網的披露，外匯市場可按照產品及交易模式兩個維度進行劃分，如圖2所示。

圖1 我國外匯市場部分監測指標Fig. 1 Some monitoring indicators of China’s foreign exchange market

圖2 我國外匯市場分類Fig. 2 Classification of China’s foreign exchange markets

在不同交易模式和產品下指標的計算邏輯有所不同。比如在競價市場上，有效報價次數被定義為報價方報出的點擊可成交報價次數；在詢價市場上，有效報價次數被定義為報價方收到詢價請求次數之后的回價次數；比如在即期市場上，點差等于報賣和報買的即期匯率之差；在遠期市場上，點差等于報買報賣的近遠端全價之差；在掉期市場上，點差等于要報賣報買的遠期點之差。除此之外，由于金融市場業務的復雜性，指標的計算往往無法一步到位，通常需要多個連續的處理步驟，比如數據處理、數據轉換、數據規約、數據計算等，最終形成一個個指標計算管道，如圖3所示，因此，指標計算具有較高的多樣性和復雜性。

圖3 指標計算管道示意圖Fig. 3 Schematic diagram of indicator calculation pipeline

海量數據 全球主流的外匯交易平臺(湯森路透和電子經紀系統)大多提供了基于應用程序編程接口(Application Programming Interface，API)的報價和成交功能，市場行情更新頻率極高，同時基于智能路由的價格發現機制，使得市場可以持續保持高度活躍性，因此外匯市場每日都會產生海量行情和交易行為數據。初步估算，基于湯森路透和電子經紀系統的外匯交易平臺產生的交易相關數據近每月10 TB。

1.2 外匯市場監測傳統技術架構的不足

目前，國內外大多大型金融機構均采用Oracle作為底層數據存儲的基礎設施[19]，圖4展示了傳統基于Oracle的指標分析和計算的一般流程。當面臨外匯市場監測的上述3個特性時，傳統的基于單機或實時應用集群(Real-time Application Cluster, RAC)的Oracle數據分析架構已無法支撐監測指標體系的計算。

圖4 基于Oracle的監測指標計算架構Fig. 4 Computing architecture of monitoring indicators based on Oracle

在海量外匯交易數據背景下，采用傳統的基于Oracle的數據分析架構遇到了諸多困擾。一方面，隨著數據量的持續增長，給Oracle數據庫帶來了巨大的存儲壓力，尤其是當數據量超過10 TB以后，數據庫出現了各種不可預知的紊亂，比如存儲過程執行到一半突然中斷導致任務失敗等；同時數據體量的增長也使得Oracle自我優化的執行計劃遠遠偏離最優執行計劃，導致指標計算總時間不可控。另一方面，外匯市場監測指標的計算依賴于多種基礎信息，如節假日信息、開閉市信息等，這些信息的變更會涉及到Oracle中大量歷史指標數據的刪除和重算，會導致索引文件劇增，數據刪除異常緩慢；此外，外匯市場監測指標體系是由多個市場組成，基于存儲過程的數據分析服務在并行計算多個市場指標的過程中存在嚴重的資源競爭問題，導致多市場指標并行計算效率和順序計算的效率相當，甚至更慢。針對上述外匯市場監測指標計算過程中的諸多困擾，對采用傳統的基于Oracle的外匯市場監測框架所存在的問題[20]總結如下。

1)數據存儲問題。外匯交易市場每日產生的海量數據無法長期存儲在Oracle系統中，最多只能保存一段時間內的數據，導致無法重算歷史監測指標和相應數據查詢。

2)數據訪問問題。Oracle作為行式存儲數據庫，適用于聯機事務處理類任務(On-Line Transaction Processing, OLTP)，而外匯市場監測指標體系的計算屬于聯機事務分析類任務(On-Line Analytical Processing， OLAP)，更適合用列式存儲的文件系統進行數據處理和分析。

3)資源分配問題。外匯市場監測指標計算是多市場、多階段的任務，Oracle缺乏良好的組件對硬件資源進行合理分配和調度，導致多指標計算任務并行處理時，會出現因多任務相互資源爭奪，反而導致整體計算時間延長[21]。

4)計算速率問題。Oracle作為關系型數據管理系統，不適用于分布式計算，很難通過橫向擴展硬件機器來提升系統算力，導致任務的計算速率非常受限。

針對以上4個方面的問題，本文基于內存計算引擎Spark大數據架構提出了一種新的外匯市場監測計算框架。

2 本文搭建的外匯市場監測計算框架

2.1 系統設計與實現

本文分析了傳統Oracle數據分析架構的弊端，采用基于內存計算的分布式Spark架構搭建外匯市場監測分析平臺?；诜植际酱髷祿脚_，能解決傳統外匯市場監測計算框架遇到的4個關鍵問題：

1)采用Hadoop的分布式文件系統(Hadoop Distributed File System，HDFS)來存儲海量交易數據，通過HDFS的動態擴容特性，理論上可以支持無窮大的交易數據。

2)通過Hive+Parquet技術實現底層外匯交易數據的列式存儲，提高數據訪問速率。

3)基于資源調度框架組件YARN(Yet Another Recourse Negotiator)實現硬件資源的量化管理和分配，使得系統資源得到充分利用。

4)基于內存計算和列式數據訪問的Spark SQL實現各項指標分析的計算邏輯，極大地提升了數據處理和指標計算的效率。

除此之外，本文還充分考慮了外匯市場監測的業務特征，進一步提出了適用于外匯市場的通用外匯市場監測框架，主要包括兩個模塊：1)基于Spark的有向無環圖(Directed Acyclic Graph, DAG)作業調度機制，結合外匯市場監測體系的指標計算管道機制和單市場多指標并行計算需求，將外匯市場監測指標體系不同市場的計算任務抽象成為市場級的有向無環圖(Market-level DAG, M-DAG)，使得監管平臺能靈活多變地處理市場多變的需求；2)考慮到外匯市場監測指標體系的多市場、多指標并行計算特征，基于YARN的資源調度池隔離機制，將不同市場的計算任務從硬件資源中隔離出來，提出了市場級資源分配策略——M-YARN(Market-level YARN)，達到并行計算的效果。本文設計實現的基于Spark大數據框架的外匯市場監測框架如圖5所示。

圖5 面向外匯市場監測的分布式計算框架Fig. 5 Distributed computing framework for foreign exchange market monitoring

2.1.1 市場級指標計算模型M-DAG

外匯市場分為多個交易市場，每個交易市場又會根據產品、交易模式形成不同類別的市場監測指標。如圖6是外匯詢價市場中的3個指標(跨產品交易監測、最優報價監測和報價分布監測)間的計算過程及依賴關系，為方便描述，將圖6中的各個階段(Stage)用“S+數字”來代替。根據指標計算的依賴關系，從圖中可以總結出如下3個特性：1)指標計算過程中依賴大量的基礎數據，如最優報價監測(S1)和報價分布監測(S2)兩者都依賴于機構基礎信息、機構退市信息、意向性報價等數據，而跨產品交易監測(S3)除此之外還依賴于起息日、詢價交易流水等數據；2)指標計算過程中存在大量過程重疊，如最優報價監測(S1)、報價分布監測(S2)和跨產品交易監測(S3)都需要復用意向性報價數據的清洗(S6)、意向性報價數據過濾(S8)等過程；3)3個指標根據不同的業務規則使得它們各自存在一些獨立計算的過程。

圖6 3個外匯詢價市場指標的計算依賴關系Fig. 6 Calculation dependency graph of three foreign exchange inquiry market indicators

對于圖6的外匯詢價市場的3個指標計算，如果每一個指標(比如S1、S2、S3三個指標)設計成一個獨立的任務來計算，如圖7所示，會存在如下兩個方面的問題：1)存在大量重復計算過程，計算效率下降，如圖7(a)、(b)和(c)需要重復計算過程S8、S6和S4；2)業務數據變更，關聯指標所有過程都需重新計算，維護性較差。如圖7中S11和S12的數據變更，三個指標任務S1、S2和S3的所有過程都需要重新計算。

因此本文結合Spark的內存計算模型和DAG任務調度的設計思想[22]，針對各市場中的指標計算流程按照其依賴關系進行融合，將指標計算的每一個過程都抽象為互相獨立的模塊，使得同一個市場的所有指標計算任務成為一個全局有向無環圖(DAG)，在DAG中通過對后續任務共同依賴的、計算耗時的重要節點進行緩存來避免重復計算，從而提出了市場級指標計算模型M-DAG。根據M-DAG的思想，圖7中的三個指標計算任務圖融合成市場級指標計算模型如圖8所示。

相對于傳統的單指標獨立計算的設計，本文提出的市場級指標計算模型M-DAG能提高整體框架的靈活性和適用性：1)充分利用各指標重復計算過程的緩存數據，提高了計算效率。例如圖8中，計算過程S8只需要計算一次，而在圖7中需要重復計算3次。2)可靈活調整各獨立模塊的依賴關系，降低了業務數據變化所引起的重復計算成本和項目迭代成本。例如在圖9(a)中當S11和S12相關的業務數據發生變化時，S1、S2和S3指標需要重新計算，但根據M-DAG的剪枝思想，S13和S10可以被剪枝，只需要計算如圖9(b)所示的DAG即可。3)任務的提交可以是特定的指標也可以是整個市場所有的指標。如圖10(a)所示，可以向系統單獨提交S1的計算任務,這樣DAG被簡化成為如圖10(b)所示；如圖11所示，在S1、S2、S3前面增加一個前驅節點S0，并向集群提交S0的計算任務，這樣全市場的指標一起計算。一般在生產環境正常情況下采用全市場一起計算的提交模式，當生產環境面臨因數據問題需要重新回溯計算時，只需要提交受影響的節點即可。

圖7 三個外匯詢價市場指標的一般計算方式Fig. 7 General calculation method of three foreign exchange inquiry market indicators

圖8 外匯詢價市場的市場級指標計算模型M-DAG示例Fig. 8 Example of market-level indicator calculation model M-DAG for foreign exchange inquiry market

2.1.2 市場級資源分配策略M-YARN

外匯市場根據交易模型和產品可以分為多個不同的市場，每日閉市之后有大量指標需要計算，并且每個市場指標計算復雜度不一，計算時長不一。如果對每一個交易市場的指標順序向集群提交任務，一方面會造成集群資源浪費的情況，另外一方面也不能滿足外匯市場監測的時效性要求，因此為進一步提高多交易市場外匯市場監測效率，充分利用集群資源，本文基于各交易市場監測指標計算復雜度，利用YARN的資源調度池隔離機制[23]，提出了基于指標復雜度的市場級資源動態分配策略——M-YARN。

YARN作為一個通用的資源管理平臺，利用其核心模塊Resource Manager(RM)對客戶端提交的所有作業(Job)進行統一資源管理和分配，充分利用集群資源，從而達到集群效率最優。本文基于YARN的資源調度池隔離機制提出的M-YARN策略在交易市場層面對每個市場按照其指標計算復雜度進行資源分配，而在一個市場內部則使用YARN自帶的主資源公平(Dominant Resource Fairness, DRF)調度算法[24]進行資源分配和管理。本文M-YARN策略的每個市場的分配資源Ri表示如下：

其中：R表示集群總資源；Ci表示第i個交易市場Mi的計算復雜度，且Ci∝(交易市場Mi的數據量，Mi的指標數，Mi任務Spark Shuffle數)。為方便計算，本文僅采用交易市場的數據量表示該市場的指標計算復雜度。

圖9 市場級指標計算模型M-DAG剪枝計算示例Fig. 9 Pruning calculation example of market-level indicator calculation model M-DAG

圖10 單一指標計算模型示例Fig. 10 Example of single indicator calculation model

圖11 全市場指標計算模型示例Fig. 11 Example of full market indicators calculation model

3 實驗結果與分析

3.1 實驗環境及實驗數據

本文實驗環境包括Oracle數據庫環境和Spark平臺環境，實驗集群中各個節點的配置屬性如表1所示。

為方便實驗對比，本文在兩個不同平臺框架上采用相同數據集進行對實驗對比，模擬了外匯撮合和詢價市場1個月的交易數據。另外，由于在實際外匯市場監測中，每一個交易市場都會涉及大量的計算指標，為方便各交易市場指標實驗對比，本文針對外匯撮合市場和外匯詢價市場各選擇了3個計算復雜度較高的指標，即外匯撮合市場：最優報價監測、有效雙邊點差分布、跨產品交易監測；外匯詢價市場：最優報價監測、報價分布監測、跨產品交易監測。數據集的詳細統計信息如表2所示。

表1 實驗環境對比 Tab. 1 Comparison of experimental environments

表2 數據集詳細統計信息 Tab. 2 Detailed statistics of dataset

3.2 M-DAG有效性驗證

為驗證本文提出的市場級計算模型M-DAG的有效性和業務應用性，針對外匯詢價市場的最優報價監測、報價分布監測、跨產品交易監測3個指標設計了以下兩組實驗：

1)使用和未使用M-DAG計算模型的情況下，外匯詢價市場三個指標的計算效率實驗對比，如圖12所示。

圖12 不同模型策略下的指標計算效率對比Fig. 12 Efficiency comparison of indicator calculation of different model strategy

2)使用和未使用M-DAG計算模型的情況下，業務數據變更(S11和S12)所引起的外匯詢價市場各指標重算代價實驗對比，如表3所示。

表3 指標重算代價對比 Tab. 3 Comparison of indicator recalculation

從圖12中可以看出，對于三個指標，“最優報價監測”指標的計算時間兩者相差不大(77.0 min vs. 75.0 min)，這是因為該指標都是兩種計算策略的第一個計算指標。在第二個指標“報價分布監測”的計算上則開始表現出計算效率的差異，使用M-DAG計算模型的時間比未使用了該計算模型的時間下降了12.0 min，約31.9%，這是因為第二個計算指標跟第一個計算指標存在大量重復計算(如圖7所示)，在未使用M-DAG計算模型情況下需要重復計算S8、S6等過程(如圖7(b)所示)，而通過使用M-DAG計算模型可以節省大量重復計算時間。M-DAG計算模型的有效性在第三個指標“跨產品交易監測”上體現得更為明顯，由于第三個指標跟前面兩個指標的計算過程存在更多重復計算，如S8、S9、S10等(如圖8所示)，通過使用M-DAG計算模型節省了約38.9%的時間?？傮w來看，使用M-DAG計算模型時三個指標的計算總時間僅為162.8 min，比未使用M-DAG計算模型時計算效率提高了24.5%，說明使用M-DAG計算模型可以充分利用各指標重復計算過程數據，提高市場所有指標的整體計算效率。

從圖6中可以發現，在外匯詢價市場中，當業務數據在S11和S12處發生變化時會涉及到三個指標的全部或部分過程計算變更，其中：“最優報價監測”和“報價分布監測”指標需變更全部計算過程，“跨產品交易監測”指標需變更部分計算過程。在未使用M-DAG計算模型的情況下，從圖7中可以知道，需要向Spark集群全量提交三個計算任務，重算時間跟初始計算時間一致(如表3所示)，而在使用了M-DAG的情況下，由于“跨產品交易監測”指標只涉及到部分業務數據的變更和部分過程變更，通過調整各模塊之間的依賴關系，僅需要向Spark集群提交如圖9(b)所示的計算任務，重算時間為146.3 min，比未使用M-DAG計算模型節約了69.3 min，提高了32.14%的計算效率，因此本文設計的M-DAG計算模型可以靈活調整各獨立模塊的依賴關系，降低業務數據變化所引起的重復計算成本和需求等待成本。

3.3 不同外匯市場監測計算框架對比與分析

為保證實驗的公平性，所有實驗使用相同的數據集，并針對不同的數據集使用單指標計算時間和交易市場總計算時間對本文設計實現的外匯市場監測計算框架進行性能評價。本文共對比了三種計算框架：傳統Oracle框架、本文框架(未使用M-YARN)、本文框架(使用M-YARN)。實驗結果比較如表4所示。

表4 不同外匯市場監測計算框架實驗結果對比 Tab. 4 Experimental result comparison of different foreign exchange market monitoring computing frameworks

從表4可以看出，對于本文提出的外匯市場監測計算框架，在未使用M-YARN資源分配策略，總計算時間為206.8 min，是傳統Oracle框架(1496.0 min)的0.14倍；當使用M-YARN資源分配策略，總計算時間為180.4 min(由于兩個市場并行計算，因此總計算時間取其大)，僅是傳統Oracle框架總耗時的0.12倍。表明在多交易市場并行計算過程中無論是否使用M-YARN資源分配策略，本文提出的外匯市場監測計算框架性能表現都遠遠優于傳統的基于Oracle的計算框架。

對于未使用M-YARN的本文框架，在計算復雜度稍低的外匯撮合市場中，總耗時為44.0 min，比傳統Oracle計算框架耗時下降了約87.5%。在計算復雜度較高的外匯詢價市場中，該框架和傳統Oracle框架耗時都所有增加，但是本文框架耗時上升幅度較小(162.8 min vs. 1 144.0 min)，耗時比傳統Oracle框架下降了85.8%，而耗時下降率相比復雜度較低的外匯撮合市場(87.5%)下降了1.7個百分點。經過實驗表明，在高復雜度的外匯詢價市場中，基于Oracle的框架未能計算出“跨產品交易監測”指標(具體表現為計算卡死和中斷等行為)，而本文框架則在較短的時間內(57.8 min)計算出了該指標。

另外，從表4中可以看出，在低指標計算復雜度和高指標計算復雜度的交易市場中，使用M-YARN市場級資源分配策略使得各市場監測指標計算耗時分別增加了50.0%(66.0 min vs. 44.0 min)和10.8%(180.4 min vs. 162.8 min)，這是因為雙市場并行計算，每一個交易市場中指標計算時的資源只占用了全部資源的部分，但是在使用M-YARN市場級資源分配策略的情況下，總耗時反而下降了12.8%(180.4 min vs. 206.8 min)，由此可見，本文提出的市場級資源分配策略對多市場并行的監管計算效果起到了很好的提升作用。

4 結語

面向海量外匯交易數據，本文提出了一種基于Spark分布式大數據處理技術的外匯市場監測計算框架，并針對外匯市場監測的特定業務需求，基于Spark內存計算和DAG任務劃分算法思想，將不同市場的業務指標計算過程抽象為市場級的DAG(M-DAG)模型，使得業務規則更加靈活多變，同時實現了基于監測指標計算復雜度的市場級資源分配策略(M-YARN)進行多市場監測指標并行計算。實驗結果表明，本文提出的外匯市場監測分布式計算框架可以滿足大數據背景下外匯監測業務指標計算的完備性、精準性和時效性，同時對于其他行業或領域的監測平臺設計也有一定的指導和借鑒意義。

在未來工作中，一方面將繼續探索更加符合外匯交易的市場級資源自動分配策略，同時針對不同市場級資源池設計更細粒度的市場資源分配算法；另一方面將基于現有分布式計算框架設計和研究金融市場的一站式大數據監測平臺，并基于該平臺搭建更多業務型應用，比如基于分布式機器學習的智能預警模型、量化研究模型和風險管理模型等[25]。