基于流計算的保險大寬表系統的應用研究

安建民，周一波，彭送庭

（英大泰和人壽保險股份有限公司，北京 100089）

0 引言

統計分析系統是公司各級管理人員了解前一日公司經營情況的重要系統。該系統通過數據抽取軟件，每日批處理生成各種統計指標。隨著互聯網的高速發展，數據規模的不斷增加，數據變化速度越來越快［1］，用戶希望得到更快的數據處理和響應時間［2］，因此建設數據實時展示系統，實現對指標數據的實時計算變得越來越重要。

為讓公司各級管理人員能夠實時看到指標數據，本文設計的保險大寬表系統，利用分布式流計算技術［3］解決了在分布式環境下數據處理的一致性問題［4］、在數據從源數據庫Oracle同步到MPP數據庫的時效問題以及在界面展示時數據的主動推送問題。對比統計分析系統是以T+1［5］的模式對數據進行分析處理，無法實時查看最新保單數據信息，該系統極大地縮短了數據處理時間，實現了保險數據基本指標T+0模式下的數據實時同步和計算［6］。

1 系統總體設計

本文主要采用分布式流計算實現個險渠道大寬表保單數據實時計算、多表關聯和分布式處理。通過采用分布式流計算技術，應用Kafka和Spark Streaming［7］實現分布式并行處理，提升系統計算性能和高可用性［8］；應用MPP數據庫，提高了對海量數據的分析處理［9］和并行數據計算能力，讓業務人員能即時知曉最新保單和匯總數據的變化情況。

圖1詳細展示了保險大寬表系統設計框架。

圖1 保險大寬表系統設計框架

2 詳細設計分析

2.1 數據采集和遷移

保險大寬表系統通過采用OGG（oracle golden gate）源端去抽取和投遞保單變化的所有數據信息。OGG-BD（OGG-bigdata）端會通過復制進程把發送過來的Trail file數據存入到分布式消息隊列Kafka中。傳統保險系統數據庫應用以Oracle為主，在進行數據采集和遷移過程中，對數據業務的處理有很高的時效性要求。OGG是一種成熟的數據遷移產品，可以在異構的基礎上實現大量數據的秒級數據采集、轉換和投遞。通過解析源數據庫在線日志或歸檔日志獲得數據的增、刪、改變化，再將這些變化應用到目標數據庫，實現源數據庫與目標數據庫的同步和遷移。

2.2 數據分析處理

本文針對海量、多源、處理效率要求高的保險業務數據特點設計了基于Kafka和Spark Stream?ing的分布式并行處理方案。Kafka是具有高吞吐量的分布式消息訂閱和發布系統［10］。主要由Pro?ducer（生產者）、Broker（代理）和Consumer（消費者）三大部分構成［11］。其中生產者負責將收集到的數據推送到代理，而代理負責接收這些數據信息，并將這些數據本地持久化。消費者則直接對這些數據進行處理［12］。

在技術上，本文應用SharePlex監控Oracle日志文件，實時獲取數據庫的操作消息，獲取數據庫系統中增、刪、改的數據。同時應用Kafka記錄監控的數據信息，Kafka進行消息隊列分發，將數據推送到下方進行處理加工操作。

Spark Streaming主要是用來抽取數據，對數據進行并行計算處理。

Spark Streaming是基于Spark上用于處理實時計算業務的框架，其實現是把輸入的流數據進行切分，切分的數據塊用批處理方式進行并行計算處理。

基本的分布式流處理框架包括數據接入層、消息緩存層、流處理業務層和集群服務［13］。Kafka主要作用于消息緩存層，Spark Streaming主要作用于流處理業務層。如圖2所示。

圖2 流數據處理基礎框架

從圖2可以看出，存儲過來的數據指將數據源加載到消息隊列的過程。對于Kafka來說，是生產者將數據載入消息隊列，并解耦數據的生產方和使用方，對數據進行消息緩存，重建分布式查詢系統，提供增量數據加載接口模式。通過這種情況可以分布到多個節點上，不同的數據節點，將其對應的數據源以消息的形式發送到對應的節點上。流處理業務層負責消費消息隊列中的數據，對這些數據進行分析處理并得到相應的結果，解決在數據載入過程中數據庫中可能存在的性能劣勢問題［14］。

2.3 數據表關聯數據庫

Spring Boot主要是用于多層架構體系的模型業務層，具有降低多層模塊間的耦合性、分層模塊化架構應用業務系統的優勢［15］。

保險大寬表系統通過Spring Boot和Spark Streaming實時消費Kafka中的數據，對數據做質控、關聯的操作，最終形成大寬表的數據。Spark Streaming從Kafka消息隊列中按照時間窗口不斷提取數據，然后進行批處理［16］，其主要是對保險大寬表系統里的保單數據進行統計分析。然后對處理的數據結果進行存儲，以保單號和時間字符串為key進行存儲，主要存儲到Redis中，如圖3所示。本系統中數據質控分析中的一些字典信息是通過Redis內存緩存來提高查詢分析的效率。在關聯過程中，首先判斷已有的Redis里是否已經存在這些數據信息，如果存在則直接調用，如果不存在，通過查找數據庫里面的內容并將其存入Redis且設置好TTL，方便下次可以直接在Redis中調用，而不需要再去數據庫中進行查找，提高了查詢效率。

圖3 Redis存儲數據

2.4 數據存儲

保險大寬表系統數據存儲采用MPP數據庫，其擁有海量的計算能力、容錯能力及優秀的擴展性。不僅如此，它還可以讓所有數據分布到每個節點上，使得每個節點去計算自己的部分數據，達到并行處理無需人工干預的目的，并可以通過增加通用硬件去擴充新的計算節點。

針對普通的節點，可以將一份數據分布在多個節點上，由此避免由于單個節點出現故障而導致數據丟失和數據不可用現象的發生，對于管理節點而言，一般也采用高可用設計，避免單點故障，提高了對數據查詢和分析的效率。

2.5 數據推送

將數據實時發送給系統前端，保險大寬表系統使用WebSocket協議作為服務化接口。Web?Socket作為一個獨立在TCP上的協議，本質是基于TCP為客戶端和服務器提供一種socket通信連接，使得客戶端和服務器端實現雙向通信［17］。對比傳統實時數據更新方案，WebSocket［18］可極大地減少網絡流量與延遲［19］。

保險大寬表系統中所有數據采用push方式進行推送，服務器接收到數據后會立即發送到客戶端上。客戶端和服務器之間通過進程創建基于WebSocket技術的通信連接，系統就能利用服務器的推送功能實現對保險數據指標的實時查看。

3 系統實現關鍵問題分析

3.1 分布式并發處理數據一致性問題

在采用分布式技術提高數據計算和系統本身的性能時，需要考慮到數據的一致性問題，比如同一份保單數據在很短的時間內先后發生兩次變更，兩次變更后的數據分別是U1和U2。在分布式環境下，U1和U2可能被隨機分配到不同機器的不同線程上執行，執行的順序可能會變成先U2后U1，這樣數據的一致性無法得到保證，導致目標數據庫的數據發生錯誤。

基于以上問題，保險大寬表系統利用Kafka同一個Topic的同一個partition順序消費且同一個Topic的同一個partition只能被同一個線程消費數據的特點。處理程序時，首先把OGG-Bigdata發送到Kafka的數據根據數據特征分類到同一張保單上，再根據保單編號為key通過shuffle動作發布到另外一個Topic，這樣就保證了同一個保單編號的數據肯定會分配到同一個partition，處理程序處理shuffle后的Topic，從而保證了數據的一致性。如圖4所示。

圖4 數據一致性處理過程

在保險大寬表系統中，通過利用Kafka實時收集數據并傳到Spark Streaming可進行實時分析，Spark Streaming可以將接收到的數據匯總成多個小數據集，Spark Streaming每次處理的是一個時間窗口的數據流，實質上是對這些數據進行批量的實時處理［20］，如圖5所示。這一特點可以很好的避免并發數據處理中頻繁的任務分配和調度問題，能達到次秒級延時的實時處理。

圖5 數據實時處理

3.2 從Or ac l e到MPP數據庫的實時同步問題

當數據從中轉數據庫導入到MPP數據庫中時，關鍵部分是要保證數據同步。MPP數據庫具備高性能、高可用和高擴展特性，但其對數據的增刪操作比較差，在并發量大的情況下，性能會更低。因為每天保單系統上都會有大量的變化數據產生，從源數據庫中采集而得到的數據直接導出文本在進行導入會占用很大的空間，耗時長，導致數據時效性差，不能達到數據實時同步。

保險大寬表系統通過利用Batch（批處理）操作，先定期刪除發生變化的數據，把這些變化的數據導出為CSV格式文件，通過load操作解決數據在增加和更改過程中效率慢的問題，實現數據的近實時變化。

4 結語

本研究通過分析目前保險系統中業務人員不能實時查看當前保單系統的保單信息和各個指標匯總數據，給出了一種基于分布式流計算實現一個流式大寬表數據系統，可以實時采集保單系統各項數據的變化。通過將Kafka和Spark Streaming結合的方式實時關聯多表數據，處理數據增刪改變化，最后存儲到MPP數據庫中，能夠讓保險業務人員實時查看最新保單數據信息及各個保單指標匯總信息。