1、引言

用戶畫(huà)像，即用戶信息標(biāo)簽化，它本質(zhì)是對(duì)用戶的一種建模，能夠幫助企業(yè)快速找到精準(zhǔn)用戶群體以及用戶需求等更為廣泛的反饋信息，在現(xiàn)如今應(yīng)用越來(lái)越廣泛。本文主要講述用戶畫(huà)像在離線、實(shí)時(shí)方面的數(shù)據(jù)鏈路處理以及基于特定場(chǎng)景要求如何將離線、實(shí)時(shí)畫(huà)像進(jìn)行在線融合的過(guò)程。

2、背景

目前的算法畫(huà)像服務(wù)分為兩部分，一部分是離線畫(huà)像，也就是批處理計(jì)算層，依賴DataWorks每天T+1的調(diào)度處理。批處理層是通過(guò)處理所有的已有歷史數(shù)據(jù)來(lái)實(shí)現(xiàn)數(shù)據(jù)的準(zhǔn)確性。這意味著它是基于完整的數(shù)據(jù)集來(lái)重新計(jì)算的，能夠修復(fù)數(shù)據(jù)錯(cuò)誤；另一部分是實(shí)時(shí)畫(huà)像，它的數(shù)據(jù)處理依賴流式計(jì)算層Flink。根據(jù)用戶實(shí)時(shí)的行為數(shù)據(jù)進(jìn)行流式處理實(shí)時(shí)更新用戶畫(huà)像。由于兩種模式提供的狀態(tài)差異，所以需要我們?yōu)榕幚砗土魈幚硖峁┎煌姆?wù)層并在這個(gè)上面做合并處理。基于此，需要我們基于離線和實(shí)時(shí)畫(huà)像進(jìn)行融合處理。

整個(gè)的數(shù)據(jù)鏈路大致如下：

主要分為三部分：批處理層、流處理層、數(shù)據(jù)融合層。接下來(lái)逐一講解每層的數(shù)據(jù)鏈路處理。

3、批處理層

批處理層依賴于定時(shí)調(diào)度，基于用戶日常的行為數(shù)據(jù)通過(guò)批處理過(guò)程以精確地計(jì)算用戶的離線畫(huà)像。離線畫(huà)像一方面用作補(bǔ)充實(shí)時(shí)鏈路的數(shù)據(jù)問(wèn)題；另一方面是當(dāng)用戶冷啟動(dòng)時(shí)，如何進(jìn)行用戶畫(huà)像的補(bǔ)充，在算法側(cè)請(qǐng)求時(shí)能夠拿到這部分用戶的畫(huà)像。同時(shí)在離線畫(huà)像數(shù)據(jù)加工完成后，需要考慮將這部分ODPS中的離線畫(huà)像及時(shí)地更新到用戶畫(huà)像服務(wù)中。在這里我們采取懶加載的方式，將離線畫(huà)像存儲(chǔ)到HBASE中，后續(xù)基于用戶當(dāng)天第一次啟動(dòng)App時(shí)，將用戶的離線畫(huà)像進(jìn)行加載，這部分懶加載流程會(huì)在下文講解。

數(shù)據(jù)鏈路如下：

主要分為兩個(gè)部分：

a、每天定時(shí)調(diào)度生成日活用戶的離線畫(huà)像T+1，導(dǎo)入HBASE中。

b、基于步驟1的完成，向HBASE中記錄一條Log，代表當(dāng)天T+1的離線畫(huà)像已經(jīng)成功寫(xiě)入，Log中包含當(dāng)天畫(huà)像的數(shù)據(jù)量、畫(huà)像的版本號(hào)及完成時(shí)間。這里的Log實(shí)際是作為標(biāo)志位，用于判斷T+1畫(huà)像的完整性，后續(xù)懶加載流程會(huì)利用當(dāng)天的Log來(lái)判斷是否加載離線畫(huà)像以及加載幾次。

4、流處理層

這里的流處理層分為兩塊，一塊是實(shí)時(shí)畫(huà)像，訂閱用戶的實(shí)時(shí)行為數(shù)據(jù)進(jìn)行Flink處理而來(lái)；另一塊實(shí)際是對(duì)批處理層提供的離線畫(huà)像進(jìn)行處理，基于用戶的實(shí)時(shí)登錄行為懶加載離線畫(huà)像。

上文提到在批處理層將用戶離線畫(huà)像導(dǎo)入HBASE后，通過(guò)懶加載的方式將離線畫(huà)像加載到畫(huà)像融合框架。

整個(gè)懶加載流程如下：

大致分為如下幾個(gè)步驟：

a、訂閱用戶的登錄行為埋點(diǎn)APPSTART。

b、根據(jù)訂閱的用戶登錄行為加載HBASE中的離線畫(huà)像。這里有一點(diǎn)需要說(shuō)明的就是上述提到HBASE中的畫(huà)像Log記錄，利用Log來(lái)判斷是否需要加載畫(huà)像。假設(shè)當(dāng)天T+1的畫(huà)像已經(jīng)完整的導(dǎo)入到HBASE中，當(dāng)天用戶第一次登錄時(shí)，就會(huì)Load離線畫(huà)像，同時(shí)利用Flink的State記錄當(dāng)天用戶已經(jīng)加載了T+1畫(huà)像，后續(xù)用戶當(dāng)天再次登錄時(shí)就不會(huì)再Load離線畫(huà)像，做到當(dāng)天只加載一次T+1畫(huà)像，降低HBASE的訪問(wèn)壓力；相反如果用戶當(dāng)天登錄時(shí)，T+1畫(huà)像并沒(méi)有Log記錄， Load畫(huà)像時(shí)State會(huì)記錄用戶當(dāng)天加載了T+2畫(huà)像，后續(xù)只有當(dāng)T+1畫(huà)像完成后用戶再次登錄，才會(huì)去獲取一次最新的離線畫(huà)像，同時(shí)更改State記錄。

c、讀取標(biāo)簽配置表，根據(jù)對(duì)應(yīng)標(biāo)簽的配置信息將畫(huà)像的格式、類(lèi)型進(jìn)行轉(zhuǎn)換滿足算法側(cè)的使用。

d、將轉(zhuǎn)換后的畫(huà)像統(tǒng)一成畫(huà)像框架消費(fèi)的Action格式發(fā)送到消息隊(duì)列中，供后續(xù)融合框架消費(fèi)和實(shí)時(shí)畫(huà)像進(jìn)行融合。

懶加載的流程整體上就是上面所述，在這里有一點(diǎn)補(bǔ)充就是步驟1中訂閱的用戶登錄埋點(diǎn)APPSTART。在實(shí)際中由于受到埋點(diǎn)上報(bào)延遲、網(wǎng)絡(luò)等一系列原因，可能會(huì)導(dǎo)致部分用戶離線畫(huà)像加載的延遲，用戶請(qǐng)求時(shí)離線畫(huà)像尚未加載到，造成畫(huà)像覆蓋率降低?；诖?，我們通過(guò)訂閱用戶的Init數(shù)據(jù)（先于推薦流請(qǐng)求）作為補(bǔ)充觸發(fā)事件來(lái)加載離線畫(huà)像，從而進(jìn)一步提升畫(huà)像覆蓋率。

另外就是Log中版本號(hào)的概念，主要是為了容錯(cuò)，防止出現(xiàn)畫(huà)像數(shù)據(jù)版本當(dāng)天迭代更新。我們要求每次迭代version都要對(duì)應(yīng)+1，這樣當(dāng)用戶登錄時(shí)假如當(dāng)天的version出現(xiàn)了變化會(huì)再次加載最新的版本畫(huà)像，從而保障用戶加載的離線畫(huà)像版本是最新的。

接下來(lái)看下實(shí)時(shí)畫(huà)像的數(shù)據(jù)鏈路，整個(gè)流程如下：

大致分為如下幾個(gè)步驟：

1）Flink訂閱用戶行為數(shù)據(jù)，根據(jù)畫(huà)像具體的業(yè)務(wù)要求處理行為數(shù)據(jù)。

2）將處理后的行為數(shù)據(jù)構(gòu)建畫(huà)像框架統(tǒng)一的Action算子發(fā)送到Kafka中。Action中包含標(biāo)簽名稱、標(biāo)簽值、標(biāo)簽對(duì)應(yīng)的處理算子、行為時(shí)間等相關(guān)信息。

3）畫(huà)像框架消費(fèi)Action信息，根據(jù)配置的信息做對(duì)應(yīng)的算子類(lèi)型處理。比如map、List、String等一系列類(lèi)型處理。

4）將處理后的實(shí)時(shí)畫(huà)像寫(xiě)入Redis。 離線畫(huà)像的懶加載流程和實(shí)時(shí)畫(huà)像處理流程大致如上，最終目的是要按照框架Action格式發(fā)送到Kafka中供畫(huà)像框架融合使用，達(dá)到離線和實(shí)時(shí)畫(huà)像的合并。

5、畫(huà)像融合層

基于批處理層和流處理層的畫(huà)像數(shù)據(jù)，我們需要將離線畫(huà)像和實(shí)時(shí)畫(huà)像進(jìn)行融合處理。

首先需要明確的一點(diǎn)就是離線、實(shí)時(shí)畫(huà)像的數(shù)據(jù)格式一定要統(tǒng)一，否則談不上融合。同時(shí)在數(shù)據(jù)處理的口徑上也是要統(tǒng)一的，這樣做的好處是校驗(yàn)數(shù)據(jù)時(shí)容易追溯、定位問(wèn)題。

那如何進(jìn)行畫(huà)像融合呢？這里以具體的標(biāo)簽舉例。假如標(biāo)簽a是用戶的點(diǎn)擊行為序列List，序列中包含用戶點(diǎn)擊商品cspuId、用戶行為時(shí)間、商品推薦渠道等信息。標(biāo)簽a的數(shù)據(jù)格式如下：

[
{"cspuId":111, "et":1663234014003, "channel":1},
{"cspuId":222, "et":1663234030023, "channel":2},
{"cspuId":333, "et":1663234050083, "channel":3},
{"cspuId":444, "et":1663234085048, "channel":4}
......
]

在畫(huà)像配置表中，我們首先會(huì)配置標(biāo)簽a的相關(guān)信息，比如sizeLimt為1000，排序字段為et，按照cspuId、et兩個(gè)字段去重等等信息。

在實(shí)時(shí)畫(huà)像層，我們知道用戶實(shí)時(shí)的點(diǎn)擊行為會(huì)產(chǎn)生實(shí)時(shí)的點(diǎn)擊畫(huà)像數(shù)據(jù)，假設(shè)產(chǎn)生的實(shí)時(shí)畫(huà)像數(shù)據(jù)如下:

{"cspuId":444, "et":1663234085048, "channel":4}

基于這個(gè)實(shí)時(shí)畫(huà)像數(shù)據(jù)我們會(huì)構(gòu)建統(tǒng)一的Action格式算子，實(shí)時(shí)的標(biāo)簽a配置的處理算子是 list.rpush，代表將針對(duì)a標(biāo)簽進(jìn)行List的add操作。

在懶加載層，加載到的離線標(biāo)簽a的數(shù)據(jù)格式如下：

[
{"cspuId":111, "et":1663234014003, "channel":1},
{"cspuId":222, "et":1663234030023, "channel":2},
{"cspuId":333, "et":1663234050083, "channel":3}
]

基于這個(gè)離線畫(huà)像我們也會(huì)構(gòu)建統(tǒng)一的Action格式算子，離線標(biāo)簽a配置的處理算子是 list.rpushl，代表對(duì)a標(biāo)簽進(jìn)行List的addAll操作。

畫(huà)像融合框架消費(fèi)Action消息隊(duì)列時(shí)，由于TTL的原因，假設(shè)Redis中用戶的a標(biāo)簽數(shù)據(jù)已經(jīng)清空，在用戶冷啟動(dòng)時(shí)畫(huà)像框架會(huì)根據(jù)消費(fèi)到的離線標(biāo)簽數(shù)據(jù)及對(duì)應(yīng)的操作算子將a標(biāo)簽數(shù)據(jù)補(bǔ)充完整。與此同時(shí)用戶后續(xù)產(chǎn)生了上述實(shí)時(shí)的畫(huà)像，同樣道理根據(jù)對(duì)應(yīng)的操作算子將實(shí)時(shí)畫(huà)像add到標(biāo)簽a中，當(dāng)然會(huì)根據(jù)標(biāo)簽a的配置信息比如大小，排序字段等取最近的sizeLimit畫(huà)像。

另外比如用戶的a標(biāo)簽中數(shù)據(jù)已經(jīng)有歷史累積了，這時(shí)候離線畫(huà)像可以用作數(shù)據(jù)修復(fù)。畫(huà)像融合框架拿到離線畫(huà)像會(huì)結(jié)合已經(jīng)存在的a標(biāo)簽數(shù)據(jù)進(jìn)行去重，按照et排序等一系列操作，補(bǔ)充實(shí)時(shí)鏈路可能出現(xiàn)的數(shù)據(jù)丟失問(wèn)題，最終得到完整的上述a標(biāo)簽數(shù)據(jù)。

考慮到不同類(lèi)型標(biāo)簽的操作差異，畫(huà)像融合框架會(huì)根據(jù)需求定制不同的操作算子，這樣可以很靈活地處理算法側(cè)不同的標(biāo)簽需求。

基于此，通過(guò)簡(jiǎn)單的標(biāo)簽舉例，能夠了解整個(gè)畫(huà)像融合的過(guò)程。當(dāng)然實(shí)際中還有更多細(xì)節(jié)問(wèn)題可以后續(xù)進(jìn)一步分享。

6、總結(jié)

整個(gè)離線、實(shí)時(shí)畫(huà)像的融合鏈路整體上如上所述。從數(shù)據(jù)準(zhǔn)備、數(shù)據(jù)加工、數(shù)據(jù)融合到最終提供完整畫(huà)像，實(shí)際上類(lèi)似于Lambda架構(gòu)。當(dāng)然在批處理層，考慮到不同業(yè)務(wù)域?qū)+1日活畫(huà)像完整性的要求，我們采用了不同的處理方式。比如直接將這部分日活畫(huà)像寫(xiě)到Redis中而不是通過(guò)懶加載方式去更新，這樣可以讓算法側(cè)自身去結(jié)合實(shí)際場(chǎng)景融合使用。另外一點(diǎn)就是在批處理層是否能夠進(jìn)一步優(yōu)化，降低維護(hù)成本，比如HBASE的中間存儲(chǔ)，目前也在探索基于每天生成離線畫(huà)像的snapshot，直接從ODPS進(jìn)行Load使用，也是在進(jìn)一步探索如何充分利用離線畫(huà)像的同時(shí)降低開(kāi)發(fā)成本。