分布式計(jì)算在券商營銷推廣活動(dòng)中的應(yīng)用

俞楓, 張忍, 鄭爽

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

0 引言

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

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

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

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

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

圖1 系統(tǒng)架構(gòu)圖

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

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

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

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

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

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

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

1.1 規(guī)則原型提取

通過對(duì)常見的營銷推廣活動(dòng)的規(guī)則進(jìn)行分析，歸納總結(jié)出簡單營銷推廣活動(dòng)通用性強(qiáng)的規(guī)則，比如以下幾個(gè)規(guī)則：

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

2)累計(jì)做某個(gè)動(dòng)作達(dá)到N次即發(fā)放獎(jiǎng)勵(lì)，比如累計(jì)簽到N天事件等。

3)每周期做某事，送獎(jiǎng)勵(lì)，比如每天簽到事件等。

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

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

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

表1 規(guī)則原型表

1.2 規(guī)則原型實(shí)例化快速構(gòu)建活動(dòng)規(guī)則

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

2 技術(shù)架構(gòu)、組件詳解

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

圖2 用戶事件流轉(zhuǎn)流程圖

2.1 MongoDB副本集

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

圖3 MongoDB分片副本集

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

2.2 Kafka消息隊(duì)列

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

圖4 Kafka分布式消息隊(duì)列

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

2.3 Storm分布式計(jì)算框架

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

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

2.4 Zookeeper分布式節(jié)點(diǎn)同步服務(wù)

Zookeeper主要作用是協(xié)調(diào)Storm分布式計(jì)算節(jié)點(diǎn)間的數(shù)據(jù)同步?；顒?dòng)配置信息是Storm拓?fù)鋯?dòng)時(shí)加載到內(nèi)存的，當(dāng)有活動(dòng)配置變更時(shí)，通過Zookeeper告知Storm各節(jié)點(diǎn)重新加載活動(dòng)配置，保證不用停止Storm計(jì)算拓?fù)湟部梢灾匦录虞d規(guī)則并保證各計(jì)算節(jié)點(diǎn)活動(dòng)配置一致。如圖5所示。

圖5 分布式計(jì)算框架Storm

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

3.1 Kafka消息接收能力測試

對(duì)Kafka性能測試結(jié)果，如表2所示。

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

表2 Kafka性能測試

3.2 MongoDB 性能測試

MongoDB性能測試結(jié)果，如表3所示。

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

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

3.3 Storm性能測試

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

表4 Storm性能測試

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

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

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

表5 架構(gòu)性能測試

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

4 總結(jié)

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