服務上億用戶，中國結算新一代數據集市技術實踐
盧向澄
金融科技領域十余年工作經驗，目前在中國證券登記結算公司從事技術架構工作，專註於技術中臺、雲平臺、大數據平臺等領域。
1.背景介紹
我國股市約有1.2億散戶，直接關乎上億家庭、數億人切身利益，保護好投資者尤其是中小投資者的合法權益，是資本市場工作人民性的具體體現，也是服務實體經濟的應有之義。黨的十九大明確提出“必須堅持以人民為中心的發展思想”。中國證監會有關負責人表示，要認真貫徹落實十九大精神和黨中央、×××關於資本市場建設的一系列決策部署，加快推動形成融資功能完備、基礎制度紮實、市場監管有效、投資者合法權益得到有效保護的多層次資本市場體系，切實做好投資者保護工作。證監會主席劉士余先後多次強調“投資者保護重如泰山”、“保護投資者合法權益是證監會職責和使命所在”、“保護中小投資者合法權益是天大的事”。
在數據集市建設之前，數據查詢主要依賴於數據倉庫。數據倉庫是一個集成的、面向主題的數據集合，設計的目的是支持決策支持系統的功能。在數據倉庫裏，每個數據單元都與特定的時間相關。
數據倉庫包括原子級別的數據和輕度匯總的數據，是面向主題的、集成的、不可更新的(穩定性)、隨時間不斷變化（不同時間）的數據集合，用以支持經營管理中的決策制定過程。數據倉庫是一個典型的OLAP系統，在高並發、快速響應的場景下具有很大的局限性，無法滿足海量投資者數據查詢服務需求。
為了保證投資者服務系統在低延時和高並發查詢的情況下具備足夠的支撐能力，可以7×24對外提供數據服務，且不影響原有數據倉庫統計分析應用的正常運行，最終決定建設面向投資者服務的專業數據集市。
投資者服務數據集市主要目標是以面向用戶體驗為基礎，具有業務敏捷、分布式服務、高伸縮、高可用、易管理維護等特點，為多渠道的投資者綜合服務專區服務。先期開始建設的數據集市主要包括有新三板市場投資者服務數據集市、基金市場投資者投票服務數據集市、全市場在線業務查詢數據集市。
其建設原則應包括：
?抓住主線功能需求；
?采用主流技術；
?滿足未來發展需求；
?充分驗證測試。
基於上述目標和原則，我們總結了如下需求。
1.功能性需求
?存儲現有數倉中滬深市場、新三板市場、基金市場等各類投資者數據；
?支持結構化和非結構化數據；
?數據庫和其他服務組件具備動態擴容能力，以支撐數據集市階段性發展的容量和計算能力需要；
?支持T+1批量數據的ETL功能，能夠從TeraData數據倉庫及其他數據庫采集數據；
?支持實時數據流處理能力，實現準實時數據同步；
?支持數據加工，主要是多表關聯和聚合運算；
?數據庫支持SQL和API訪問接口，方便應用開發；
?支持數據備份恢復；
?具備完善的管理功能，例如監控、配置和任務調度等；
2.非功能性需求
?海量數據存儲。初期至少支撐100TB存儲容量，遠期支撐PB級；
?高並發訪問。初期至少支撐1萬並發查詢，遠期支撐10萬並發查詢；
?低延時。在高並發情況下，查詢響應時間不超過100毫秒；
?7x24持續穩定運行。在高可用集群技術支撐下能夠實現集群級別的不間斷持續穩定運行，並能夠在絕大部分場景下進行不停止集群的數據庫維護工作。
3.安全性需求
?數據高可用。需要支持多副本冗余。在部分副本數據損毀情況下，保障數據不丟失；
?用戶身份驗證和權限管理。用戶不可越權訪問數據；
?完善的審計功能。能夠完全記錄所有數據訪問和數據操作。
3.技術架構
邏輯架構
截止目前，三個數據集市的數據分別來源於數據倉庫和基金投票系統。所有數據需經過ETL處理後存儲到數據集市中，部分數據還需經過批量加工處理後，供下遊數據使用者查詢。數據集市的邏輯架構如圖一所示。
圖一：邏輯架構
如圖一所示，從上遊數據源到下遊使用者，中間提供服務的數據集市內部包含數據采集、數據傳輸、數據處理、數據存儲和平臺服務這五大功能模塊。
其中，數據采集、數據傳輸、數據處理可以類比為傳統的ETL功能模塊。但是，這個數據集市的ETL功能模塊包含了兩種ETL方式：批量數據ETL和準實時數據同步。
1.數據采集途徑一：批量數據ETL
新三板市場和全市場在線業務數據集市要求數據每日更新。因此這兩個數據集市均采用傳統的ETL方式，即每日定時導出批量數據到文件（Extract），然後經過文件傳輸、數據轉換（Transform）和數據加載（Load），最終將數據放入數據集市的數據庫中存儲，以供下遊使用者查詢。我們稱這個流程為批量數據ETL。主要包含以下步驟：
1）定時抽取：每天夜間，數據倉庫裏邊的數據加工處理完畢之後，數據集市的抽取任務定時啟動，將約定數據接口的新增數據或者全量數據抽取到數據文件中。
2）文件緩存：抽取環節生成的數據文件需要存放到文件系統中，以備後續數據處理之用。另外，數據文件需要壓縮緩存多天，作為數據備份使用。
3）批量處理：兩個數據集市的大部分接口數據只需要數據轉換和加載入庫。少量接口數據需要在數據入庫之後進行加工處理。加工的主要需求是預關聯，即將兩表或者多表數據關聯形成更多字段的新表，以滿足兩個數據集市的數據查詢需求。
2.數據采集途徑二：準實時數據同步
基金市場投資者投票服務數據集市對數據時效要求較高，要求數據準實時同步，以數據準實時查詢。具體而言，即要求上遊系統（基金市場投資者投票服務系統）的數據發生變化（包含增刪改）之後，數據集市內的數據也需在短時間之內（5秒之內）實現相同的變化。我們稱這種ETL方式為數據準實時同步，也可稱為實時數據流處理。主要包含以下步驟：
1）實時采集：該步驟要求最短時間內發現源數據庫的數據變化，包含對應庫表的數據的增刪改，並且不對源數據庫產生明顯的性能影響。
2）緩存隊列：為了增加穩定性和吞吐量，在實時采集和實時數據加工處理環節中間使用數據緩存。該緩存以隊列的方式，保障數據先進先出的順序關系。該緩存隊列要求具備優秀的響應性能、並發能力、高吞吐量、穩定性和高可用能力，以保障數據同步流程安全可用。
3）實時處理：該環節包括數據加載和實時統計兩方面作業內容。每條投票數據順序進入緩存隊列之後，由實時處理程序順序的讀出並加載入庫，同時實時統計投票數等重要數據，用於基金投票狀態的實時展示。
3.數據存儲
數據經過ETL過程之後，被存入數據庫，主要包括賬戶數據和交易明細數據。
4.數據服務
數據查詢是數據集市最核心的服務。新三板市場投資者服務數據集市和全市場在線業務數據集市這兩個數據集市主要提供賬戶數據、證券交易流水查詢服務。基金市場投資者投票服務數據集市主要提供投票詳情及實時統計結果查詢。歸結起來，這些主要是高並發的精準查詢。
數據架構
數據進入數據倉庫之後，將根據分析或者查詢的需求，加工和匯總成相應主題。因此，數據集市的數據也將按照查詢主題進行組織和管理。
根據數據主題及數據處理加工流程，我們規劃設計了數據架構如圖二所示。
圖二：數據架構圖
目前已經實施了如下三個數據集市：
?新三板市場投資者服務數據集市；
?全市場在線業務查詢數據集市；
?基金市場投資者投票服務數據集市。
上述三個數據集市之間不共享數據、不需要關聯查詢、不存在交叉訪問權限，是可以完全獨立運行的。但是，在數據庫中不是分庫管理的，而是通過權限控制形成邏輯層面的獨立數據集市，這樣可以共享軟硬件資源。
新三板市場投資者服務數據集市和全市場在線業務查詢數據集市的數據來源均為數據倉庫。數據接口形式為T+1的批量數據文件，即每日證券市場收市清算交收批量處理產生的數據。兩個數據集市由不同的邏輯數據域存儲，管理隔離。同時，由於這兩個數據集市的數據查詢需求中存在表關聯情況，而頻繁的並發關聯查詢需要消耗大量磁盤I/O、內存和CPU計算時間，所以要對多表關聯進行預加工處理，即將多表關聯到一張表中，以便於將多表關聯查詢轉變為單表查詢，從而提升查詢效率。
基金市場投資者投票服務數據集市的數據來源於上遊交易系統數據庫的數據實時采集，即數據變化實時同步到數據集市中。同時，由於基金投票場景中存在實時顯示投票進展的需求，所以需要實時統計各投票選項的票數，對每條投票數據進行實時累加統計，並將結果更新入數據集市的統計表中。
下遊各業務系統通過查詢服務接口可以隨時查詢對應數據集市的數據。查詢服務提供身份驗證、權限管理和查詢接口，不允許修改數據。
物理架構
根據數據集市功能需求、邏輯架構和數據架構，我們規劃的物理架構可以用圖三來表述。
圖三：物理架構圖
包括以下四個部分：
1.批量數據ETL服務器
該服務器用於批量數據ETL流程。服務器中運轉ETL主控程序、數據轉碼程序和數據批量加載程序。這些應用均為Java語言開發。ETL主控程序使用統一調度監控系統（外部系統）的定時作業調起，完成指定數據接口的指定ETL過程，例如檢查數據文件到達情況，調用數據轉碼或者數據裝載等動作。數據轉碼使用Java程序調用Python轉碼程序完成，能夠做到GBK編碼到UTF-8編碼的轉換，並且吐出轉碼失敗的數據。數據批量加載程序主要是通過快速加載工具完成，並且檢查加載結果是否正確。所有程序均具備錯誤檢測及告警能力。
另外，該服務器的文件系統作為數據文件緩存使用，並由一個清理程序自動維護。超過緩存期限的數據文件將被自動清理，以保持文件系統剩余空間足夠使用。
該服務器為X86 Linux虛擬服務器，配備4TB磁盤空間。
2.數據緩存隊列服務器集群
該服務器集群由三臺服務器組成，其中部署三副本的Kafka集群，並配合外部Zookeeper集群的一致性服務，從而實現高可用的消息隊列服務集群。Spark集群中，在Spark streaming分布式數據流引擎中運行Java應用程序實現小批次的實時數據加載入庫和實時數據統計計算。使用Spark SQL作為批量數據加工引擎，主要實現多表關聯的預處理作業。
這個集群中的服務器均為X86 Linux虛擬服務器，每臺服務器配備1TB磁盤空間。
3.集市數據庫及並行計算服務器集群
該服務器集群中部署了兩個邏輯集群，分別是分布式數據庫集群和Spark集群。分布式數據庫作為數據存儲層，Spark作為計算層。這樣規劃的原因主要有兩點：1）兩者資源需求互補，即數據庫最耗I/O，而Spark最耗CPU和內存，能夠充分利用服務器硬件資源；2）分布式數據庫和Spark均為分布式架構，Spark計算單元訪問本服務器的分布式數據庫節點可以具備最好的性能。
分布式數據庫集群部署為三副本高可用模式。其高可用機制由數據庫引擎自身提供，無需借助Zookeeper。
Spark集群的高可用機制借助處於系統外部的Zookeeper實現。
這個集群中的服務器均為X86 Linux物理服務器，每臺服務器配備10塊4TB硬盤。
4.應用服務器集群
數據查詢服務及管理服務均部署於應用服務器中，並且集群化部署，以提供負載均衡和主備容災能力。這些應用服務通過分布式數據庫提供的訪問接口（SQL JDBC和Java API兩種方式）的連接池方式連接分布式數據庫。
應用服務同時需要提供管理功能，例如用戶管理、權限管理、配置管理、監控等功能。下遊業務系統通過F5負載均衡服務器訪問應用服務。
應用服務器集群的服務器均為X86 Linux虛擬服務器。
4.關鍵技術
依據數據集市的整體需求，設計上述系統架構的過程中采用了抓住主線功能需求、采用主流技術、滿足未來發展需求、充分驗證測試的設計原則。
通過大量的功能、性能、穩定性驗證測試，該平臺最終選擇了如下軟件以實現對應的需求：
?分布式數據庫選擇國產巨杉數據庫，可支撐海量數據存儲和低延時高並發的數據查詢，並具有金融企業級數據訪問安全審計功能
?Spark SQL支撐批量數據加工和統計
?Spark streaming支撐實時數據流計算處理
?Kafka支撐實時數據流的數據緩存
下面具體介紹一下主要軟件技術特性。
分布式數據庫：巨杉數據庫SequoiaDB
我們也對Hbase和巨杉數據庫進行了調研和測試，對比分析結果如表二所示，最終考慮到業務場景和技術支持服務情況，選擇了巨杉數據庫。
表二：巨杉數據庫對比Hbase
巨杉數據庫作為典型的Share-Nothing的分布式數據庫，同時具備如下特性：
?分布式、可擴展、高容量；
?高性能、高並發；
?高可用、高穩定性；
?支持SQL；
?企業級管理功能。
巨杉數據庫采用分片技術為系統提供了橫向擴展機制，其分片過程對於應用程序來說完全透明。該機制解決了單臺服務器硬件資源（如內存、CPU、磁盤 I/O）受限的問題，而且並不會增加應用程序開發的復雜性。
巨杉數據庫采用經典的分布式技術架構，如圖四所示。
圖四：巨杉數據庫整體架構
巨杉分布式數據庫引擎主要由三種節點組成：
?協調節點：負責調度、分配、匯總，是巨杉數據庫的數據分發節點，本身不存儲任何數據，主要負責接收應用程序的訪問請求；
?編目節點：負責存儲整個數據庫的部署結構與節點狀態信息，並且記錄集合空間與集合的參數信息，同時記錄每個集合的數據切分狀況；
?數據節點：承載數據存儲、計算的進程，為用戶提供高性能的讀寫服務，並且在多索引的支持下針對海量數據查詢性能優越。多個數據節點可以組成一個數據節點組，根據選舉算法自動選擇一個主數據節點，其余節點為備數據節點。
數據集市部署巨杉數據庫時采用三副本冗余高可用方式。各副本之間，由數據庫引擎實現自動的同步或者異步日誌復制機制。保證了多副本之間的數據一致性。當其中一副本(主節點)出現故障時候，其他兩副本能夠快速選舉新的主節點，並且繼續提供數據讀寫服務。該部署方式可以保證不出現單點故障。
圖五：巨杉數據庫多副本高可用
巨杉數據庫的分布式和多副本的部署方式，可以最大程度實現高效數據庫高並發訪問，並且保障平臺整體平穩運行。
應用訪問巨杉數據庫的接口方式主要有Json API方式和SQL方式。Java應用通常采用Java API驅動或者JDBC驅動來連接巨杉數據庫。巨杉數據庫兼容標準SQL語法，也可采用Java API接口方式可以在簡單查詢的場景下獲得最高的性能。實際上，API方式對於互聯網應用開發者而言，才是更加熟悉和習慣的數據訪問方式。
巨杉數據庫支持完整的企業級數據庫管理的各項功能：
?審計日誌可以記錄完整的數據訪問和數據操作；
?備份和恢復；
?快照和列表（監控）；
?支持實時同城災備、準實時異地災備；
?支持靈活的強一致性和最終一致性配置；
?集群擴容；
?可視化管理頁面；
批量數據加工：Spark SQL
新三板市場投資者服務數據集市和全市場在線業務數據集市這兩個數據集市的ETL流程完成之後，需要對部分數據接口進行預處理，主要是多表關聯。由於數據倉庫不提供某些需要聚合運算的接口，所以改由數據集市進行一些數據加工處理，主要是關聯和聚合處理。加工之後的數據供集市應用查詢。數據加工的模式主要是SQL處理，類似於： INSERT INTO ... SELECT ... FROM A LEFT OUTER JOIN B ON (...) GROUP BY 。我們使用Spark SQL進行這樣的數據批量處理。
Spark SQL作為計算引擎，通過巨杉數據庫提供的連接器可以無縫訪問巨杉數據庫，並且盡量訪問本地節點。其架構示意圖如圖六所示。
圖六：Spark連接巨杉數據庫
上述流程的核心技術基礎是Spark SQL可以無縫平滑的訪問巨杉數據庫，而且都是分布式並行計算和分布式並行存儲引擎。巨杉數據庫提供了Spark連接器。該連接器可以充分下壓查詢條件到數據存儲節點，並且能夠根據數據分布特征自動的盡量從本地節點讀取數據。這樣的連接器充分利用了分布式並行系統的並發I/O和計算優勢。
針對批量數據加工場景，我們對比了Spark SQL和DB2及MySQL做兩表關聯的性能。兩張表的數據量分別為6000萬條和2000萬條。 Spark SQL+SDB耗時10秒之內，DB2耗時2分鐘左右，而MySQL耗時過長沒有統計結果。Spark SQL+SDB的架構可以滿足我們的需要。
實時數據流處理：Kafka + Spark streaming
實時數據流技術可以將傳統的批量數據ETL方式的數據延遲程度從1天（或者幾個小時）大幅度提升到1分鐘以內，可以將源系統數據變化及時同步到目標數據庫中，並且實時計算統計數據。
實時數據入庫數據處理流程如圖七所示。
圖七：實時數據入庫流程
數據的源頭系統處，需要部署實時數據采集軟件或者應用。源數據庫是MySQL數據庫，所以采用了能夠實時解析binlog的軟件愛可生的DTS。
基金投票應用將投票數據實時寫入MySQL數據庫投票明細表，由DTS服務集群實時將MySQL庫投票明細表的 BinLog數據解析為JSON字符串格式的消息，插入Kafka集群的topic中。Spark Streaming 應用實時的從kafka的消息隊列即topic中讀取消息。實時數據處理應用拿到Spark streaming提供的數據流之後，根據統計規則做實時統計，將實時統計結果入MySQL庫，並將數據實時插入巨杉數據庫。庫中的投票數據再由Spark SQL應用每天定時統計，並將統計的結果插入MySQL數據庫中，供基金投票應用查詢每天的投票結果。
基金投票實時數據讀取入庫應用基於Spark streaming流處理技術，應用程序不間斷的從kafka topic中獲取JSON字符串類型的數據，將獲取的一條條消息數據封裝為一個個RDD，再將多個RDD封裝為一個DStream(離散數據流)。Spark Streaming基於Dstream開始對數據進行第一步轉換，清洗，加工得到符合入庫條件的DStream流。最終使用Spark Streaming輸出流算子對數據進行輸出操作，將數據通過SequoiaDB JavaAPI 插入至SequoiaDB。
值得註意的是，Kafka內部配置多個topic，每個topic對應一張數據表。如果某張數據表的數據更新允許基於某個字段的分布式並發，則將該表對應的topic設置為多個partition的，以通過並發機制提升數據吞吐量。否則，切不可配置為partitioned，以免並發機制造成數據更新次序錯亂，從而造成數據集市的數據與數據源不一致。
我們在少量硬件資源情況下進行的性能壓測中得到的該實時數據流處理架構的基本性能是1.8萬TPS，並且延時少於5秒，完全可以滿足基金大會投票的數據同步需要。
5.總結和展望
項目成果
投資者服務數據集市已成功上線實施，很好的滿足了如下業務需求：
?擺脫人工受理和手工查詢的現狀，全市場和新三板投資者可以通過APP等渠道自助在線查詢，例如證券持有情況查詢、歷史交易明細查詢等；
?將數據集市單獨建設，減輕了數據倉庫壓力，不影響數倉的主要功能，即統計分析，實現數倉減負；
?幫助基金市場投資者投票系統實現大數據量存儲和實時在線查詢，及實時統計；
?高並發在線查詢情況下，依然提供快速響應的能力，提升用戶體驗；
?7x24持續在線服務，不發生故障；
?數據庫高可用特性充分保障數據安全，不丟失數據；
?完善了用戶身份驗證及數據權限管理，避免越權訪問；
?數據訪問的完全審計記錄，實現所有操作可追溯；
通過上線前的性能測試及上線試運行，已經體現了分布式數據庫+Spark技術架構應用在數據集市的優異性能。主要的性能指標如下：
?批量數據加載性能：大於50MB/S，大於5萬條/秒
?實時數據流吞吐量：大於20MB/S
?實時數據流延遲：小於5秒
?查詢性能：1000並發情況下，>28000TPS
未來規劃
投資者服務數據集市的未來規劃是發展為統一數據查詢平臺，為不同業務系統提供歷史數據和準實時數據的低延時高並發查詢服務。各業務系統的數據集中存放於統一數據查詢平臺，可以提升數據關聯價值，也可以起到數據歸檔作用。
基於此遠期規劃和已經完成的工作，下一步工作的主要內容是批量數據ETL流程優化、生命周期管理和元數據管理。
1.批量數據ETL流程優化
目前，數據集市的數據主要來自於數據倉庫的T+1天數據。數據的承載形式是數據文件。ETL流程采用傳統的成熟方案，即，抽取源數據庫的數據落地成為數據文件，然後FTP傳輸到ETL服務器的文件系統中，數據文件經過校驗和轉換處理，最後通過數據庫提供的批量加載工具完成數據高速裝載入庫。這種傳統的ETL方式的優勢是技術成熟、容易查錯、數據落地存檔。但是也有明顯的劣勢，例如環節多、易出錯、分隔符問題、整體速度慢、占用文件系統空間等。由於有明顯缺點存在，對其優化就是下一步的主要工作。
未來，考慮對ETL流程進行優化。主要考慮采用數據不落地成文件的方式。目前已經了解的方式主要有商用ETL軟件、Spark SQL應用、定制化開發Java應用。我們將通過對比測試的方法來評估這些方式的優劣勢。
可以預見的是，定制化開發Java應用可以具備最好的需求匹配度，能夠彌補傳統ETL方式及其他方式所固有的各種劣勢。但是，定制化開發Java應用必然存在成熟度不足的風險，以及開發成本較高、管理功能薄弱等問題。定制化Java應用的關鍵環節是處理好字符集轉換、數據類型轉換和結果校驗。相較而言，利用Spark SQL的外表功能實現的insert into ... select ... 方式的ETL流程如果在性能、功能、穩定性等方面能夠滿足生產需要，該Spark SQL方式將會有可能獲得最大的優勢。由於Spark SQL內部的復雜容錯機制，該方式的關鍵技術環節是完整捕捉錯誤異常以及結果正確性校驗。
2.數據生命周期管理
未來隨著時間推移，某些舊數據可能徹底失去保存意義或者失去查詢價值，需要將其刪除或者歸檔到離線存儲中。這時就需要配備數據生命周期管理功能。
數據生命周期管理的技術不復雜。它的技術重點在於方便的配置管理和後臺定時自動清理和歸檔數據功能。
3.元數據管理
未來隨著平臺接入業務系統增多，以及時間的推移，數據自然會不斷增多。沒有良好的元數據管理，將降低管理效率、提高管理成本，甚至造成失控。元數據管理的重要性會逐漸顯現。
對於統一查詢平臺而言，元數據管理復雜度並不算高。我們可以設計一個簡單明了的元數據管理模型，重點管理如下信息：
?數據來源
?所有者/權限
?數據批次/時間
?數據屬性
?數據關系
?數據存儲位置
服務上億用戶，中國結算新一代數據集市技術實踐