Flink 在廣告業(yè)務(wù)的實(shí)踐

一、業(yè)務(wù)場(chǎng)景

實(shí)時(shí)數(shù)據(jù)在廣告業(yè)務(wù)的使用場(chǎng)景主要可以分為四個(gè)方面：

• 數(shù)據(jù)大屏：包括曝光、點(diǎn)擊、收入等核心指標(biāo)的展示，以及故障率等監(jiān)控指標(biāo)；
  
  
• 異常監(jiān)測(cè)：因?yàn)閺V告投放的鏈路比較?，所以如果鏈路上發(fā)生任何波動(dòng)的話(huà)，都會(huì)對(duì)整體的投放效果產(chǎn)生影響。除此之外，各個(gè)團(tuán)隊(duì)在上線(xiàn)過(guò)程中是否會(huì)對(duì)整體投放產(chǎn)生影響，都是通過(guò)異常監(jiān)測(cè)系統(tǒng)能夠觀測(cè)到的。我們還能夠觀測(cè)業(yè)務(wù)指標(biāo)走勢(shì)是否合理，比如在庫(kù)存正常的情況下，曝光是否有不同的波動(dòng)情況，這可以用來(lái)實(shí) 時(shí)發(fā)現(xiàn)問(wèn)題;

  
  

• 數(shù)據(jù)分析：主要用于數(shù)據(jù)賦能業(yè)務(wù)發(fā)展。我們可以實(shí)時(shí)分析廣告投放過(guò)程中的一些異常問(wèn)題，或者基于當(dāng)前的投放效果去研究怎樣優(yōu)化，從而達(dá)到更好的效果;

  
  

• 特征工程：廣告算法團(tuán)隊(duì)主要是做一些模型訓(xùn)練，用于支持線(xiàn)上投放。技術(shù)特征最初大部分是離線(xiàn)，隨著實(shí)時(shí)的發(fā)展，開(kāi)始把一些工程轉(zhuǎn)到實(shí)時(shí)。

二、業(yè)務(wù)實(shí)踐

業(yè)務(wù)實(shí)踐主要分為兩類(lèi)，第一個(gè)是實(shí)時(shí)數(shù)倉(cāng)，第二個(gè)是特征工程。

1. 實(shí)時(shí)數(shù)倉(cāng)

■?1.1 實(shí)時(shí)數(shù)倉(cāng) - 目標(biāo)

實(shí)時(shí)數(shù)倉(cāng)的目標(biāo)包括數(shù)據(jù)完整性、服務(wù)穩(wěn)定性和查詢(xún)能力。

• 數(shù)據(jù)完整性：在廣告業(yè)務(wù)里，實(shí)時(shí)數(shù)據(jù)主要是用于指導(dǎo)決策，比如廣告主需要根據(jù)當(dāng)前投放的實(shí)時(shí)數(shù)據(jù)，指導(dǎo)后面的出價(jià)或調(diào)整預(yù)算。另外，故障率的監(jiān)控需要數(shù)據(jù)本身是穩(wěn)定的。如果數(shù)據(jù)是波動(dòng)的，指導(dǎo)意義就非常差，甚至沒(méi)有什么指導(dǎo)意義。因此完整性本身是對(duì)時(shí)效性和完整性之間做了一個(gè)權(quán)衡；

  
  

• 服務(wù)穩(wěn)定性：生產(chǎn)鏈包括數(shù)據(jù)接入、計(jì)算（多層）、數(shù)據(jù)寫(xiě)入、進(jìn)度服務(wù)和查詢(xún)服務(wù)。除此之外還有數(shù)據(jù)質(zhì)量，包括數(shù)據(jù)的準(zhǔn)確性以及數(shù)據(jù)趨勢(shì)是否符合預(yù)期；

  
  

• 查詢(xún)能力：在廣告業(yè)務(wù)有多種使用場(chǎng)景，在不同場(chǎng)景里可能使用了不同的 OLAP 引擎，所以查詢(xún)方式和性能的要求不一致。另外，在做數(shù)據(jù)分析的時(shí)候，除了最新最穩(wěn)定的實(shí)時(shí)數(shù)據(jù)之外，同時(shí)也會(huì)實(shí)時(shí) + 離線(xiàn)做分析查詢(xún)，此外還包括數(shù)據(jù)跨源和查詢(xún)性能等要求。

■?1.2 實(shí)時(shí)數(shù)倉(cāng) - 挑戰(zhàn)

• 數(shù)據(jù)進(jìn)度服務(wù)：需要在時(shí)效性和完整性之間做一個(gè)權(quán)衡；

  
  

• 數(shù)據(jù)穩(wěn)定性：由于生產(chǎn)鏈路比較長(zhǎng)，中間可能會(huì)用到多種功能組件，所以端到端的服務(wù)穩(wěn)定性對(duì)整體數(shù)據(jù)準(zhǔn)確性的影響是比較關(guān)鍵的；

  
  

• 查詢(xún)性能：主要包括 OLAP 分析能力。在實(shí)際場(chǎng)景中，數(shù)據(jù)表包含了離線(xiàn)和實(shí)時(shí)，單表規(guī)模達(dá)上百列，行數(shù)也是非常大的。

■?1.3 廣告數(shù)據(jù)平臺(tái)架構(gòu)

上圖為廣告數(shù)據(jù)平臺(tái)基礎(chǔ)架構(gòu)圖，從下往上看：

• 底部是數(shù)據(jù)采集層，這里與大部分公司基本一致。業(yè)務(wù)數(shù)據(jù)庫(kù)主要包含了廣告主的下單數(shù)據(jù)以及投放的策略；埋點(diǎn)日志和計(jì)費(fèi)日志是廣告投放鏈路過(guò)程中產(chǎn)生的日志；

  
  

• 中間是數(shù)據(jù)生產(chǎn)的部分，數(shù)據(jù)生產(chǎn)的底層是大數(shù)據(jù)的基礎(chǔ)設(shè)施，這部分由公司的一個(gè)云平臺(tái)團(tuán)隊(duì)提供，其中包含 Spark / Flink 計(jì)算引擎，Babel 統(tǒng)一的管理平臺(tái)。Talos 是實(shí)時(shí)數(shù)倉(cāng)服務(wù)，RAP 和 OLAP 對(duì)應(yīng)不同的實(shí)時(shí)分析以及 OLAP 存儲(chǔ)和查詢(xún)服務(wù)。
  
  
  數(shù)據(jù)生產(chǎn)的中間層是廣告團(tuán)隊(duì)包含的一些服務(wù)，例如在生產(chǎn)里比較典型的離線(xiàn)計(jì)算和實(shí)時(shí)計(jì)算。




• 離線(xiàn)是比較常見(jiàn)的一個(gè)分層模型，調(diào)度系統(tǒng)是對(duì)生產(chǎn)出的離線(xiàn)任務(wù)做有效的管理和調(diào)度。
• 實(shí)時(shí)計(jì)算這邊使用的引擎也比較多，我們的實(shí)時(shí)化是從 2016 年開(kāi)始，當(dāng)時(shí)選的是 Spark Streaming，后面隨著大數(shù)據(jù)技術(shù)發(fā)展以及公司業(yè)務(wù)需求產(chǎn)生了不同場(chǎng)景，又引入了計(jì)算引擎 Flink。
• 實(shí)時(shí)計(jì)算底層調(diào)度依賴(lài)于云計(jì)算的 Babel 系統(tǒng)，除了計(jì)算之外還會(huì)伴隨數(shù)據(jù)治理，包括進(jìn)度管理，就是指實(shí)時(shí)計(jì)算里一個(gè)數(shù)據(jù)報(bào)表當(dāng)前已經(jīng)穩(wěn)定的進(jìn)度到哪個(gè)時(shí)間點(diǎn)。離線(xiàn)里其實(shí)就對(duì)應(yīng)一個(gè)表，有哪些分區(qū)。
• 血緣管理包括兩方面，離線(xiàn)包括表級(jí)別的血緣以及字段血緣。實(shí)時(shí)主要還是在任務(wù)層面的血緣。
• 至于生命周期管理，在離線(xiàn)的一個(gè)數(shù)倉(cāng)里，它的計(jì)算是持續(xù)迭代的。但是數(shù)據(jù)保留時(shí)間非常長(zhǎng)的話(huà)，數(shù)據(jù)量對(duì)于底層的存儲(chǔ)壓力就會(huì)比較大。
• 數(shù)據(jù)生命周期管理主要是根據(jù)業(yè)務(wù)需求和存儲(chǔ)成本之間做一個(gè)權(quán)衡。
• 質(zhì)量管理主要包括兩方面，一部分在數(shù)據(jù)接入層，判斷數(shù)據(jù)本身是否合理；另外一部分在數(shù)據(jù)出口，就是結(jié)果指標(biāo)這一層。因?yàn)槲覀兊臄?shù)據(jù)會(huì)供給其他很多團(tuán)隊(duì)使用，因此在數(shù)據(jù)出口這一層要保證數(shù)據(jù)計(jì)算沒(méi)有問(wèn)題。

  
  

• 再上層是統(tǒng)一查詢(xún)服務(wù)，我們會(huì)封裝很多接口進(jìn)行查詢(xún)。

• 因?yàn)閿?shù)據(jù)化包括離線(xiàn)和實(shí)時(shí)，另外還有跨集群，所以在智能路由這里會(huì)進(jìn)行一些選集群、選表以及復(fù)雜查詢(xún)、拆分等核心功能。
• 查詢(xún)服務(wù)會(huì)對(duì)歷史查詢(xún)進(jìn)行熱度的統(tǒng)一管理。這樣一方面可以更應(yīng)進(jìn)一步服務(wù)生命周期管理，另一方面可以去看哪些數(shù)據(jù)對(duì)于業(yè)務(wù)的意義非常大。
• 除了生命周期管理之外，它還可以指導(dǎo)我們的調(diào)度系統(tǒng)，比如哪些報(bào)表比較關(guān)鍵，在資源緊張的時(shí)候就可以?xún)?yōu)先調(diào)度這些任務(wù)。

  
  

• 再往上是數(shù)據(jù)應(yīng)用，包括報(bào)表系統(tǒng)、Add - hoc 查詢(xún)、數(shù)據(jù)可視化、異常監(jiān)控和下游團(tuán)隊(duì)。

■?1.4 實(shí)時(shí)數(shù)倉(cāng) - 生產(chǎn)鏈路

數(shù)據(jù)生產(chǎn)鏈路是從時(shí)間粒度來(lái)講的，我們最開(kāi)始是離線(xiàn)數(shù)倉(cāng)鏈路，在最底層的這一行，隨著實(shí)時(shí)化需求推進(jìn)，就產(chǎn)生了一個(gè)實(shí)時(shí)鏈路，整理來(lái)說(shuō)，是一個(gè)典型的 Lambda 架構(gòu)。

另外，我們的一些核心指標(biāo)，比如計(jì)費(fèi)指標(biāo)，因?yàn)樗姆€(wěn)定性對(duì)下游比較關(guān)鍵，所以我們這邊采用異路多活。異路多活是源端日志產(chǎn)生之后，在計(jì)算層和下游存儲(chǔ)層做了完全的冗余，在后面的查詢(xún)里做統(tǒng)一處理。

■?1.5 實(shí)時(shí)數(shù)倉(cāng) - 進(jìn)度服務(wù)

上文介紹了我們要求提供出去的實(shí)時(shí)數(shù)據(jù)的指標(biāo)是穩(wěn)定不變的，進(jìn)度服務(wù)實(shí)現(xiàn)的核心點(diǎn)包括時(shí)間窗口里指標(biāo)的變化趨勢(shì)，同時(shí)結(jié)合了實(shí)時(shí)計(jì)算任務(wù)本身的狀態(tài)，因?yàn)樵趯?shí)時(shí)數(shù)倉(cāng)里，很多指標(biāo)是基于時(shí)間窗口做聚合計(jì)算。

比如一個(gè)實(shí)時(shí)指標(biāo)，我們輸出的指標(biāo)是 3 分鐘，也就是說(shuō) 4：00 這個(gè)時(shí)間點(diǎn)的指標(biāo)的就包括了 4：00～4：03 的數(shù)據(jù)，4：03 包括了 4：03～4：06 的數(shù)據(jù)，其實(shí)就是指一個(gè)時(shí)間窗口的數(shù)據(jù)，什么時(shí)候是對(duì)外可見(jiàn)的。因?yàn)樵趯?shí)時(shí)計(jì)算里，數(shù)據(jù)不斷進(jìn)來(lái)， 4：00 的時(shí)間窗口的數(shù)據(jù)從 4：00 開(kāi)始，指標(biāo)就已經(jīng)開(kāi)始產(chǎn)生了。隨著時(shí)間疊加，指標(biāo)不斷上升，最后趨于穩(wěn)定。我們基于時(shí)間窗口指標(biāo)的變化率，來(lái)判斷它是否趨于穩(wěn)定。

但如果只是基于這個(gè)點(diǎn)來(lái)看，那么它還存在一定的弊端。

因?yàn)檫@個(gè)結(jié)果表的計(jì)算鏈會(huì)依賴(lài)很多個(gè)計(jì)算任務(wù)，如果這個(gè)鏈路上面哪個(gè)任務(wù)出現(xiàn)問(wèn)題，可能會(huì)導(dǎo)致當(dāng)前的指標(biāo)雖然走勢(shì)已經(jīng)趨于正常，但是最終并不完整。所以在這基礎(chǔ)之上，我們又引入了實(shí)時(shí)計(jì)算任務(wù)狀態(tài)，在指標(biāo)趨于穩(wěn)定的時(shí)候，同時(shí)去看生產(chǎn)鏈路上這些計(jì)算任務(wù)是否正常，如果是正常的話(huà)，表示任務(wù)本身時(shí)間點(diǎn)的指標(biāo)已經(jīng)穩(wěn)定，可以對(duì)外提供服務(wù)。

如果計(jì)算有卡頓、堆積，或者已經(jīng)有異常在重啟過(guò)程中，就需要繼續(xù)等待迭代處理。

■?1.6 實(shí)時(shí)數(shù)倉(cāng) - 查詢(xún)服務(wù)

上圖為查詢(xún)服務(wù)架構(gòu)圖。

最下方是數(shù)據(jù)，里面有實(shí)時(shí)存儲(chǔ)引擎，包括 Druid 等。在離線(xiàn)中，數(shù)據(jù)在 Hive 里邊，但是在做查詢(xún)的時(shí)候，會(huì)把它們進(jìn)行 OLAP 的同步，在這邊使用的是兩種引擎。為了和 Kudu 做 union 查詢(xún)，會(huì)把它同步到 OLAP 引擎，然后上面去統(tǒng)一使用 Impala 做查詢(xún)。另外，對(duì)于使用場(chǎng)景里比較固定的方式，可以導(dǎo)到 Kylin 里，然后在上面做數(shù)據(jù)分析。

基于這些數(shù)據(jù)，會(huì)有多個(gè)查詢(xún)節(jié)點(diǎn)，再上面是一個(gè)智能路由層。從最上面查詢(xún)網(wǎng)關(guān)，當(dāng)有一個(gè)查詢(xún)請(qǐng)求進(jìn)來(lái)，首先判斷它是不是一個(gè)復(fù)雜場(chǎng)景。比如在一個(gè)查詢(xún)里，如果它的時(shí)長(zhǎng)同時(shí)跨越了離線(xiàn)和實(shí)時(shí)，這里就會(huì)同時(shí)使用到離線(xiàn)表和實(shí)時(shí)表。

另外，離線(xiàn)表里還有更復(fù)雜的選表邏輯，比如小時(shí)級(jí)別，天級(jí)別。經(jīng)過(guò)復(fù)雜場(chǎng)景分析之后，就會(huì)把最終選擇的表大概確定下來(lái)。其實(shí)在做智能路由的時(shí)候，才會(huì)去參考左邊的一些基礎(chǔ)服務(wù)，比如元數(shù)據(jù)管理，當(dāng)前這些表的進(jìn)度到哪個(gè)點(diǎn)了。

對(duì)于查詢(xún)性能的優(yōu)化，在數(shù)據(jù)里，底層掃描的數(shù)據(jù)量對(duì)最終性能的影響是非常大的。所以會(huì)有一個(gè)報(bào)表降維，根據(jù)歷史的查詢(xún)?nèi)プ龇治觥１热缭谝粋€(gè)降維表包含哪些維度，可以覆蓋到百分之多少的查詢(xún)。

■?1.7 數(shù)據(jù)生產(chǎn) - 規(guī)劃

之前在實(shí)時(shí)數(shù)據(jù)報(bào)表生產(chǎn)里提到，它主要是基于 API 的方式實(shí)現(xiàn)的。Lambda 架構(gòu)本身有一個(gè)問(wèn)題就是實(shí)時(shí)跟離線(xiàn)是兩個(gè)計(jì)算團(tuán)隊(duì)，對(duì)于同一個(gè)需求，需要兩個(gè)團(tuán)隊(duì)同時(shí)去開(kāi)發(fā)，這樣會(huì)帶來(lái)幾個(gè)問(wèn)題。

• 一方面是他們的邏輯可能會(huì)發(fā)生差異，最終導(dǎo)致結(jié)果表不一致；

• 另一方面是人力成本，同時(shí)需要兩個(gè)團(tuán)隊(duì)進(jìn)行開(kāi)發(fā)。

因此我們的訴求是流批一體，思考在計(jì)算層是否可以使用一個(gè)邏輯來(lái)表示同一個(gè)業(yè)務(wù)需求，比如可以同時(shí)使用流或者批的計(jì)算引擎來(lái)達(dá)到計(jì)算的效果。

在這個(gè)鏈路里邊，原始數(shù)據(jù)通過(guò) Kafka 的方式接入進(jìn)來(lái)，經(jīng)過(guò)統(tǒng)一的 ETL 邏輯，接著把數(shù)據(jù)放在數(shù)據(jù)湖里。因?yàn)閿?shù)據(jù)湖本身可以同時(shí)支持流和批的方式進(jìn)行讀寫(xiě)，而且數(shù)據(jù)湖本身可以實(shí)時(shí)消費(fèi)，所以它既可以做實(shí)時(shí)計(jì)算，也可以做離線(xiàn)計(jì)算，然后統(tǒng)一把數(shù)據(jù)再寫(xiě)回?cái)?shù)據(jù)湖。

前文提到在做查詢(xún)的時(shí)候，會(huì)使用離線(xiàn)跟實(shí)時(shí)做統(tǒng)一整合，所以在數(shù)據(jù)湖里寫(xiě)同一個(gè)表，在存儲(chǔ)層面可以省去很多工作，另外也可以節(jié)省存儲(chǔ)空間。

■?1.8 數(shù)據(jù)生產(chǎn) - SQL 化

SQL 化是 Talos 實(shí)時(shí)數(shù)倉(cāng)平臺(tái)提供的能力。

從頁(yè)面上來(lái)看，它包括了幾個(gè)功能，左邊是項(xiàng)目管理，右邊包括 Source、Transform 和 Sink。

• 有一些業(yè)務(wù)團(tuán)隊(duì)本身對(duì)于計(jì)算引擎算子非常熟，那么他們便可以做一些代碼開(kāi)發(fā)；

  
  

• 但是很多業(yè)務(wù)團(tuán)隊(duì)可能對(duì)引擎并不是那么了解，或者沒(méi)有強(qiáng)烈的意愿去了解，他們就可以通過(guò)這種可視化的方式，拼接出一個(gè)作業(yè)。

例如，可以拖一個(gè) Kafka 的數(shù)據(jù)源進(jìn)來(lái)，在上面做數(shù)據(jù)過(guò)濾，然后就可以拖一個(gè) Filter 算子達(dá)到過(guò)濾邏輯，后面可以再去做一些 Project，Union 的計(jì)算，最后輸出到某個(gè)地方就可以了。

對(duì)于能力稍微高一些的同學(xué)，可以去做一些更高層面的計(jì)算。這里也可以實(shí)現(xiàn)到實(shí)時(shí)數(shù)倉(cāng)的目的，在里面創(chuàng)建一些數(shù)據(jù)源，然后通過(guò) SQL 的方式，把邏輯表示出來(lái)，最終把這個(gè)數(shù)據(jù)輸出到某種存儲(chǔ)。

上面是從開(kāi)發(fā)層面來(lái)講，在系統(tǒng)層面上，它其實(shí)還提供了一些其他的功能，比如規(guī)則校驗(yàn)，還有開(kāi)發(fā)/測(cè)試/上線(xiàn)，在這里可以統(tǒng)一管理。此外還有監(jiān)控，對(duì)線(xiàn)上跑的實(shí)時(shí)任務(wù)有很多實(shí)時(shí)指標(biāo)，可以通過(guò)查看這些指標(biāo)來(lái)判斷當(dāng)前的任務(wù)是不是正常的狀態(tài)。

2. 特征工程

特征工程有兩方面的需求：

• 第一個(gè)需求是實(shí)時(shí)化，因?yàn)閿?shù)據(jù)價(jià)值隨著時(shí)間的遞增會(huì)越來(lái)越低。比如某用戶(hù)表現(xiàn)出來(lái)的觀影行為是喜歡看兒童內(nèi)容，平臺(tái)就會(huì)推薦兒童相關(guān)的廣告。另外，用戶(hù)在看廣告過(guò)程中，會(huì)有一些正/負(fù)反饋的行為，如果把這些數(shù)據(jù)實(shí)時(shí)迭代到特征里，就可以有效提升后續(xù)的轉(zhuǎn)化效果。
  
  

實(shí)時(shí)化的另一個(gè)重點(diǎn)是準(zhǔn)確性，之前很多特征工程是離線(xiàn)的，在生產(chǎn)環(huán)節(jié)里面存在計(jì)算時(shí)的數(shù)據(jù)跟投放過(guò)程中的特征有偏差，基礎(chǔ)特征數(shù)據(jù)不是很準(zhǔn)確，因此我們要求數(shù)據(jù)要更實(shí)時(shí)、更準(zhǔn)確。

• 特征工程的第二個(gè)需求是服務(wù)穩(wěn)定性。

• 首先是作業(yè)容錯(cuò)，比如作業(yè)在異常的時(shí)候能否正常恢復(fù)；
• 另外是數(shù)據(jù)質(zhì)量，在實(shí)時(shí)數(shù)據(jù)里追求端到端精確一次。

■?2.1 點(diǎn)擊率預(yù)估

下面是在特征實(shí)時(shí)化里的實(shí)踐，首先是點(diǎn)擊率預(yù)估的需求。

點(diǎn)擊率預(yù)估案例的背景如上所示，從投放鏈路上來(lái)說(shuō)，在廣告前端用戶(hù)產(chǎn)生觀影行為，前端會(huì)向廣告引擎請(qǐng)求廣告，然后廣告引擎在做廣告召回粗排/精排的時(shí)候會(huì)拿到用戶(hù)特征和廣告特征。把廣告返回給前端之后，后續(xù)用戶(hù)行為可能產(chǎn)生曝光、點(diǎn)擊等行為事件，在做點(diǎn)擊率預(yù)估的時(shí)候，需要把前面請(qǐng)求階段的特征跟后續(xù)用戶(hù)行為流里的曝光和點(diǎn)擊關(guān)聯(lián)起來(lái)，形成一個(gè) Session 數(shù)據(jù)，這就是我們的數(shù)據(jù)需求。

落實(shí)到具體實(shí)踐的話(huà)包括兩方面：

• 一方面是 Tracking 流里曝光、點(diǎn)擊事件的關(guān)聯(lián)；

  
  

• 另一方面是特征流跟用戶(hù)行為的關(guān)聯(lián)。

在實(shí)踐過(guò)程中有哪些挑戰(zhàn)？

• 第一個(gè)挑戰(zhàn)是數(shù)據(jù)量；

  
  

• 第二個(gè)挑戰(zhàn)是實(shí)時(shí)數(shù)據(jù)亂序和延遲；

  
  

• 第三個(gè)挑戰(zhàn)是精確性要求高。

在時(shí)序上來(lái)說(shuō)，特征肯定是早于 Tracking，但是兩個(gè)流成功關(guān)聯(lián)率在 99% 以上的時(shí)候，這個(gè)特征需要保留多久？因?yàn)樵趶V告業(yè)務(wù)中，用戶(hù)可以離線(xiàn)下載一個(gè)內(nèi)容，在下載的時(shí)候就已經(jīng)完成了廣告請(qǐng)求和返回了。但是后續(xù)如果用戶(hù)在沒(méi)有網(wǎng)的情況下觀看，這個(gè)事件并不會(huì)立馬返回，只有當(dāng)狀態(tài)恢復(fù)的時(shí)候，才會(huì)有后續(xù)曝光和點(diǎn)擊事件回傳。

所以這個(gè)時(shí)候，其實(shí)特征流和 Tracking 的時(shí)間概括是非常長(zhǎng)的。我們經(jīng)過(guò)離線(xiàn)的數(shù)據(jù)分析，如果兩個(gè)流的關(guān)聯(lián)率達(dá) 99% 以上，那么特征數(shù)據(jù)就需要保留比較長(zhǎng)的時(shí)間，目前是保留 7 天，這個(gè)量級(jí)還是比較大的。

上圖為點(diǎn)擊率預(yù)測(cè)的整體架構(gòu)，剛才我們提到關(guān)聯(lián)包括兩部分：

• 第一個(gè)部分是用戶(hù)行為流里曝光跟點(diǎn)擊事件的關(guān)聯(lián)，這里通過(guò) CEP 實(shí)現(xiàn)。

• 第二個(gè)部分是兩個(gè)流的關(guān)聯(lián)，前面介紹特征需要保留 7 天，它的狀態(tài)較大，已經(jīng)是上百 TB。這個(gè)量級(jí)在內(nèi)存里做管理，對(duì)數(shù)據(jù)穩(wěn)定性有比較大的影響，所以我們把特征數(shù)據(jù)放在一個(gè)外部存儲(chǔ) (Hbase) 里，然后和 HBase 特征做一個(gè)實(shí)時(shí)數(shù)據(jù)查詢(xún)，就可以達(dá)到這樣一個(gè)效果。

但是因?yàn)閮蓚€(gè)流的時(shí)序本身可能是錯(cuò)開(kāi)的，就是說(shuō)，當(dāng)曝光、點(diǎn)擊出現(xiàn)的時(shí)候，可能這個(gè)特征還沒(méi)有到，那么就拿不到這個(gè)特征。所以我們做了一個(gè)多級(jí)重試隊(duì)列，保證最終兩個(gè)流關(guān)聯(lián)的完整性。

■?2.2 點(diǎn)擊率預(yù)估 - 流內(nèi)事件關(guān)聯(lián)

上圖右邊是更細(xì)的講解，闡述了流內(nèi)事件關(guān)聯(lián)為什么選擇 CEP 方案。業(yè)務(wù)需求是把用戶(hù)行為流里屬于同一次廣告請(qǐng)求，并且是同一個(gè)廣告的曝光跟點(diǎn)擊關(guān)聯(lián)起來(lái)。曝光之后，比如 5 分鐘之內(nèi)產(chǎn)生點(diǎn)擊，作為一個(gè)正樣本，5 分鐘之后出現(xiàn)的點(diǎn)擊則拋棄不要了。

可以想象一下，當(dāng)遇到這樣的場(chǎng)景，通過(guò)什么樣的方案可以實(shí)現(xiàn)這樣的效果。其實(shí)在一個(gè)流里多個(gè)事件的處理，可以用窗口來(lái)實(shí)現(xiàn)。但窗口的問(wèn)題是：

• 如果事件序列本身都在同一個(gè)窗口之內(nèi)，數(shù)據(jù)沒(méi)有問(wèn)題；

  
  

• 但是當(dāng)事件序列跨窗口的時(shí)候，是達(dá)不到正常關(guān)聯(lián)效果的。

所以當(dāng)時(shí)經(jīng)過(guò)很多技術(shù)調(diào)研后，發(fā)現(xiàn) Flink 里的 CEP 可以實(shí)現(xiàn)這樣的效果，用類(lèi)似政策匹配的方式，描述這些序列需要滿(mǎn)足哪些匹配方式。另外它可以指定一個(gè)時(shí)間窗口，比如曝光和點(diǎn)擊間隔 15 分鐘。

上圖左邊是匹配規(guī)則的描述，begin 里定義一個(gè)曝光，實(shí)現(xiàn)曝光之后 5 分鐘之內(nèi)的點(diǎn)擊，后面是描述一個(gè)可以出現(xiàn)多次的點(diǎn)擊，within 表示關(guān)聯(lián)窗口是多長(zhǎng)時(shí)間。

在生產(chǎn)實(shí)踐過(guò)程中，這個(gè)方案大部分情況下可以關(guān)聯(lián)上，但是在做數(shù)據(jù)對(duì)比的時(shí)候，才發(fā)現(xiàn)存在某些曝光點(diǎn)擊沒(méi)有正常關(guān)聯(lián)到。

經(jīng)過(guò)數(shù)據(jù)分析，發(fā)現(xiàn)這些數(shù)據(jù)本身的特點(diǎn)是曝光跟點(diǎn)擊的時(shí)間戳都是毫秒級(jí)別，當(dāng)它們有相同毫秒時(shí)間戳的時(shí)候，這個(gè)事件就不能正常匹配。于是我們采用一個(gè)方案，人為地對(duì)于點(diǎn)擊事件加一毫秒，進(jìn)行人工錯(cuò)位，這樣就保證曝光跟點(diǎn)擊能夠成功關(guān)聯(lián)上。

■?2.3 點(diǎn)擊率預(yù)估-雙流關(guān)聯(lián)

前文提到特征數(shù)據(jù)需要保留 7 天，所以狀態(tài)是上百 TB。需要把數(shù)據(jù)放在一個(gè)外部存儲(chǔ)里，因此在做技術(shù)選型時(shí)對(duì)外部存儲(chǔ)有一定的要求：

• 首先支持比較高的讀寫(xiě)并發(fā)能力；

  
  

• 另外它的時(shí)效性需要非常低；

  
  

• 同時(shí)因?yàn)閿?shù)據(jù)要保留 7 天，所以它最好具備生命周期管理能力。

基于以上幾個(gè)點(diǎn)，最終選擇了 HBase，形成上圖的解決方案。

上面一行表示通過(guò) CEP 之后把曝光點(diǎn)擊序列關(guān)聯(lián)在一起，最下面是把特征流通過(guò) Flink 寫(xiě)到 HBase 里，去做外部狀態(tài)存儲(chǔ)，中間核心模塊是用于達(dá)到兩個(gè)流的關(guān)聯(lián)。拿到曝光點(diǎn)擊關(guān)聯(lián)之后去查 HBase 數(shù)據(jù)，如果能夠正常查到，就會(huì)把它輸出到一個(gè)正常結(jié)果流里。而對(duì)于那些不能構(gòu)成關(guān)聯(lián)的數(shù)據(jù)，做了一個(gè)多級(jí)重試隊(duì)列，在多次重試的時(shí)候會(huì)產(chǎn)生隊(duì)列降級(jí)，并且在重試的時(shí)候?yàn)榱藴p輕對(duì) HBase 的掃描壓力，重試 Gap 會(huì)逐級(jí)增加。

另外還有一個(gè)退出機(jī)制，因?yàn)橹卦嚥皇菬o(wú)限進(jìn)行的。退出機(jī)制的存在原因主要包括兩個(gè)點(diǎn)：

• 第一點(diǎn)是特征數(shù)據(jù)保留了 7 天，如果對(duì)應(yīng)特征是在 7 天之前，那么它本身是關(guān)聯(lián)不到的。

• 另外在廣告業(yè)務(wù)里，存在一些外部的刷量行為，比如刷曝光或刷點(diǎn)擊，但它本身并沒(méi)有真實(shí)存在的廣告請(qǐng)求，所以這種場(chǎng)景也拿不到對(duì)應(yīng)特征。

因此，退出機(jī)制意味著在重試多次之后就會(huì)過(guò)期，然后會(huì)到重試過(guò)期的數(shù)據(jù)里。

■?2.4 有效點(diǎn)擊

在有效點(diǎn)擊場(chǎng)景里，其實(shí)也是兩個(gè)流的關(guān)聯(lián)，但是兩個(gè)場(chǎng)景里的技術(shù)選型是完全不一樣的。

首先看一下項(xiàng)目背景，在網(wǎng)大場(chǎng)景里，影片本身就是一個(gè)廣告。用戶(hù)在點(diǎn)擊之后，就會(huì)進(jìn)入到一個(gè)播放頁(yè)面。在播放頁(yè)面里，用戶(hù)可以免費(fèi)觀看 6 分鐘，6 分鐘之后想要繼續(xù)觀看，需要是會(huì)員或者購(gòu)買(mǎi)才行，在這里需要統(tǒng)計(jì)的數(shù)據(jù)是有效點(diǎn)擊，定義是在點(diǎn)擊之后觀影時(shí)長(zhǎng)超過(guò) 6 分鐘即可。

這種場(chǎng)景落實(shí)到技術(shù)上是兩個(gè)流的關(guān)聯(lián)，包括了點(diǎn)擊流和播放心跳流。

• 點(diǎn)擊流比較好理解，包括用戶(hù)的曝光和點(diǎn)擊等行為，從里面篩選點(diǎn)擊事件即可。

• 播放行為流是在用戶(hù)觀看的過(guò)程，會(huì)定時(shí)地把心跳信息回傳，比如三秒鐘回傳一個(gè)心跳，表明用戶(hù)在持續(xù)觀看。在定義時(shí)長(zhǎng)超過(guò) 6 分鐘的時(shí)候，需要把這個(gè)狀態(tài)本身做一些處理，才能滿(mǎn)足 6 分鐘的條件。

在這個(gè)場(chǎng)景里，兩個(gè)流動(dòng) Gap 相對(duì)比較小，而在電影里時(shí)長(zhǎng)一般是兩個(gè)多小時(shí)，所以點(diǎn)擊之后的行為，Gap 基本是在三個(gè)小時(shí)以?xún)?nèi)才能完成，因此這里本身的狀態(tài)是相對(duì)比較小的，使用 Flink 的狀態(tài)管理可以達(dá)到這樣的效果。

接下來(lái)我們看一個(gè)具體的方案。

從流上來(lái)看，綠色部分是點(diǎn)擊流，藍(lán)色部分是播放心跳流。

• 在左邊的狀態(tài)里面，一個(gè)點(diǎn)擊事件進(jìn)來(lái)之后，會(huì)對(duì)這個(gè)點(diǎn)擊做一個(gè)狀態(tài)記錄，同時(shí)會(huì)注冊(cè)一個(gè)定時(shí)器做定期清理，定時(shí)器是三個(gè)小時(shí)。因?yàn)榇蟛糠钟捌臅r(shí)長(zhǎng)在三小時(shí)以?xún)?nèi)，如果這個(gè)時(shí)候?qū)?yīng)的播放事件還沒(méi)有一個(gè)目標(biāo)狀態(tài)，點(diǎn)擊事件基本就可以過(guò)期了。

• 在右邊的播放心跳流里，這個(gè)狀態(tài)是對(duì)時(shí)長(zhǎng)做累計(jì)，它本身是一個(gè)心跳流，比如每三秒傳一個(gè)心跳過(guò)來(lái)。我們需要在這里做一個(gè)計(jì)算，看它累計(jì)播放時(shí)長(zhǎng)是不是達(dá)到 6 分鐘了，另外也看當(dāng)前記錄是不是到了 6 分鐘。對(duì)應(yīng) Flink 里的一個(gè)實(shí)現(xiàn)就是把兩個(gè)流通過(guò) Connect 算子關(guān)系在一起，然后可以制定一個(gè) CoProcessFunction，在這里面有兩個(gè)核心算子。




• 第一個(gè)算子是拿到狀態(tài) 1 的流事件之后，需要做一些什么樣的處理；
• 第二個(gè)算子是拿到第 2 個(gè)流事件之后，可以自定義哪些功能。




算子給用戶(hù)提供了很多靈活性，用戶(hù)可以在里面做很多邏輯控制。相比很多的 Input Join，用戶(hù)可發(fā)揮的空間比較大。

■?2.5 特征工程 - 小結(jié)

針對(duì)以上案例做一個(gè)小結(jié)。現(xiàn)在雙流管理已經(jīng)非常普遍，有許多方案可以選擇，比如 Window join，Interval join，還有我們使用的 Connect + CoProcessFunction。除此之外，還有一些用戶(hù)自定義的方案。

在選型的時(shí)候，建議從業(yè)務(wù)出發(fā)，去做對(duì)應(yīng)的技術(shù)選型。首先要思考多個(gè)流之間的事件關(guān)系，然后判斷出狀態(tài)是什么規(guī)模，一定程度上可以從上面很多方案里排除不可行的方案。

三、Flink 使用過(guò)程中的問(wèn)題及解決

1. 容錯(cuò)

在 Flink 內(nèi)部主要是通過(guò) Checkpoint 做容錯(cuò)，Checkpoint 本身是對(duì)于 Job 內(nèi)部的 Task 級(jí)別的容錯(cuò)，但是當(dāng) Job 主動(dòng)或異常重啟時(shí)，狀態(tài)無(wú)法從歷史狀態(tài)恢復(fù)。

因此我們這邊做了一個(gè)小的改進(jìn)，就是一個(gè)作業(yè)在啟動(dòng)的時(shí)候，它也會(huì)去 Checkpoint 里把最后一次成功的歷史狀態(tài)拿到，然后做初始化管理，這樣就達(dá)到狀態(tài)恢復(fù)的效果。

2. 數(shù)據(jù)質(zhì)量

Flink 本身實(shí)現(xiàn)端到端精確一次，首先需要開(kāi)啟 Checkpoint 功能，并且在 Checkpoint 里指定精確一次的語(yǔ)義。另外，如果在下游比如 Sink 端，它本身支持事務(wù)，就可以結(jié)合兩階段提交與 Checkpoint 以及下游的事務(wù)做聯(lián)動(dòng)，達(dá)到端到端精確一次。

在上圖右邊就是描述了這個(gè)過(guò)程。這是一個(gè)預(yù)提交的過(guò)程，就是 Checkpoint 協(xié)調(diào)器在做 Checkpoint 的時(shí)候，會(huì)往 Source 端注入一些 Barrier 數(shù)據(jù)，每個(gè) Source 拿到 Barrier 之后會(huì)做狀態(tài)存儲(chǔ)，然后把完成狀態(tài)反饋給協(xié)調(diào)器。這樣每個(gè)算子拿到 Barrier，其實(shí)是做相同的一個(gè)功能。

到 Sink 端之后，它會(huì)在 Kafka 里提交一個(gè)預(yù)提交標(biāo)記，后面主要是 Kafka 本身事務(wù)機(jī)制來(lái)保證的。在所有的算子都完成 Checkpoint 之后，協(xié)調(diào)器會(huì)給所有的算子發(fā)一個(gè) ACK，發(fā)送一個(gè)確認(rèn)狀態(tài)，這時(shí)候 Sink 端做一個(gè)提交動(dòng)作就可以了。

3. Sink Kafka

在之前的實(shí)踐中我們發(fā)現(xiàn)，下游 Kafka 增加分區(qū)數(shù)時(shí)，新增分區(qū)無(wú)數(shù)據(jù)寫(xiě)入。

原理是 FlinkKafkaProducer 默認(rèn)使用 FlinkFixedPartitioner，每個(gè) Task 只會(huì)發(fā)送到下游對(duì)應(yīng)的一個(gè) Partition 中，如果下游 Kafka 的 Topic 的 Partition 大于當(dāng)前任務(wù)的并行度，就會(huì)出現(xiàn)該問(wèn)題。

解決辦法有兩個(gè)：

• 第一個(gè)辦法是用戶(hù)自定義一個(gè) FlinkKafkaPartitioner；

• 另一個(gè)辦法是默認(rèn)不配置，默認(rèn)輪詢(xún)寫(xiě)入各個(gè) Partition。

4. 監(jiān)控加強(qiáng)

對(duì)于運(yùn)行中的 Flink 作業(yè)，我們需要查看它本身的一些狀態(tài)。比如在 Flink UI 里面，它的很多指標(biāo)都是在 Task 粒度，沒(méi)有整體的效果。

平臺(tái)這邊對(duì)這些指標(biāo)做了進(jìn)一步的聚合，統(tǒng)一在一個(gè)頁(yè)面里面展示。

從上圖可以看到，展示信息包括反壓狀態(tài)，時(shí)延情況以及運(yùn)行過(guò)程中 JobManager 和 TaskManage 的 CPU / 內(nèi)存的利用率。另外還有 Checkpoint 的監(jiān)控，比如它是否超時(shí)，最近是否有 Checkpoint 已經(jīng)失敗了，后面我們會(huì)針對(duì)這些監(jiān)控指標(biāo)做一些報(bào)警通知。

5. 監(jiān)控報(bào)警

當(dāng)實(shí)時(shí)任務(wù)運(yùn)營(yíng)異常的時(shí)候，用戶(hù)是需要及時(shí)知道這個(gè)狀態(tài)的，如上圖所示，有一些報(bào)警項(xiàng)，包括報(bào)警訂閱人、報(bào)警級(jí)別，下面還有一些指標(biāo)，根據(jù)前面設(shè)置的指標(biāo)值，如果滿(mǎn)足這些報(bào)警策略規(guī)則，就會(huì)給報(bào)警訂閱人推送報(bào)警，報(bào)警方式包括郵件、電話(huà)以及內(nèi)部通訊工具，從而實(shí)現(xiàn)任務(wù)異常狀態(tài)通知。

通過(guò)這種方式，當(dāng)任務(wù)異常的時(shí)候，用戶(hù)可以及時(shí)知曉這個(gè)狀態(tài)，然后進(jìn)行人為干預(yù)。

6. 實(shí)時(shí)數(shù)據(jù)生產(chǎn)

最后總結(jié)一下愛(ài)奇藝廣告業(yè)務(wù)在實(shí)時(shí)鏈路生產(chǎn)上面的關(guān)鍵節(jié)點(diǎn)。

• 我們的實(shí)時(shí)是從 2016 年開(kāi)始起步，當(dāng)時(shí)主要功能點(diǎn)是做一些指標(biāo)實(shí)時(shí)化，使用的是 SparkStreaming；

  
  

• 2018 年上線(xiàn)了點(diǎn)擊率實(shí)時(shí)特征；

  
  

• 2019 年上線(xiàn)了 Flink 的端到端精確到一次和監(jiān)控強(qiáng)化。

  
  

• 2020 年上線(xiàn)了有效點(diǎn)擊實(shí)時(shí)特征；

  
  

• 同年10月，逐步推進(jìn)實(shí)時(shí)數(shù)倉(cāng)的改進(jìn)，把 API 生產(chǎn)方式逐漸 SQL 化；

• 2021 年 4 月，進(jìn)行流批一體的探索，目前先把流批一體放在 ETL 實(shí)現(xiàn)。

之前我們的 ETL 實(shí)時(shí)跟離線(xiàn)是分別做的，通過(guò)批處理的方式，然后換到 Hive 表里邊，后面跟的是離線(xiàn)數(shù)倉(cāng)。在實(shí)時(shí)里，經(jīng)過(guò)實(shí)時(shí) ETL，放到 Kafka 里邊，然后去做后續(xù)的實(shí)時(shí)數(shù)倉(cāng)。

先在 ETL 做流批一體的第一個(gè)好處是離線(xiàn)數(shù)倉(cāng)時(shí)效性提升，因?yàn)閿?shù)據(jù)需要做反作弊，所以我們給廣告算法提供基礎(chǔ)特征的時(shí)候，反作弊之后的時(shí)效性對(duì)于后續(xù)整體效果的提升是比較大的，所以如果把 ETL 做成統(tǒng)一實(shí)時(shí)化之后，對(duì)于后續(xù)的指導(dǎo)意義非常大。

ETL 做到流批一體之后，我們會(huì)把數(shù)據(jù)放在數(shù)據(jù)湖里面，后續(xù)離線(xiàn)數(shù)倉(cāng)和實(shí)時(shí)數(shù)倉(cāng)都可以基于數(shù)據(jù)湖實(shí)現(xiàn)。流批一體可以分為兩個(gè)階段，第一階段是先把 ETL 做到一體，另外報(bào)表端也可以放在數(shù)據(jù)湖里邊，這樣我們的查詢(xún)服務(wù)可以做到一個(gè)更新的量級(jí)。因?yàn)橹靶枰x線(xiàn)表跟實(shí)時(shí)表做一個(gè) Union 的計(jì)算，在數(shù)據(jù)湖里面，我們通過(guò)離線(xiàn)和實(shí)時(shí)寫(xiě)一個(gè)表就可以實(shí)現(xiàn)了。

四、未來(lái)規(guī)劃

關(guān)于未來(lái)規(guī)劃：

• 首先是流批一體，這里包括兩個(gè)方面：




• 第一個(gè)是 ETL 一體，目前已經(jīng)是基本達(dá)到可線(xiàn)上的狀態(tài)。
• 第二個(gè)是實(shí)時(shí)報(bào)表 SQL 化和數(shù)據(jù)湖的結(jié)合。

  
  

• 另外，現(xiàn)在的反作弊主要是通過(guò)離線(xiàn)的方式實(shí)現(xiàn)，后面可能會(huì)把一些線(xiàn)上的反
  作弊模型轉(zhuǎn)成實(shí)時(shí)化，把風(fēng)險(xiǎn)降到最低。