2018剛過(guò)去，趁著春節(jié)放假對(duì)過(guò)去一年主導(dǎo)開發(fā)的項(xiàng)目做個(gè)梳理和總結(jié)
項(xiàng)目背景
平臺(tái)運(yùn)營(yíng)到一定階段屎慢，一定會(huì)累積大批量的用戶數(shù)據(jù)蒿往，這些用戶數(shù)據(jù)是運(yùn)營(yíng)人員的黃金財(cái)產(chǎn)。而如何利用用戶的數(shù)據(jù)來(lái)做運(yùn)營(yíng)（消息推送氯庆、觸達(dá)消息、優(yōu)惠券發(fā)送扰付、廣告位等）堤撵，正是精準(zhǔn)運(yùn)營(yíng)系統(tǒng)需要解決的問(wèn)題。本文是基于信貸業(yè)務(wù)實(shí)踐后寫出來(lái)的羽莺，其它行業(yè)如保險(xiǎn)实昨、電商、航旅盐固、游戲等也可以參考荒给。
業(yè)務(wù)場(chǎng)景
先看幾個(gè)具有代表性的需求
用戶可用額度在20000～50000元，而且有借款記錄刁卜，未還本金為0志电，性別為“男”
用戶發(fā)生了A行為且未還本金大于5000
用戶在1天內(nèi)發(fā)生A行為次數(shù)大于等于3次
用戶在A行為前24小時(shí)內(nèi)未發(fā)生B行為
用戶在A行為后一個(gè)月內(nèi)未發(fā)生B行為
業(yè)務(wù)上有兩種消息類型
- 日常消息：由業(yè)務(wù)人員通過(guò)條件篩選鎖定用戶群，定時(shí)或即時(shí)給批量用戶發(fā)送消息或者優(yōu)惠券
- 觸達(dá)消息：主要由用戶自身的行為觸發(fā)蛔趴，比如登陸挑辆、進(jìn)件申請(qǐng)、還款等孝情，滿足一定篩選條件實(shí)時(shí)給用戶發(fā)送消息或優(yōu)惠券
對(duì)于用戶篩選條件鱼蝉，也主要有兩種類型
- 用戶狀態(tài)：包括用戶自身屬性如性別、年齡箫荡、學(xué)歷魁亦、收入等，還有用戶相關(guān)聯(lián)實(shí)體如進(jìn)件訂單羔挡、賬戶信息洁奈、還款計(jì)劃、優(yōu)惠券等的屬性婉弹，以及用戶畫像數(shù)據(jù)如行為偏好睬魂、進(jìn)件概率等
- 用戶行為：即用戶的動(dòng)作，包括登陸镀赌、進(jìn)件申請(qǐng)氯哮、還款，甚至前端點(diǎn)擊某個(gè)按鈕、在某個(gè)文本框輸入都算
早期方案
早期方案存在以下痛點(diǎn)
- 至少兩次跨部門溝通配合成本喉钢，周期被拉長(zhǎng)
- 非實(shí)時(shí)消息推送姆打，無(wú)法實(shí)現(xiàn)基于用戶行為的實(shí)時(shí)推送場(chǎng)景
- 非實(shí)時(shí)效果驗(yàn)證，無(wú)法及時(shí)調(diào)整運(yùn)營(yíng)策略
系統(tǒng)搭建的目標(biāo)
- 需要定義規(guī)則肠虽，提供可視化界面給業(yè)務(wù)人員動(dòng)態(tài)配置幔戏，無(wú)需重啟系統(tǒng)即使生效，減少溝通成本和避免重復(fù)開發(fā)税课，總之就是要更加 自動(dòng)化 和 易配置
- 采集實(shí)時(shí)數(shù)據(jù)闲延，根據(jù)實(shí)時(shí)事件做實(shí)時(shí)推送，總之就是要 實(shí)時(shí)
技術(shù)選型
數(shù)據(jù)采集韩玩、轉(zhuǎn)換垒玲、存儲(chǔ)
- 采集：狀態(tài)類的數(shù)據(jù)主要放在各個(gè)業(yè)務(wù)系統(tǒng)的關(guān)系型數(shù)據(jù)庫(kù)中，由于歷史原因有postgres和mysql找颓，需要實(shí)時(shí)采集表的數(shù)據(jù)變更合愈，這里使用kafka connector讀取mysql的binlog或postgres的xlog，另外還有標(biāo)簽系統(tǒng)計(jì)算出來(lái)的標(biāo)簽击狮，在kafka中佛析；而事件類數(shù)據(jù)主要來(lái)源于前端上報(bào)事件（有專門的服務(wù)接收再丟到kafka），關(guān)系型數(shù)據(jù)庫(kù)里面也可以提取一些事件彪蓬。
- 轉(zhuǎn)換：采集出來(lái)的數(shù)據(jù)需要做一些格式統(tǒng)一等操作寸莫，用kafka connector。
- 存儲(chǔ)：采用Elasticsearch存儲(chǔ)用戶數(shù)據(jù)档冬，ES查詢不像mysql或mongoDB用B-tree 或B+tree實(shí)現(xiàn)索引储狭，而是使用bitset和skip list來(lái)處理聯(lián)合索引，特別適合多字段的復(fù)雜查詢條件捣郊。
下面重點(diǎn)看下kafka connector和Elasticsearch如何使用
kafka connector
kafka connector有Source和Sink兩種組件，Source的作用是讀取數(shù)據(jù)到kafka慈参，這里用開源實(shí)現(xiàn)debezium來(lái)采集mysql的binlog和postgres的xlog呛牲。Sink的作用是從kafka讀數(shù)據(jù)寫到目標(biāo)系統(tǒng)，這里自己研發(fā)一套組件驮配，根據(jù)配置的規(guī)則將數(shù)據(jù)格式化再同步到ES娘扩。
kafka connector有以下優(yōu)點(diǎn)：
- 提供大量開箱即用的插件，比如我們直接用debezium就能解決讀取mysql和pg數(shù)據(jù)變更的問(wèn)題
- 伸縮性強(qiáng)壮锻，對(duì)于不同的connector可以配置不同數(shù)量的task琐旁，分配給不同的worker，猜绣，我們可以根據(jù)不同topic的流量大小來(lái)調(diào)節(jié)配置灰殴。
- 容錯(cuò)性強(qiáng)，worker失敗會(huì)把task遷移到其它worker上面
- 使用rest接口進(jìn)行配置掰邢，我們可以對(duì)其進(jìn)行包裝很方便地實(shí)現(xiàn)一套管理界面
Elasticsearch
對(duì)于狀態(tài)數(shù)據(jù)牺陶，由于狀態(tài)的寫操作相對(duì)較少伟阔，我們采取嵌套文檔的方式，將同個(gè)用戶的相關(guān)實(shí)體數(shù)據(jù)都同步寫入到同個(gè)文檔掰伸，具體實(shí)現(xiàn)用painless腳本做局部更新操作皱炉。效果類似這樣：
{
"id":123,
"age":30,
"credit_line":20000,
"education":"bachelor",
...
"last_loan_applications":{
"loan_id":1234,
"status":"reject",
...
}
...
}
事件數(shù)據(jù)寫入比較頻繁，數(shù)據(jù)量比較多狮鸭，我們使用父子文檔的方式做關(guān)聯(lián)合搅，效果類似這樣：
{
"e_uid":123,
"e_name":"loan_application",
"e_timestamp":"2019-01-01 10:10:00"
...
}
（e_前綴是為了防止同個(gè)index下同名字段沖突）
ES這樣存儲(chǔ)一方面是方便做統(tǒng)計(jì)報(bào)表，另一方面跟用戶篩選和觸達(dá)有關(guān)歧蕉。
規(guī)則引擎
在設(shè)計(jì)規(guī)則引擎前灾部，我們對(duì)業(yè)界已有的規(guī)則引擎，主要包括Esper, Drools, Flink CEP廊谓，進(jìn)行了初步調(diào)研梳猪。
Esper
Esper設(shè)計(jì)目標(biāo)為CEP的輕量級(jí)解決方案，可以方便的嵌入服務(wù)中蒸痹，提供CEP功能春弥。
優(yōu)勢(shì):
- 輕量級(jí)可嵌入開發(fā)，常用的CEP功能簡(jiǎn)單好用叠荠。
- EPL語(yǔ)法與SQL類似匿沛，學(xué)習(xí)成本較低。
劣勢(shì):
- 單機(jī)全內(nèi)存方案榛鼎，需要整合其他分布式和存儲(chǔ)逃呼。
- 以內(nèi)存實(shí)現(xiàn)時(shí)間窗功能，無(wú)法支持較長(zhǎng)跨度的時(shí)間窗者娱。
- 無(wú)法有效支持定時(shí)觸達(dá)（如用戶在瀏覽發(fā)生一段時(shí)間后觸達(dá)條件判斷）抡笼。
Drools Fusion
Drools開始于規(guī)則引擎，后引入Drools Fusion模塊提供CEP的功能黄鳍。
優(yōu)勢(shì):
- 功能較為完善推姻，具有如系統(tǒng)監(jiān)控、操作平臺(tái)等功能框沟。
- 規(guī)則支持動(dòng)態(tài)更新
劣勢(shì):
- 以內(nèi)存實(shí)現(xiàn)時(shí)間窗功能藏古，無(wú)法支持較長(zhǎng)跨度的時(shí)間窗。
- 無(wú)法有效支持定時(shí)觸達(dá)（如用戶在瀏覽發(fā)生一段時(shí)間后觸達(dá)條件判斷）忍燥。
Flink CEP
Flink 是一個(gè)流式系統(tǒng)拧晕，具有高吞吐低延遲的特點(diǎn)，F(xiàn)link CEP是一套極具通用性梅垄、易于使用的實(shí)時(shí)流式事件處理方案厂捞。
優(yōu)勢(shì):
- 繼承了Flink高吞吐的特點(diǎn)
- 事件支持存儲(chǔ)到外部，可以支持較長(zhǎng)跨度的時(shí)間窗。
- 可以支持定時(shí)觸達(dá)（用followedBy＋PartternTimeoutFunction實(shí)現(xiàn)）
劣勢(shì):
- 無(wú)法動(dòng)態(tài)更新規(guī)則（痛點(diǎn)）
自定義規(guī)則
綜上對(duì)比了幾大開源規(guī)則引擎蔫敲，發(fā)現(xiàn)都無(wú)法滿足業(yè)務(wù)特點(diǎn):
- 業(yè)務(wù)方要求支持長(zhǎng)時(shí)間窗口（n天甚至n個(gè)月饲嗽，比如放款一個(gè)月后如果沒(méi)產(chǎn)生還款事件就要發(fā)消息）
- 動(dòng)態(tài)更新規(guī)則，而且要可視化（無(wú)論用哪個(gè)規(guī)則引擎都需要包裝奈嘿，需要考慮二次開發(fā)成本）
- 除了匹配事件貌虾，還需要匹配用戶狀態(tài)
最終我們選擇自己根據(jù)業(yè)務(wù)需要，開發(fā)基于json的自定義規(guī)則裙犹，規(guī)則類似下面例子：
{
"batchId": "xxxxxxxx", //流水號(hào)尽狠，創(chuàng)建每條運(yùn)營(yíng)規(guī)則時(shí)生成
"type": "trigger", //usual
"triggerEvent": "login",
"after": "2h", //分鐘m,小時(shí)h,天d,月M
"pushRules": [//支持同時(shí)推送多條不同類型的消息
{
"pushType": "sms", //wx,app,coupon
"channel": "cl",
"content": "hello #{userInfo.name}"
},
{
"pushType": "coupon",
"couponId": 1234
}
],
"statusConditions": [
{
"name": "and", //邏輯條件，支持與(and)或(or)非(not)
"conditions": [
{
"name": "range",
"field": "credit_line",
"left": 2000,
"right": 10000,
"includeLeft": true,
"includeRight": false
},
{
"name":"in",
"filed":"education",
"values":["bachelor","master"]
}
]
}
],
"eventConditions": [
{
"name": "or",//邏輯條件叶圃，支持與(and)或(or)非(not)
"conditions": [
{
"name": "event",
"function": "count", //聚合函數(shù),目前只支持count
"eventName": "xxx_button_click",
"range": { //聚合結(jié)果做判斷
"left": 1,
"includeLeft": true
},
"timeWindow": {
"type": "fixed", //fixed為固定窗口袄膏，sliding為滑動(dòng)窗口
"start": "2019-01-01 01:01:01",
"end": "2019-02-01 01:01:01"
},
"conditions": [ //event查詢條件繼承and邏輯條件，所以事件也可以過(guò)濾字段
{
"name": "equals",
"field": "f1",
"value": "v1"
}
]
}
]
}
]
}
使用面向?qū)ο笏季S對(duì)過(guò)濾條件做抽象后掺冠，過(guò)濾條件繼承關(guān)系如下：
然后代碼里加一層parser把Condition都轉(zhuǎn)成ES查詢語(yǔ)句沉馆，實(shí)現(xiàn)輕量級(jí)的業(yè)務(wù)規(guī)則配置功能。
整體技術(shù)方案
系統(tǒng)組成模塊及功能如下：
mysql binlog：mysql的數(shù)據(jù)變更德崭，由kafka connector插件讀取到kafka斥黑，數(shù)據(jù)源之一
postgres xlog：pg的數(shù)據(jù)變更，由kafka connector插件讀取到kafka眉厨，數(shù)據(jù)源之一
report server：事件上報(bào)服務(wù)锌奴，數(shù)據(jù)源之一
tags：用戶畫像系統(tǒng)計(jì)算出來(lái)的標(biāo)簽，數(shù)據(jù)源之一
觸發(fā)場(chǎng)景路由：分實(shí)時(shí)觸發(fā)和延遲觸發(fā)憾股，實(shí)時(shí)觸發(fā)直接到下一步鹿蜀，延遲觸發(fā)基于 redis的延遲隊(duì)列實(shí)現(xiàn)
用戶篩選處理器：將篩選規(guī)則翻譯為ES查詢語(yǔ)句到ES查詢用戶數(shù)據(jù)，可以是批量的和單個(gè)用戶的
冪等處理器：對(duì)數(shù)據(jù)做冪等處理服球，防止重復(fù)消費(fèi)
變量渲染處理器：對(duì)推送內(nèi)容做處理
推送適配器：兼容不同的推送方式
BloomFilter記錄器：將推送用戶和流水號(hào)記錄到redis茴恰，用于冪等處理
推送事件記錄器：將推送事件推入kafka
定時(shí)任務(wù)模塊：基于elastic-job，處理定時(shí)推送任務(wù)
規(guī)則配置控制臺(tái)：提供可視化配置界面（運(yùn)營(yíng)規(guī)則配置斩熊、數(shù)據(jù)采集規(guī)則配置琐簇、字段元數(shù)據(jù)配置等）
報(bào)表服務(wù)：提供報(bào)表查詢功能
運(yùn)營(yíng)位服務(wù)：提供外部接口，根據(jù)條件匹配運(yùn)營(yíng)位（如啟動(dòng)圖座享、首頁(yè)banner圖片等）
總結(jié)與展望
- 系統(tǒng)基本滿足了目前的業(yè)務(wù)需求，對(duì)轉(zhuǎn)化率等運(yùn)營(yíng)指標(biāo)提升顯著
- 可以擴(kuò)展其它業(yè)務(wù)似忧，如推薦渣叛、風(fēng)控、業(yè)務(wù)監(jiān)控等
- 規(guī)則定時(shí)拉取盯捌，實(shí)時(shí)性差淳衙，可以用zk做發(fā)布訂閱實(shí)現(xiàn)即時(shí)更新
- 目前事件的聚合函數(shù)只支持count，能滿足業(yè)務(wù)需求但是未來(lái)可能還需要支持其它函數(shù)
- 系統(tǒng)只經(jīng)過(guò)千萬(wàn)級(jí)用戶，日千萬(wàn)級(jí)事件數(shù)據(jù)的生產(chǎn)驗(yàn)證箫攀，再高數(shù)量級(jí)的話可能還有很多性能優(yōu)化的工作,如ES并行查詢（目前用scroll api批量拉取用戶數(shù)據(jù)是串行的）
- 事件類數(shù)據(jù)越來(lái)越多肠牲，目前采取定時(shí)刪除半年前數(shù)據(jù)的方式，防止持續(xù)增長(zhǎng)過(guò)快不可控靴跛，所以事件類條件不可超過(guò)半年的時(shí)間窗口
- 雖然系統(tǒng)對(duì)業(yè)務(wù)無(wú)入侵缀雳，但是反過(guò)來(lái)看本系統(tǒng)依賴于上游數(shù)據(jù)，上游數(shù)據(jù)發(fā)生變化時(shí)如何做到影響最猩揖Α肥印？
未來(lái)會(huì)繼續(xù)從技術(shù)及業(yè)務(wù)兩方面入手，將系統(tǒng)建設(shè)的更加易用绝葡、高效深碱。
