一.OPPO 實(shí)時(shí)數(shù)倉(cāng)的演進(jìn)思路

1.1.OPPO 業(yè)務(wù)與數(shù)據(jù)規(guī)模

大家都知道 OPPO 是做智能手機(jī)的，但并不知道 OPPO 與互聯(lián)網(wǎng)以及大數(shù)據(jù)有什么關(guān)系，下圖概要介紹了 OPPO 的業(yè)務(wù)與數(shù)據(jù)情況： 

OPPO 作為手機(jī)廠商，基于 Android 定制了自己的 ColorOS 系統(tǒng)，當(dāng)前日活躍用戶超過 2 億。圍繞 ColorOS，OPPO 構(gòu)建了很多互聯(lián)網(wǎng)應(yīng)用，比如應(yīng)用商店、瀏覽器、信息流等。

在運(yùn)營(yíng)這些互聯(lián)網(wǎng)應(yīng)用的過程中，OPPO 積累了大量的數(shù)據(jù)，上圖右邊是整體數(shù)據(jù)規(guī)模的演進(jìn)：從 2012 年開始每年都是 2~3 倍的增長(zhǎng)速度，截至目前總數(shù)據(jù)量已經(jīng)超過 100PB，日增數(shù)據(jù)量超過 200TB。

要支撐這么大的一個(gè)數(shù)據(jù)量，OPPO 研發(fā)出一整套的數(shù)據(jù)系統(tǒng)與服務(wù)，并逐漸形成了自己的數(shù)據(jù)中臺(tái)體系。

1.2.OPPO 數(shù)據(jù)中臺(tái) 

今年大家都在談數(shù)據(jù)中臺(tái)，OPPO 是如何理解數(shù)據(jù)中臺(tái)的呢?我們把它分成了 4 個(gè)層次：

最下層是統(tǒng)一工具體系，涵蓋了"接入 - 治理 - 開發(fā) - 消費(fèi)"全數(shù)據(jù)鏈路;

基于工具體系之上構(gòu)建了數(shù)據(jù)倉(cāng)庫(kù)，劃分成"原始層 - 明細(xì)層 - 匯總層 - 應(yīng)用層"，這也是經(jīng)典的數(shù)倉(cāng)架構(gòu);

再往上是全域的數(shù)據(jù)體系，什么是全域呢?就是把公司所有的業(yè)務(wù)數(shù)據(jù)都打通，形成統(tǒng)一的數(shù)據(jù)資產(chǎn)，比如 ID-Mapping、用戶標(biāo)簽等;

最終，數(shù)據(jù)要能被業(yè)務(wù)用起來，需要場(chǎng)景驅(qū)動(dòng)的數(shù)據(jù)產(chǎn)品與服務(wù)。

以上就是 OPPO 數(shù)據(jù)中臺(tái)的整個(gè)體系，而數(shù)據(jù)倉(cāng)庫(kù)在其中處于非常基礎(chǔ)與核心的位置。

1.3. 構(gòu)建 OPPO 離線數(shù)倉(cāng) 

過往 2、3 年，我們的重點(diǎn)聚焦在離線數(shù)倉(cāng)的構(gòu)建。上圖大致描述了整個(gè)構(gòu)建過程：首先，數(shù)據(jù)來源基本是手機(jī)、日志文件以及 DB 數(shù)據(jù)庫(kù)，我們基于 Apache NiFi 打造了高可用、高吞吐的接入系統(tǒng)，將數(shù)據(jù)統(tǒng)一落入 HDFS，形成原始層;緊接著，基于 Hive 的小時(shí)級(jí) ETL 與天級(jí)匯總 Hive 任務(wù)，分別負(fù)責(zé)計(jì)算生成明細(xì)層與匯總層;

最后，應(yīng)用層是基于 OPPO 內(nèi)部研發(fā)的數(shù)據(jù)產(chǎn)品，主要是報(bào)表分析、用戶畫像以及接口服務(wù)。此外，中間的明細(xì)層還支持基于 Presto 的即席查詢與自助提數(shù)。 伴隨著離線數(shù)倉(cāng)的逐步完善，業(yè)務(wù)對(duì)實(shí)時(shí)數(shù)倉(cāng)的訴求也愈發(fā)強(qiáng)烈。

1.5. 離線到實(shí)時(shí)的平滑遷移 

無論是一個(gè)平臺(tái)還是一個(gè)系統(tǒng)，都離不開上下兩個(gè)層次的構(gòu)成：上層是 API，是面向用戶的編程抽象與接口;下層是 Runtime，是面向內(nèi)核的執(zhí)行引擎。我們希望從離線到實(shí)時(shí)的遷移是平滑的，是什么意思呢?從 API 這層來看，數(shù)倉(cāng)的抽象是 Table、編程接口是 SQL+UDF，離線數(shù)倉(cāng)時(shí)代用戶已經(jīng)習(xí)慣了這樣的 API，遷移到實(shí)時(shí)數(shù)倉(cāng)后最好也能保持一致。而從 Runtime 這層來看，計(jì)算引擎從 Hive 演進(jìn)到了 Flink，存儲(chǔ)引擎從 HDFS 演進(jìn)到了 Kafka。

基于以上的思路，只需要把之前提到的離線數(shù)倉(cāng) pipeline 改造下，就得到了實(shí)時(shí)數(shù)倉(cāng) pipeline。

1.6. 構(gòu)建 OPPO 實(shí)時(shí)數(shù)倉(cāng)

從上圖可以看到，整個(gè) pipeline 與離線數(shù)倉(cāng)基本相似，只是把 Hive 替換為 Flink，把 HDFS 替換為 Kafka。從總體流程來看，基本模型是不變的，還是由原始層、明細(xì)層、匯總層、應(yīng)用層的級(jí)聯(lián)計(jì)算來構(gòu)成。

因此，這里的核心問題是如何基于 Flink 構(gòu)建出這個(gè) pipeline，下面就介紹下我們基于 Flink SQL 所做的一些工作。

二. 基于 Flink SQL 的擴(kuò)展工作

2.1.Why Flink SQL

首先，為什么要用 Flink SQL? 下圖展示了 Flink 框架的基本結(jié)構(gòu)，最下面是 Runtime，這個(gè)執(zhí)行引擎我們認(rèn)為最核心的優(yōu)勢(shì)是四個(gè)：第一，低延遲，高吞吐;第二，端到端的 Exactly-once;第三，可容錯(cuò)的狀態(tài)管理;第四，Window & Event time 的支持。基于 Runtime 抽象出 3 個(gè)層次的 API，SQL 處于最上層。

Flink SQL API 有哪些優(yōu)勢(shì)呢?我們也從四個(gè)方面去看：第一，支持 ANSI SQL 的標(biāo)準(zhǔn);第二，支持豐富的數(shù)據(jù)類型與內(nèi)置函數(shù)，包括常見的算術(shù)運(yùn)算與統(tǒng)計(jì)聚合;第三，可自定義 Source/Sink，基于此可以靈活地?cái)U(kuò)展上下游;第四，批流統(tǒng)一，同樣的 SQL，既可以跑離線也可以跑實(shí)時(shí)。 

那么，基于 Flink SQL API 如何編程呢?下面是一個(gè)簡(jiǎn)單的演示： 

首先是定義與注冊(cè)輸入 / 輸出表，這里創(chuàng)建了 2 張 Kakfa 的表，指定 kafka 版本是什么、對(duì)應(yīng)哪個(gè) topic;接下來是注冊(cè) UDF，篇幅原因這里沒有列出 UDF 的定義;最后是才是執(zhí)行真正的 SQL。可以看到，為了執(zhí)行 SQL，需要做這么多的編碼工作，這并不是我們希望暴露給用戶的接口。

2.2. 基于 WEB 的開發(fā) IDE 

2.5.Flink SQL 對(duì)接外部數(shù)據(jù)源

搞清楚了 Flink SQL 注冊(cè)庫(kù)表的過程，給我們帶來這樣一個(gè)思路：如果外部元數(shù)據(jù)創(chuàng)建的表也能被轉(zhuǎn)換成 TableFactory 可識(shí)別的 map，那么就能被無縫地注冊(cè)到 TableEnvironment。基于這個(gè)思路，我們實(shí)現(xiàn)了 Flink SQL 與已有元數(shù)據(jù)中心的對(duì)接，大致過程參見下圖： 

通過元數(shù)據(jù)中心創(chuàng)建的表，都會(huì)將元數(shù)據(jù)信息存儲(chǔ)到 MySQL，我們用一張表來記錄 Table 的基本信息，然后另外三張表分別記錄 Connector、Format、Schema 轉(zhuǎn)換成 key-value 后的描述信息。之所以拆開成三張表，是為了能夠能獨(dú)立的更新這三種描述信息。接下來是定制實(shí)現(xiàn)的 ExternalCatalog，能夠讀取 MySQL 這四張表，并轉(zhuǎn)換成 map 結(jié)構(gòu)。

2.6. 實(shí)時(shí)表 - 維表關(guān)聯(lián)

到目前為止，我們的平臺(tái)已經(jīng)具備了元數(shù)據(jù)管理與 SQL 作業(yè)管理的能力，但是要真正開放給用戶使用，還有一點(diǎn)基本特性存在缺失。通過我們?nèi)?gòu)建數(shù)倉(cāng)，星型模型是無法避免的。這里有一個(gè)比較簡(jiǎn)單的案例：中間的事實(shí)表記錄了廣告點(diǎn)擊流，周邊是關(guān)于用戶、廣告、產(chǎn)品、渠道的維度表。

假定我們有一個(gè) SQL 分析，需要將點(diǎn)擊流表與用戶維表進(jìn)行關(guān)聯(lián)，這個(gè)目前在 Flink SQL 中應(yīng)該怎么來實(shí)現(xiàn)?我們有兩種實(shí)現(xiàn)方式，一個(gè)基于 UDF，一個(gè)基于 SQL 轉(zhuǎn)換。 

三.構(gòu)建實(shí)時(shí)數(shù)倉(cāng)的應(yīng)用案例

下面分享幾個(gè)典型的應(yīng)用案例，都是在我們的平臺(tái)上用 Flink SQL 來實(shí)現(xiàn)的。

3.1. 實(shí)時(shí) ETL 拆分

這里是一個(gè)典型的實(shí)時(shí) ETL 鏈路，從大表中拆分出各業(yè)務(wù)對(duì)應(yīng)的小表： 

OPPO 的最大數(shù)據(jù)來源是手機(jī)端埋點(diǎn)，從手機(jī) APP 過來的數(shù)據(jù)有一個(gè)特點(diǎn)，所有的數(shù)據(jù)是通過統(tǒng)一的幾個(gè)通道上報(bào)過來。因?yàn)椴豢赡苊恳淮螛I(yè)務(wù)有新的埋點(diǎn)，都要去升級(jí)客戶端，去增加新的通道。比如我們有個(gè) sdk_log 通道，所有 APP 應(yīng)用的埋點(diǎn)都往這個(gè)通道上報(bào)數(shù)據(jù)，導(dǎo)致這個(gè)通道對(duì)應(yīng)的原始層表巨大，一天幾十個(gè) TB。但實(shí)際上，每個(gè)業(yè)務(wù)只關(guān)心它自身的那部分?jǐn)?shù)據(jù)，這就要求我們?cè)谠紝舆M(jìn)行 ETL 拆分。

這個(gè) SQL 邏輯比較簡(jiǎn)單，無非是根據(jù)某些業(yè)務(wù)字段做篩選，插入到不同的業(yè)務(wù)表中去。它的特點(diǎn)是，多行 SQL 最終合并成一個(gè) SQL 提交給 Flink 執(zhí)行。大家擔(dān)心的是，包含了 4 個(gè) SQL，會(huì)不會(huì)對(duì)同一份數(shù)據(jù)重復(fù)讀取 4 次?其實(shí)，在 Flink 編譯 SQL 的階段是會(huì)做一些優(yōu)化的，因?yàn)樽罱K指向的是同一個(gè) kafka topic，所以只會(huì)讀取 1 次數(shù)據(jù)。

另外，同樣的 Flink SQL，我們同時(shí)用于離線與實(shí)時(shí)數(shù)倉(cāng)的 ETL 拆分，分別落入 HDFS 與 Kafka。Flink 中本身支持寫入 HDFS 的 Sink，比如 RollingFileSink。

3.2. 實(shí)時(shí)指標(biāo)統(tǒng)計(jì)

這里是一個(gè)典型的計(jì)算信息流 CTR 的這個(gè)案例，分別計(jì)算一定時(shí)間段內(nèi)的曝光與點(diǎn)擊次數(shù)，相除得到點(diǎn)擊率導(dǎo)入 Mysql，然后通過我們內(nèi)部的報(bào)表系統(tǒng)來可視化。這個(gè) SQL 的特點(diǎn)是它用到了窗口 (Tumbling Window) 以及子查詢。 

3.3. 實(shí)時(shí)標(biāo)簽導(dǎo)入

這里是一個(gè)實(shí)時(shí)標(biāo)簽導(dǎo)入的案例，手機(jī)端實(shí)時(shí)感知到當(dāng)前用戶的經(jīng)緯度，轉(zhuǎn)換成具體 POI 后導(dǎo)入 ES，最終在標(biāo)簽系統(tǒng)上做用戶定向。

這個(gè) SQL 的特點(diǎn)是用了 AggregateFunction，在 5 分鐘的窗口內(nèi)，我們只關(guān)心用戶最新一次上報(bào)的經(jīng)緯度。AggregateFunction 是一種 UDF 類型，通常是用于聚合指標(biāo)的統(tǒng)計(jì)，比如計(jì)算 sum 或者 average。在這個(gè)示例中，由于我們只關(guān)心最新的經(jīng)緯度，所以每次都替換老的數(shù)據(jù)即可。 

四. 未來工作的思考和展望

最后，給大家分享一下關(guān)于未來工作，我們的一些思考與規(guī)劃，還不是太成熟，拋出來和大家探討一下。

4.1. 端到端的實(shí)時(shí)流處理

什么是端到端?一端是采集到的原始數(shù)據(jù)，另一端是報(bào)表 / 標(biāo)簽 / 接口這些對(duì)數(shù)據(jù)的呈現(xiàn)與應(yīng)用，連接兩端的是中間實(shí)時(shí)流。當(dāng)前我們基于 SQL 的實(shí)時(shí)流處理，源表是 Kafka，目標(biāo)表也是 Kafka，統(tǒng)一經(jīng)過 Kafka 后再導(dǎo)入到 Druid/ES/HBase。

這樣設(shè)計(jì)的目的是提高整體流程的穩(wěn)定性與可用性：首先，kafka 作為下游系統(tǒng)的緩沖，可以避免下游系統(tǒng)的異常影響實(shí)時(shí)流的計(jì)算(一個(gè)系統(tǒng)保持穩(wěn)定，比起多個(gè)系統(tǒng)同時(shí)穩(wěn)定，概率上更高點(diǎn));其次，kafka 到 kafka 的實(shí)時(shí)流，exactly-once 語義是比較成熟的，一致性上有保證。

然后，上述的端到端其實(shí)是由割裂的三個(gè)步驟來完成的，每一步可能需要由不同角色人去負(fù)責(zé)處理：數(shù)據(jù)處理需要數(shù)據(jù)開發(fā)人員，數(shù)據(jù)導(dǎo)入需要引擎開發(fā)人員，數(shù)據(jù)資產(chǎn)化需要產(chǎn)品開發(fā)人員。 

我們的平臺(tái)能否把端到端給自動(dòng)化起來，只需要一次 SQL 提交就能打通處理、導(dǎo)入、資產(chǎn)化這三步?在這個(gè)思路下，數(shù)據(jù)開發(fā)中看到的不再是 Kafka Table，而應(yīng)該是面向場(chǎng)景的展示表 / 標(biāo)簽表 / 接口表。比如對(duì)于展示表，創(chuàng)建表的時(shí)候只要指定維度、指標(biāo)等字段，平臺(tái)會(huì)將實(shí)時(shí)流結(jié)果數(shù)據(jù)從 Kafka 自動(dòng)導(dǎo)入 Druid，再在報(bào)表系統(tǒng)自動(dòng)導(dǎo)入 Druid 數(shù)據(jù)源，甚至自動(dòng)生成報(bào)表模板。 

4.2. 實(shí)時(shí)流的血緣分析

關(guān)于血緣分析，做過離線數(shù)倉(cāng)的朋友都很清楚它的重要性，它在數(shù)據(jù)治理中都起著不可或缺的關(guān)鍵作用。對(duì)于實(shí)時(shí)數(shù)倉(cāng)來說也莫不如此。我們希望構(gòu)建端到端的血緣關(guān)系，從采集系統(tǒng)的接入通道開始，到中間流經(jīng)的實(shí)時(shí)表與實(shí)時(shí)作業(yè)，再到消費(fèi)數(shù)據(jù)的產(chǎn)品，都能很清晰地展現(xiàn)出來。基于血緣關(guān)系的分析，我們才能評(píng)估數(shù)據(jù)的應(yīng)用價(jià)值，核算數(shù)據(jù)的計(jì)算成本。 

4.3. 離線 - 實(shí)時(shí)數(shù)倉(cāng)一體化

最后提一個(gè)方向是離線實(shí)時(shí)數(shù)倉(cāng)的一體化。我們認(rèn)為短期內(nèi)，實(shí)時(shí)數(shù)倉(cāng)無法替代離線數(shù)倉(cāng)，兩者并存是新常態(tài)。在離線數(shù)倉(cāng)時(shí)代，我們積累的工具體系，如何去適配實(shí)時(shí)數(shù)倉(cāng)，如何實(shí)現(xiàn)離線與實(shí)時(shí)數(shù)倉(cāng)的一體化管理?理論上來講，它們的數(shù)據(jù)來源是一致的，上層抽象也都是 Table 與 SQL，但本質(zhì)上也有不同的點(diǎn)，比如時(shí)間粒度以及計(jì)算模式。對(duì)于數(shù)據(jù)工具與產(chǎn)品來說，需要做哪些改造來實(shí)現(xiàn)完全的一體化，這也是我們?cè)谔剿骱退伎嫉摹?nbsp;