一.OPPO 實時數(shù)倉的演進思路

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

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

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

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

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

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

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

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

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

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

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

以上就是 OPPO 數(shù)據(jù)中臺的整個體系，而數(shù)據(jù)倉庫在其中處于非?；A(chǔ)與核心的位置。

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

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

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

1.5. 離線到實時的平滑遷移 

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

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

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

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

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

二. 基于 Flink SQL 的擴展工作

2.1.Why Flink SQL

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

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

那么，基于 Flink SQL API 如何編程呢?下面是一個簡單的演示： 

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

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

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

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

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

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

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

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

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

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

3.1. 實時 ETL 拆分

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

OPPO 的最大數(shù)據(jù)來源是手機端埋點，從手機 APP 過來的數(shù)據(jù)有一個特點，所有的數(shù)據(jù)是通過統(tǒng)一的幾個通道上報過來。因為不可能每一次業(yè)務(wù)有新的埋點，都要去升級客戶端，去增加新的通道。比如我們有個 sdk_log 通道，所有 APP 應(yīng)用的埋點都往這個通道上報數(shù)據(jù)，導(dǎo)致這個通道對應(yīng)的原始層表巨大，一天幾十個 TB。但實際上，每個業(yè)務(wù)只關(guān)心它自身的那部分?jǐn)?shù)據(jù)，這就要求我們在原始層進行 ETL 拆分。

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

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

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

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

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

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

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

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

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

4.1. 端到端的實時流處理

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

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

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

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

4.2. 實時流的血緣分析

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

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

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