分布式計算在券商營銷推廣活動中的應用

俞楓, 張忍, 鄭爽

(國泰君安證券股份有限公司, 上海 200120)

0 引言

隨著移動互聯(lián)網(wǎng)技術的普遍應用, 券商行業(yè)開發(fā)客戶從傳統(tǒng)的線下營銷推廣模式逐步轉變?yōu)榫€下、線上并行的營銷推廣模式。線上營銷推廣活動具有用戶群體大、時效性強、規(guī)則變動大等特點, 保證并提升用戶的參與體驗、減少或消除營銷推廣活動對股票交易業(yè)務的影響、減少營銷推廣活動開發(fā)周期、變更周期是進行日常性、大規(guī)模線上營銷推廣活動需要解決的根本技術問題。

目前券商行業(yè)普遍采用的營銷推廣活動后臺框架是恒生統(tǒng)一金融接入系統(tǒng)(UFX)服務節(jié)點和Oracle關系型數(shù)據(jù)庫, 該框架存在以下弊端。

1)當參與營銷推廣活動的用戶數(shù)目激增時將過度占用統(tǒng)一金融總線(目前券商業(yè)務大部分通過該消息總線接入和傳遞)的消息通道資源，此時股票交易客戶的交易事件和營銷推廣活動參與客戶的參與事件都會出現(xiàn)響應超時現(xiàn)象，影響用戶體驗乃至給客戶造成經(jīng)濟損失。

2)采用傳統(tǒng)的Oracle關系型數(shù)據(jù)庫，不同的營銷推廣活動因需要記錄的用戶參與狀態(tài)數(shù)據(jù)范式不同，需要預先定義不同的用戶參與狀態(tài)記錄表，并實現(xiàn)與之相對應的數(shù)據(jù)庫CRUD操作，活動難以實現(xiàn)動態(tài)、自由定義，拉長活動的開發(fā)、變更周期。采用傳統(tǒng)的Oracle關系型數(shù)據(jù)庫，需要額外的分庫分表中間件(如Mycat)以水平擴展數(shù)據(jù)容量，當用戶數(shù)目激增時難以快速擴展數(shù)據(jù)庫節(jié)點。

本文采用分布式流處理架構[1]，如圖1所示。

圖1 系統(tǒng)架構圖

消息隊列中間件Kafka接收線上營銷推廣活動中各種渠道來源的用戶操作事件，Storm分布式計算框架并行處理用戶操作事件，MongoDB存儲活動規(guī)則，記錄用戶的活動參與狀態(tài)、用戶的獎勵狀態(tài)、事件處理狀態(tài)，Zookeeper分布式應用協(xié)調(diào)服務觸發(fā)Storm節(jié)點動態(tài)加載活動規(guī)則。

采用該技術方案具有支持并發(fā)用戶數(shù)大、事件流并發(fā)處理能力強、 營銷推廣活動可以模板化配置并動態(tài)生效、 不同活動用戶記錄可以同一個數(shù)據(jù)表記錄且無須預先定義字段、基本不占用統(tǒng)一金融總線資源、具備消息緩沖能力、無須額外中間件便可分庫分表等特性。

Apache Storm是目前主流的流式分布式計算框架之一，它是由Twitter開發(fā)，阿里巴巴發(fā)展的開源軟件。Storm可以在工作進程、線程、任務3個層級進行擴展以提升并行計算能力，同時Storm還具有高可用、高容錯特性以及完整的消息處理確認機制[2][3]。

Apache Kafka是目前主流的高吞吐量分布式發(fā)布訂閱消息系統(tǒng)之一，他是由LinkedIn開發(fā)的開源軟件。Kafka具有輕量級、分布式、可分區(qū)、分區(qū)多備份等技術特點，具有高可用、高可靠、高吞吐量等性能特點[4][5]。

MongoDB是基于分布式文件系統(tǒng)的非關系型數(shù)據(jù)庫。MongoDB是基于文檔存儲的，文檔的數(shù)據(jù)結構非常松散，類似于json數(shù)據(jù)格式的bjson格式。在使用MongoDB時無須預定義表字段，表里面的每條記錄可以具有不同的數(shù)據(jù)字段[6][7]。MongoDB分片副本集是高可用、易擴展的分庫分表數(shù)據(jù)庫方案，提升數(shù)據(jù)存儲能力及數(shù)據(jù)操作性能，同時不需要使用額外的數(shù)據(jù)庫中間件][8][9]。

Apache ZooKeeper是一個分布式的分布式應用程序協(xié)調(diào)服務，是Hadoop大數(shù)據(jù)生態(tài)圈的基礎設施之一。它可以在分布式計算框架Storm中作為計算節(jié)點的配置信息同步協(xié)調(diào)者[10]。

1 營銷推廣活動的特征抽取及規(guī)則原型的定義

1.1 規(guī)則原型提取

通過對常見的營銷推廣活動的規(guī)則進行分析，歸納總結出簡單營銷推廣活動通用性強的規(guī)則，比如以下幾個規(guī)則：

1)首次做某事送獎勵，比如首次注冊APP事件、首次上傳頭像事件等。

2)累計做某個動作達到N次即發(fā)放獎勵，比如累計簽到N天事件等。

3)每周期做某事，送獎勵，比如每天簽到事件等。

4)某指標在某范圍內(nèi)送獎勵，比如賬戶資產(chǎn)超過某值事件，充值超過某值事件。

對簡單活動的通用規(guī)則抽象是實現(xiàn)營銷推廣活動模板化配置的前提。規(guī)則原型的定義含有原型ID、原型名稱、觸發(fā)事件列表、具體事件處理類、用戶狀態(tài)記錄字段定義信息，如表1所示。

表1中具體事件處理類是與觸發(fā)事件列表無關的處理類，比如每周期做某事的處理類、首次做某事的處理類。觸發(fā)事件列表包含了可以觸發(fā)該原型的用戶操作事件，比如用戶登錄APP。用戶狀態(tài)記錄字段主要是記錄用戶發(fā)生操作事件后事件的處理結果，比如首次登錄APP事件，將記錄首次登錄APP的時間，送過對應獎勵的標志。實際開發(fā)過程中只有具體事件處理類是需要開發(fā)的。記錄字段、規(guī)則入?yún)ⅰ㈥P聯(lián)動作等都是通過配置實現(xiàn)。具體事件處理類通過抽象可以高度復用，比如處理首次事件的處理類，可以復用于首次注冊APP、首次設置昵稱、首次上傳頭像等規(guī)則原型，極大地減少了代碼開發(fā)量、縮短了上線周期。規(guī)則原型配置可以在多個營銷推廣活動中復用，比如首次登錄APP事件可以在多個APP促活活動中重復使用。

表1 規(guī)則原型表

1.2 規(guī)則原型實例化快速構建活動規(guī)則

選擇營銷推廣活動所需要的規(guī)則原型，快速構建活動處理規(guī)則，比如APP簽到活動，可以選擇每次簽到規(guī)則原型、累計簽到規(guī)則原型、連續(xù)簽到規(guī)則原型，設置每個實例規(guī)則的獎勵條件(如累計簽到多少天)和獎勵類型、數(shù)目，設置活動開始、結束日期、面向的客戶范圍等必要信息就能產(chǎn)生一個活動完整的規(guī)則配置。規(guī)則具體處理類、規(guī)則原型高度復用可以減少上線新營銷活動所需的開發(fā)測試時間，同時活動規(guī)則參數(shù)、活動規(guī)則獎勵支持動態(tài)自由配置。

2 技術架構、組件詳解

本系統(tǒng)的組件如圖1所示，用戶參與事件流轉如圖2所示。

圖2 用戶事件流轉流程圖

2.1 MongoDB副本集

MongoDB主要作用是存儲規(guī)則原型定義、活動規(guī)則定義、事件處理狀態(tài)、用戶的活動參與狀態(tài)、用戶獎勵數(shù)據(jù)及發(fā)放狀態(tài)等。如果采用傳統(tǒng)的關系型數(shù)據(jù)庫則每次創(chuàng)建新活動時要創(chuàng)建不同的用戶參與狀態(tài)記錄表，因為每個活動需要記錄的活動參與狀態(tài)數(shù)據(jù)字段是不一樣的，比如簽到活動和新開戶用戶活動，一個活動需要記錄的是簽到狀態(tài)，一個需要記錄的新開戶狀態(tài)，兩者在表結構上并不兼容。采用MongoDB的優(yōu)點是MongoDB數(shù)據(jù)表中的字段無須預先定義，同一個表里可以容納數(shù)據(jù)字段完全不同的記錄，通過活動ID和用戶ID作為MongoDB用戶狀態(tài)表的索引和公共字段，可以快速檢索到每個活動下每個用戶的參與狀態(tài)，而用戶的參與狀態(tài)記錄字段通過規(guī)則配置已經(jīng)確定，無須再額外定義。為了滿足大量線上用戶同時參與活動的需求，MongoDB集群采用分片副本集群如圖3所示。

圖3 MongoDB分片副本集

分片類似于傳統(tǒng)關系型數(shù)據(jù)庫的分庫分表，可以提升數(shù)據(jù)容量，副本集通過數(shù)據(jù)多節(jié)點備份保證數(shù)據(jù)高可用，并可主從讀寫分離提升讀寫性能。MongoS節(jié)點主要負責數(shù)據(jù)路由，MongoC節(jié)點主要負責維持數(shù)據(jù)分片信息，MongoD節(jié)點主要負責存儲數(shù)據(jù)，客戶端Client通過訪問MongoS從MongoC中獲取數(shù)據(jù)所在數(shù)據(jù)副本集并從對應的MongoD數(shù)據(jù)副本集中獲取、存儲數(shù)據(jù)。以活動ID、用戶ID作為數(shù)據(jù)分片依據(jù)，將用戶參與狀態(tài)、用戶獎勵數(shù)據(jù)分散到各數(shù)據(jù)副本集中，可以控制每個數(shù)據(jù)副本集的數(shù)據(jù)量，提高系統(tǒng)的數(shù)據(jù)容量、數(shù)據(jù)讀寫速度，以滿足互聯(lián)網(wǎng)化的應用需求。

2.2 Kafka消息隊列

Kafka消息通道主要作用是采用異步消息發(fā)送模式將營銷推廣活動的用戶操作事件快速從統(tǒng)一金融總線剝離，減少營銷活動對統(tǒng)一金融總線資源的占用，使其更專注于證券交易核心業(yè)務。利用Kafka的高吞吐量、高可用特性以滿足互聯(lián)網(wǎng)環(huán)境下大量用戶同時產(chǎn)生的操作事件消息的及時投遞。對各種觸發(fā)事件采用統(tǒng)一的Kafka通道和消息格式可以減少各種操作事件消息的對接時間，便于各類型消息的統(tǒng)一校驗和備份。如圖4所示。

圖4 Kafka分布式消息隊列

Kafka Producer 異步發(fā)送消息到Kafka 集群，Kafka Consumer(Storm Spout)從Kafka消費消息。Producer和Consumer互不干擾，Consumer未能及時消費消息并不會阻塞Producer發(fā)送消息，Kafka會將未消費的消息存儲到文件系統(tǒng)，Consumer按自己的消費能力從Kafka拉取消息，所以Kafka在本系統(tǒng)中還起到訪問壓力緩沖的功能。目前對接的事件包含注冊、開戶、登錄、簽到、邀請、點擊頁面、賬戶操作等事件。

2.3 Storm分布式計算框架

Storm流式分布式計算框架分為三級處理節(jié)點如圖5，第1級 KafkaSpout按分區(qū)從Kafka獲取數(shù)據(jù)。第 2級 MsgParseBolt進行Kafka消息格式校驗、并轉換為特定數(shù)據(jù)格式。第 3級 MsgProcessBolt匹配活動ID和觸發(fā)事件，找出對應的處理規(guī)則，判斷是否滿足規(guī)則，記錄事件處理狀態(tài)和用戶參與狀態(tài)并決定是否產(chǎn)生獎勵流水。MsgParseBolt轉換消息格式后會將屬于同一個用戶的操作事件發(fā)送到同一個ProcessBolt 進行串行處理，這樣可以減少同一用戶數(shù)據(jù)并發(fā)操作時的鎖等待時間，同時也可避免少量客戶通過未知方式攻擊系統(tǒng)時造成系統(tǒng)完全不可用的可能性，如圖5客戶U1、U2的操作事件EA、EB達到MsgProcessBolt時同一個客戶的事件由同一個MsgProcessBolt處理。MsgProcessBolt 處理完用戶操作事件會發(fā)確認消息給Spout，當Spout未確認處理的用戶操作事件超過一定數(shù)目時，Spout會停止從Kafka拉取數(shù)據(jù)，避免Storm內(nèi)部各節(jié)點消息堆積造成內(nèi)存占用過多等問題。

Storm的三級處理節(jié)點可以按照需要獨立靈活配置所需節(jié)點數(shù)目，可以在進程、線程、任務等多個層級進行配置，可根據(jù)用戶流量快速進行系統(tǒng)擴容以滿足互聯(lián)網(wǎng)環(huán)境下大量用戶參與活動的需要。

2.4 Zookeeper分布式節(jié)點同步服務

Zookeeper主要作用是協(xié)調(diào)Storm分布式計算節(jié)點間的數(shù)據(jù)同步。活動配置信息是Storm拓撲啟動時加載到內(nèi)存的，當有活動配置變更時，通過Zookeeper告知Storm各節(jié)點重新加載活動配置，保證不用停止Storm計算拓撲也可以重新加載規(guī)則并保證各計算節(jié)點活動配置一致。如圖5所示。

圖5 分布式計算框架Storm

3 系統(tǒng)性能測試

3.1 Kafka消息接收能力測試

對Kafka性能測試結果，如表2所示。

可以發(fā)現(xiàn)采用異步模式能提升Kafka的消息接收能力，本文采用了該種模式，各種來源的用戶操作事件統(tǒng)一成Kafka消息格式并異步發(fā)送到多分區(qū)的Kafka消息隊列上由Storm統(tǒng)一消費并處理。采用多分區(qū)是為了提升Storm Spout節(jié)點的并發(fā)處理能力，Storm Spout的數(shù)目和Kafka分區(qū)的數(shù)目保持一致時能使Spout的性能最優(yōu)，即不會有Storm Spout空閑也不會有Storm Spout要處理多個分區(qū)，每個Storm Spout專注處理一個分區(qū)。消息隊列3個分區(qū)可以由3個Storm Spout并發(fā)處理消息，增加Storm第1級的并發(fā)處理能力。

表2 Kafka性能測試

3.2 MongoDB 性能測試

MongoDB性能測試結果，如表3所示。

表3 MongoDB數(shù)據(jù)庫性能測試

通過上述測試可以發(fā)現(xiàn)分片副本集INSERT，UPDATE性能上優(yōu)于普通副本集，因此本文采用的時MongoDB分片副本集。

3.3 Storm性能測試

測試不同Storm 節(jié)點下的Storm的消息處理能力。如表4所示。

表4 Storm性能測試

目前而言，本系統(tǒng)的峰值壓力每秒1000個用戶操作事件，通過測試可以發(fā)現(xiàn)目前系統(tǒng)架構能夠承受該峰值壓力。

3.4 傳統(tǒng)架構和分布式流計算架構性能比較

傳統(tǒng)架構和分布式流計算架構的性能對比(一個用戶操作事件如圖2觸發(fā)1次消息處理狀態(tài)查詢、1次用戶狀態(tài)查詢、1次用戶狀態(tài)修改、1次消息處理狀態(tài)修改，1次獎勵庫存查詢，1次獎勵發(fā)放流水新增共6個數(shù)據(jù)庫操作)，如表5所示。

表5 架構性能測試

可以得出結論分布式流計算架構對總線通道的占用時間遠小于傳統(tǒng)的金融總線服務節(jié)點+Oracle架構的占用時間，極大地減少了營銷推廣活動對核心交易業(yè)務產(chǎn)生影響的可能性。在事件處理性能上，分布式流計算架構每秒處理用戶操作事件數(shù)顯著高于傳統(tǒng)架構，且當用戶操作事件數(shù)高于分布式流計算架構處理能力時，Kafka可以起到消息緩沖作用，操作事件并不會堆積在金融總線隊列上，而使用金融總線服務節(jié)點是沒有緩沖機制的，超過處理能力將導致事件堆積在金融總線隊列上，可能影響正常交易。

4 總結

本文通過歸納總結一般營銷推廣活動常見的用戶事件類型，演化出通用性強的活動規(guī)則原型，并基于MongoDB數(shù)據(jù)表無模式的特點形成可以動態(tài)配置無須預定義用戶參與狀態(tài)記錄表結構的活動規(guī)則。通過使用Kafka異步多分區(qū)模式承接營銷活動中大量用戶同時產(chǎn)生的操作事件并起到緩沖作用，通過使用分布式Storm計算框架和MongoDB 分片副本集提升實時流數(shù)據(jù)處理能力，并可根據(jù)需要快速水平擴展計算能力，通過Zookeeper協(xié)調(diào)Storm分布式計算節(jié)點進行活動配置同步，實現(xiàn)活動配置的動態(tài)設置和動態(tài)生效并保證各節(jié)點活動配置一致，經(jīng)性能測試本系統(tǒng)能夠承受目前券商公司線上營銷推廣活動的用戶并發(fā)操作壓力。