從零搭建精準運營系統
2018剛過去，趁著春節放假對過去一年主導開發的專案做個梳理和總結
專案背景
平臺運營到一定階段，一定會累積大批量的使用者資料，這些使用者資料是運營人員的黃金財產。而如何利用使用者的資料來做運營（訊息推送、觸達訊息、優惠券傳送、廣告位等），正是精準運營系統需要解決的問題。本文是基於信貸業務實踐後寫出來的，其它行業如保險、電商、航旅、遊戲等也可以參考。
業務場景
先看幾個具有代表性的需求
使用者可用額度在20000～50000元，而且有借款記錄，未還本金為0，性別為“男”
使用者發生了A行為且未還本金大於5000
使用者在1天內發生A行為次數大於等於3次
使用者在A行為前24小時內未發生B行為
使用者在A行為後一個月內未發生B行為
業務上有兩種訊息型別
- 日常訊息：由業務人員通過條件篩選鎖定使用者群，定時或即時給批量使用者傳送訊息或者優惠券
- 觸達訊息：主要由使用者自身的行為觸發，比如登陸、進件申請、還款等，滿足一定篩選條件 實時 給使用者傳送訊息或優惠券
對於使用者篩選條件，也主要有兩種型別
- 使用者狀態：包括使用者自身屬性如性別、年齡、學歷、收入等，還有使用者相關聯實體如進件訂單、賬戶資訊、還款計劃、優惠券等的屬性，以及使用者畫像資料如行為偏好、進件概率等
- 使用者行為：即使用者的動作，包括登陸、進件申請、還款，甚至前端點選某個按鈕、在某個文字框輸入都算
早期方案
早期方案存在以下痛點
- 至少兩次跨部門溝通配合成本，週期被拉長
- 非實時訊息推送，無法實現基於使用者行為的實時推送場景
- 非實時效果驗證，無法及時調整運營策略
系統搭建的目標
- 需要定義規則，提供視覺化介面給業務人員動態配置，無需重啟系統即使生效，減少溝通成本和避免重複開發，總之就是要更加 自動化 和 易配置
- 採集實時資料，根據實時事件做實時推送，總之就是要 實時
技術選型
資料採集、轉換、儲存
- 採集：狀態類的資料主要放在各個業務系統的關係型資料庫中，由於歷史原因有postgres和mysql，需要實時採集表的資料變更，這裡使用kafka connector讀取mysql的binlog或postgres的xlog，另外還有標籤系統計算出來的標籤，在kafka中；而事件類資料主要來源於前端上報事件（有專門的服務接收再丟到kafka），關係型資料庫裡面也可以提取一些事件。
- 轉換：採集出來的資料需要做一些格式統一等操作，用kafka connector。
- 儲存：採用Elasticsearch儲存使用者資料，ES查詢不像mysql或mongoDB用B-tree 或B+tree實現索引，而是使用bitset和skip list來處理聯合索引，特別適合多欄位的複雜查詢條件。
下面重點看下kafka connector和Elasticsearch如何使用
kafka connector
kafka connector有Source和Sink兩種元件，Source的作用是讀取資料到kafka，這裡用開源實現debezium來採集mysql的binlog和postgres的xlog。Sink的作用是從kafka讀資料寫到目標系統，這裡自己研發一套元件，根據配置的規則將資料格式化再同步到ES。
kafka connector有以下優點：
- 提供大量開箱即用的外掛，比如我們直接用debezium就能解決讀取mysql和pg資料變更的問題
- 伸縮性強，對於不同的connector可以配置不同數量的task，分配給不同的worker，，我們可以根據不同topic的流量大小來調節配置。
- 容錯性強，worker失敗會把task遷移到其它worker上面
- 使用rest介面進行配置，我們可以對其進行包裝很方便地實現一套管理介面
Elasticsearch
對於狀態資料，由於狀態的寫操作相對較少，我們採取巢狀文件的方式，將同個使用者的相關實體資料都同步寫入到同個文件，具體實現用painless指令碼做區域性更新操作。效果類似這樣：
{
"id":123,
"age":30,
"credit_line":20000,
"education":"bachelor",
...
"last_loan_applications":{
"loan_id":1234,
"status":"reject",
...
}
...
}
事件資料寫入比較頻繁，資料量比較多，我們使用父子文件的方式做關聯，效果類似這樣：
{
"e_uid":123,
"e_name":"loan_application",
"e_timestamp":"2019-01-01 10:10:00"
...
}
（e_字首是為了防止同個index下同名欄位衝突）
ES這樣儲存一方面是方便做統計報表，另一方面跟使用者篩選和觸達有關。
規則引擎
在設計規則引擎前，我們對業界已有的規則引擎，主要包括 Esper , Drools , Flink CEP ，進行了初步調研。
Esper
Esper設計目標為CEP的輕量級解決方案，可以方便的嵌入服務中，提供CEP功能。
優勢:
- 輕量級可嵌入開發，常用的CEP功能簡單好用。
- EPL語法與SQL類似，學習成本較低。
劣勢:
- 單機全記憶體方案，需要整合其他分散式和儲存。
- 以記憶體實現時間窗功能，無法支援較長跨度的時間窗。
- 無法有效支援定時觸達（如使用者在瀏覽發生一段時間後觸達條件判斷）。
Drools
Drools開始於規則引擎，後引入Drools Fusion模組提供CEP的功能。
優勢:
- 功能較為完善，具有如系統監控、操作平臺等功能。
- 規則支援動態更新
劣勢:
- 以記憶體實現時間窗功能，無法支援較長跨度的時間窗。
- 無法有效支援定時觸達（如使用者在瀏覽發生一段時間後觸達條件判斷）。
Flink
Flink 是一個流式系統，具有高吞吐低延遲的特點，Flink CEP是一套極具通用性、易於使用的實時流式事件處理方案。
優勢:
- 繼承了Flink高吞吐的特點
- 事件支援儲存到外部，可以支援較長跨度的時間窗。
- 可以支援定時觸達（用followedBy＋PartternTimeoutFunction實現）
劣勢:
- 無法動態更新規則（痛點）
自定義規則
綜上對比了幾大開源規則引擎，發現都無法滿足業務需求:
- 業務方要求支援長時間視窗（n天甚至n個月，比如放款一個月後如果沒產生還款事件就要發訊息）
- 動態更新規則，而且要視覺化（無論用哪個規則引擎都需要包裝，需要考慮二次開發成本）
最終我們選擇自己根據業務需要，開發基於json的自定義規則，規則類似下面例子：
{
"batchId": "xxxxxxxx", //流水號，建立每條運營規則時生成
"type": "trigger", //usual
"triggerEvent": "login",
"after": "2h", //分鐘m,小時h,天d,月M
"pushRules": [//支援同時推送多條不同型別的訊息
{
"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", //聚合函式,目前只支援count
"eventName": "xxx_button_click",
"range": { //聚合結果做判斷
"left": 1,
"includeLeft": true
},
"timeWindow": {
"type": "fixed", //fixed為固定視窗，sliding為滑動視窗
"start": "2019-01-01 01:01:01",
"end": "2019-02-01 01:01:01"
},
"conditions": [ //event查詢條件繼承and邏輯條件，所以事件也可以過濾欄位
{
"name": "equals",
"field": "f1",
"value": "v1"
}
]
}
]
}
]
}
使用面向物件思維對過濾條件做抽象後，過濾條件繼承關係如下：
然後程式碼里加一層parser把Condition都轉成ES查詢語句，實現輕量級的業務規則配置功能。
整體技術方案
系統組成模組及功能如下：
mysql binlog：mysql的資料變更，由kafka connector外掛讀取到kafka，資料來源之一
postgres xlog：pg的資料變更，由kafka connector外掛讀取到kafka，資料來源之一
report server：事件上報服務，資料來源之一
tags：使用者畫像系統計算出來的標籤，資料來源之一
觸發場景路由：分實時觸發和延遲觸發，實時觸發直接到下一步，延遲觸發基於 rabbitmq的延遲佇列實現
使用者篩選模組：將篩選規則翻譯為ES查詢語句到ES查詢使用者資料，可以是批量的和單個使用者的
變數渲染模組：對推送內容做處理
推送介面卡：相容不同的推送方式
定時任務排程器：基於elastic-job，處理定時推送任務
規則配置控制檯：提供視覺化配置介面（運營規則配置、資料採集規則配置、欄位元資料配置等）
報表服務：提供報表查詢功能
運營位服務：提供外部介面，根據條件匹配運營位（如啟動圖、首頁banner圖片等）
總結與展望
- 系統基本滿足了目前的業務需求，對轉化率等運營指標提升顯著
- 可以擴充套件其它業務，如推薦、風控、業務監控等
- 規則定時拉取，實時性差，可以用zk做釋出訂閱實現即時更新
- 目前事件的聚合函式只支援count，能滿足業務需求但是未來可能還需要支援其它函式
- 系統只經過千萬級使用者的生產驗證，再高數量級的話可能還有很多效能優化的工作,如ES並行查詢（目前用scroll api批量拉取使用者資料是序列的）
- 事件類資料越來越多，目前採取定時刪除半年前資料的方式，防止持續增長過快不可控，所以事件類條件不可超過半年的時間視窗
- 雖然系統對業務無入侵，但是反過來看本系統依賴於上游資料，上游資料發生變化時如何做到影響最小？
未來會繼續從技術及業務兩方面入手，將系統建設的更加易用、高效。
