《基于Lambda架構的股票市場事件處理引擎實踐》要點：
本文介紹了基于Lambda架構的股票市場事件處理引擎實踐，希望對您有用。如果有疑問，可以聯系我們。

CEP(Complex Event Processing)是證券行業很多業務應用的重要支撐技術.CEP的概念本身并不新鮮,相關技術已經被運用超過15年以上,但是證券界肯定是運用CEP技術最為充分、最為前沿的行業之一,從算法交易(algorithmic trading)、風險管理(risk management)、關鍵時刻管理(Moment of Truth – MOT)、委托與流動性分析(order and liquidity analysis)到量化交易(quantitative trading)乃至向投資者推送投資信號(signal generation)等等,不一而足.

CEP技術通常與Time-series Database(時序數據庫)結合,最理想的解決方案是CEP技術平臺向應用提供一個歷史序列(historical time-series)與實時序列(real-time series)無差異融合的數據流連續體(continuum)- 對于證券類應用而言,昨天、上周、上個月的數據不過是當下此刻數據的延續,而處理算法卻是無邊際的 – 只要開發者能構想出場景與模型.

廣發證券的IT研發團隊,一直關注Storm、Spark、Flink等流式計算的開源技術,也經歷了傳統Lambda架構的技術演進,在Kappa架構的技術尚未成熟之際,團隊針對證券行業的技術現狀與特點,采用改良的Lambda架構實現了一個CEP引擎,本文介紹了此引擎的架構并分享了一些股票業務較為有趣的應用場景,以饗同好.

隨著移動互聯和物聯網的到來,大數據迎來了高速和蓬勃發展時期.一方面,移動互聯和物聯網產生的大量數據為孕育大數據技術提供了肥沃的土壤；一方面,各個公司為了應對大數據量的挑戰,也急切的需要大數據技術解決生產實踐中的問題.短時間內各種技術層出不窮,在這個過程中Hadoop脫穎而出,并營造了一個豐富的生態圈.雖然大數據一提起Hadoop,好像有點老生常談,甚至覺得這個技術已經過時了,但是不能否認的是Hadoop的出現確實有非凡的意義.不管是它分布式處理數據的理念,還是高可用、容錯的處理都值得好好借鑒和學習.

剛開始,大家可能都被各種分布式技術、思想所吸引,一頭栽進去,掉進了技術的漩渦,不能自拔.一方面大數據處理技術和系統確實復雜、繁瑣；另一方面大數據生態不斷的推陳出新,新技術和新理念層出不窮,確實讓人目不暇接.如果想要把生態圈中各個組件玩精通確實不是件容易的事情.本人一開始也是深陷其中,皓首窮經不能自拔.但騰出時間,整理心緒,回頭反顧,突然有種釋然之感.大數據并沒有大家想象的那么神秘莫測與復雜,從技術角度看無非是解決大數據量的采集、計算、展示的問題.

因此本文參考Lambda/Kappa架構理念,提出了一種有行業針對性的實現方法.盡量讓系統層面更簡單,技術更同構,初衷在讓大家聚焦在大數據業務應用上來,從而真正讓大數據發揮它應有的價值.

1、 背景

Lambda架構是由Storm的作者Nathan Marz 在BackType和Twitter多年進行分布式大數據系統的經驗總結提煉而成,用數學表達式可以表示如下：

batch view = function(all data)
realtime view = function(realtime view,new data)
query = function(batch view .realtime view)

邏輯架構圖如下：

從圖上可以看出,Lambda架構主要分為三層：批處理層,加速層和服務層.它整合了離線計算和實時計算,融合了不可變性(immutable),讀寫分離和復雜性隔離等一系列架構原則設計而成,是一個滿足大數據系統關鍵特性的架構.Nathan Marz認為大數據系統應該具有以下八個特性,Lambda都具備它們分別是：

• Robustness and fault tolerance(魯棒性和容錯性)
• Low latency reads and updates(讀和更新低延時)
• Scalability(可伸縮)
• Generalization(通用性)
• Extensibility(可擴展)
• Ad hoc queries(可即席查詢)
• Minimal maintenance(易運維)
• Debuggability(可調試)

由于Lambda架構的數據是不可變的(immutable),因此帶來的好處也是顯而易見的：

• Human-fault tolerance(對人為的容錯性)：數據流水被按時序記錄下來,而且數據只寫一次,不做更改,而不像RDBMS只是保留最后的狀態.因此不會丟失數據信息.即使平臺升級或者計算程序中不小心出現Bug,修復Bug后重新計算就好.強調了數據的重新計算問題,這個特性對一個生產的數據平臺來說是十分重要的.
• Simplicity(簡易性)：可變的數據模型一般要求數據能必須被索引,以便于數據可被再次被檢索到和可以被更新.但是不變的數據模型相對來說就很簡單了,只是一味的追加新數據即可.大大簡化了系統的復雜度.

但是Lambda也有自身的局限性,舉個例子：在大數據量的情況下,要即席查詢過去24小時某個網站的pv數.根據前面的數學表達式,Lambda架構需要實現三部分程序,一部分程序是批處理程序,比如可能用Hive或者MapReduce批量計算最近23.5個小時pv數,一部分程序是Storm或Spark Streaming流式計算程序,計算0.5個小時內的pv數,然后還需要一個服務程序將這兩部分結果進行合并,返回最終結果.因此Lambda架構包含固有的開發和運維的復雜性.

因為以上的缺陷,Linkedin的Jay Kreps在2014年7月2日在O’reilly《Questioning the Lambda Architecture》提出了Kappa架構,如下圖：

Kappa在Lambda做的最大的改進是用同一套實時計算框架代替了Lambda的批處理層,這樣做的好處是一套代碼或者一套技術?？梢越鉀Q一個問題.它的做法是這樣的：

1. 用Kafka做持久層,根據需求需要,用Kafka保留需要重新計算的歷史數據長度,比如計算的時候可能用30天的數據,那就配置Kafka的值,讓它保留最近30天的數據.
2. 當你程序因為升級或者修復了缺陷,需要重新計算的時候,就再啟一個流式計算程序,從你所需的Offset開始計算,并將結果輸入到一個新的表里.
3. 當這個流式計算程序追平第一個程序的時候,將應用切換到第二個程序的輸出上.
4. 停止第一個程序,刪除第一個程序的輸出結果表.

這樣相當于用同一套計算框架和代碼解決了Lambda架構中開發和運維比較復雜的問題.當然如果數據量很大的情況下,可以增加流式計算程序的并發度來解決速度的問題.

2、 廣發證券Lambda架構的實現

由于金融行業在業務上受限于T+1交易,在技術上嚴重依賴關系型數據庫(特別是Oracle).在很多場景下,數據并不是以流的形式存在的,而且數據的更新頻率也并不是很實時.比如為了做技術面分析的行情數據,大多數只是使用收盤價和歷史收盤價(快照數據)作為輸入,來計算各類指標,產生買賣點信號.

因此這是一個典型的批處理的場景.另一方面,比如量化交易場景,很多實時的信號又是稍縱即逝,只有夠實時才存在套利的空間,而且回測和實盤模擬又是典型的流處理.鑒于以上金融行業特有的場景,我們實現了我們自己的架構(GF-Lambda),它介于Lambda和Kappa之間.一方面能夠滿足我們處理數據的需求；一方面又可以達到技術上的同構,減少開發運維成本.根據對數據實時性要求,將整個計算部分分為三類：

• Spark SQL：代替MapReduce或者Hive的功能,實現數據的批量預處理；
• Spark Streaming：近實時高吞吐的mini batching數據處理功能；
• Storm：完成實時的流式數據處理；

GF-Lambda的優勢如下：

• 在PipeLine的驅動方面,采用Airbnb開源的Airflow,Airflow使用腳本語言來實現整個PipeLine的定義,而且任務實例也是動態生成的；相比Oozie和Azkaban采用標記語言來完成PipeLine的定義,Airflow的優勢是顯而易見的,例如：

  整個data flow采用腳本編寫,便于配置管理和升級.而Oozie只能使用XML定義,升級遷移成本較大.

  觸發方式靈活,整個PipeLine可以動態生成,切實的做到了“analytics as a service”或者 “analysis automation”.

• 另外一個與Lambda或者Kappa最大的不同之處是我們采用了Redis作為緩存來存儲各個計算服務的狀態；雖然Spark和Storm都有Checkpoint機制,但是CheckPoint會影響到程序復雜度和性能,并且以上兩種技術的CheckPoint機制并不是很完善.通過Redis和Kafka的Offset機制,不僅可以做到無狀態的計算服務,而且即使升級或者系統故障,數據的可用性也不會受到影響.
• 整個batch layer采用Spark SQL,使用Spark SQL的好處是能做到密集計算的后移.由于歷史原因,券商Oracle等關系型數據庫使用比較多,而且在開市期間數據庫壓力也比較大,此處的Spark SQL只是不斷的從Oracle批量加載數據(除了Filter基本在Oracle上做任何計算)或者主動的通過Oracle日志旁錄數據,對數據庫壓力較小,同時又能達到數據準實時性的要求；另外所有的計算都后置到Yarn集群上進行,不僅利于程序的運維,也利于資源的有效管控和伸縮.架構實現如下圖所示：

3、 應用場景

CEP在證券市場的應用的有非常多,為了讀者更好的理解上述技術架構的設計,在此介紹幾個典型應用場景.

1)自選股到價和漲跌幅提醒

自選股到價和漲跌幅提醒是股票交易軟件的一個基礎服務器,目的在于方便用戶簡單、及時的盯盤.其中我們使用MongoDB來存儲用戶的個性化設置信息,以便各類應用可以靈活的定制自身的Schema.在功能上主要包括以下幾種：

• 股價高于設定值提醒.
• 股價低于設定值提醒.
• 漲幅高于設定值提醒.
• 一分鐘、五分鐘漲幅高于設定值提醒.
• 跌幅高于設定值提醒.
• 一分鐘、五分鐘跌幅高于設定值提醒.

主要的挑戰在于大數據量的實時計算,而采用GF-Lambda可以輕松解決這個問題.數據處理流程如下：

首先從Kafka訂閱實時行情數據并進行解析,轉化成RDD對象,然后再衍生出Key(market+stockCode),同時從Mongo增量加載用戶自選股預警設置數據,然后將這兩份數據進行一個Join,再分片對同一個Key的兩個對象做一個Filter,產生出預警信息,并進行各個終端渠道推送.

2)自選股實時資訊

實時資訊對各類交易用戶來說是非常重要的,特別是和自身嚴重相關的自選股實時資訊.一個公告、重大事項或者關鍵新聞的出現可能會影響到用戶的投資回報,因此這類事件越實時,對用戶來說價值就越大.

在GF-Lambda平臺上,自選股實時資訊主要分為兩部分：實時資訊的采集及預處理(適配)、資訊信息與用戶信息的撮合.整個處理流程如下圖所示：

在上圖分割線左側是實時資訊的預處理部分,首先使用Spark JDBC接口從Oracle數據庫加載數據到Spark,形成DataFrame,再使用Spark SQL的高級API做數據的預處理(此處主要做表之間的關聯和過濾),最后將每個Partition上的數據轉化成協議要求的格式,寫入Kafka中等待下游消費.

左側數據ETL的過程是完全由Airflow來進行驅動調度的,而且每次處理完就將狀態cache到Redis中,以便下次增量處理.在上圖的右側則是與用戶強相關的業務邏輯,將用戶配置的信息與實時資訊信息進行撮合匹配,根據用戶設置的偏好來產生推送事件.

此處用Kafka來做數據間的解耦,好處是不言而喻的.首先是保證了消息之間的靈活性,因為左側部分產生的事件是一個基礎公共事件,而右側才是一個與業務緊密耦合的邏輯事件.基礎公共事件只有事件的基礎屬性,是可以被很多業務同時訂閱使用的.

其次從技術角度講左側是一個類似批處理的過程,而右側是一個流處理的過程,中間通過Kafka做一個轉換與對接.這個應用其實是很具有代表性的,因為在大部分情況下,數據源并不是以流的形式存在,更新的頻率也并不是那么實時,所以大多數情況下都會涉及到batch layer與speed layer之間的轉換對接.

3)資金流選股策略

上面兩個應用相對來說處理流程比較簡單,以下這個case是一個業務
稍微繁瑣的CEP應用-資金流策略交易模型,該模型使用資金流流向來判斷股票在未來一段時間的漲跌情況.它基于這樣一個假設,如果是資金流入的股票,則股價在未來一段時間上漲是大概率事件；如果是資金流出的股票,則股價在未來一段時間下跌是大概率事件.那么我們可以基于這個假設來構建我們的策略交易模型.如下圖所示,這個模型主要分為三部分：

1)個股資金流指標的實時計算

由于涉及到一些業務術語,這里先做一個簡單的介紹.

資金流是一種反映股票供求關系的指標,它的定義如下：證券價格在約定的時間段中處于上升狀態時產生的成交額是推動指數上漲的力量,這部分成交額被定義為資金流入；證券價格在約定的時間段中下跌時的成交額是推動指數下跌的力量,這部分成交額被定義為資金流出；若證券價格在約定的時間段前后沒有發生變化,則這段時間中的成交額不計入資金流量.當天資金流入和流出的差額可以認為是該證券當天買賣兩種力量相抵之后,推動價格變化的凈作用量,被定義為當天資金凈流量.數量化定義如下：

其中,Volume為成交量,為i時刻收盤價,為上一時刻收盤價.

嚴格意義上講,每一個買單必須有一個相應的賣單,因此真實的資金流入無法準確的計算,只能通過其他替代方法來區分資金的流入和流出,通過高頻數據,將每筆交易按照驅動股價上漲和下跌的差異,確定為資金的流入或流出,最終匯聚成一天的資金流凈額數據.根據業界開發的CMSMF指標,采用高頻實時數據進行資金流測算,主要出于以下兩方面考慮：一是采用高頻數據進行測算,可以盡可能反映真實的市場信息；二是采取報價(最近買價、賣價)作為比較基準,成交價大于等于上期最優賣價視為流入,成交價小于等于上期最優買價視為流出.

除了資金的流入、流出、凈額,還有一系列衍生指標,比如根據流通股本數多少衍生出的大、中、小單流入、流出、凈額,及資金流信息含量(IC)、資金流強度(MFP),資金流杠桿倍數(MFP),在這里就不一一介紹.

從技術角度講,第一部分我們通過訂閱實時行情信息,開始計算當天從開市到各個時刻點的資金流入、流出的累計值,及衍生指標,并將這些指標計算完成后重新寫回到Kafka進行存儲,方便下游消費.因此第一部分完全是一個大數據量的實時流處理應用,屬于Lambda的speed layer.

2)買賣信號量的產生及交易

第二部分在業務上屬于模型層,即根據當前實時資金流指標信息,構建自己的策略模型,輸出買賣信號.比如以一個簡單的策略模型為例,如果同時滿足以下三個條件產生買的信號.反之,產生賣的信號：

• (大單資金流入-大單資金流出>0) && (中單資金流入-中單資金流出>0)
• 大單的資金流信息含量>50%
• 大單的資金流強度>20%

在技術上,這個應用也屬于Lambda的 speed layer,通過訂閱Kafka中的資金流指標,根據上面簡單的模型,不斷的判斷是否要買或者賣,并調用接口發起買賣委托指令,最后根據回報結果操作持倉表或者成交表.(注意此處在業務上只是以簡單的模型舉例,沒有涉及到更多的細節)

3)持倉盈虧實時追蹤及交易

第三部分在業務上主要是準實時的盈虧計算.在技術層面,屬于Lambda 的batch layer.通過訂閱實時行情和加載持倉表/成交表,實時計算用戶的盈虧情況.當然此處還有一些簡單的止損策略,也可以根據盈利情況,發起賣委托指令,并操作持倉表和成交表.最后將盈利情況報給服務層,進行展示或者提供回調接口.詳細的處理流程如下圖所示：

總結

正如文章前面強調的一樣,寫這篇文章的初衷是希望大家從大數據豐富的生態中解放出來,與業務深度的跨界融合,從而開發出更多具有價值的大數據應用,真正發揮大數據應有的價值.這和Lambda架構的作者Nathan Marz的理念也是十分吻合的,記得他還在BackType工作的時候,他們的團隊才五個人,卻開發了一個社會化媒體分析產品——在100TB的數據上提供各種豐富的實時分析,同時這個小的團隊還負責上百臺機器的集群的部署、運維和監控.

當他向別人展示產品的時候,很多人都很震驚他們只有五個人.經常有人問他：“How can so few people do so much?”.他的回答是：“It’s not what we’re doing, but what we’re not doing.”通過使用Lambda架構,他們避免了傳統大數據架構的復雜性,從而產出變得非常顯著.

在五花八門的大數據技術層出不窮的當下,Marz的理念更加重要.我們一方面需要與時俱進關注最新的技術進步 – 因為新技術的出現可能反過來讓以前沒有考慮過或者不敢想的應用場景變成可能,但另一方面更重要的是,大數據技術的合理運用需要建立在對行業領域知識深刻理解的基礎上.大數據是金融科技的核心支撐技術之一,我們將持續關注最前沿的大數據技術與架構理念,持續優化最符合金融行業特點的解決方案,構建能放飛業務專家專業創新能力的技術平臺.

作者介紹

鄧昌甫,畢業于中山大學,廣發證券IT研發工程師,一直從事大數據平臺的架構、及大數據應用的開發、運維和敏捷相關工具的引入和最佳實踐的推廣(Git/Jenkins/Gerrit/Zenoss).

文章出處：聊聊架構(訂閱號ID:archtime)

轉載請注明本頁網址：
http://www.fzlkiss.com/jiaocheng/4431.html