一.OPPO 實時數倉的演進思路

1.1.OPPO 業務與數據規模

大家都知道 OPPO 是做智能手機的，但并不知道 OPPO 與互聯網以及大數據有什么關系，下圖概要介紹了 OPPO 的業務與數據情況： 

OPPO 作為手機廠商，基于 Android 定制了自己的 ColorOS 系統，當前日活躍用戶超過 2 億。圍繞 ColorOS，OPPO 構建了很多互聯網應用，比如應用商店、瀏覽器、信息流等。

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

要支撐這么大的一個數據量，OPPO 研發出一整套的數據系統與服務，并逐漸形成了自己的數據中臺體系。

1.2.OPPO 數據中臺 

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

最下層是統一工具體系，涵蓋了"接入 - 治理 - 開發 - 消費"全數據鏈路;

基于工具體系之上構建了數據倉庫，劃分成"原始層 - 明細層 - 匯總層 - 應用層"，這也是經典的數倉架構;

再往上是全域的數據體系，什么是全域呢?就是把公司所有的業務數據都打通，形成統一的數據資產，比如 ID-Mapping、用戶標簽等;

最終，數據要能被業務用起來，需要場景驅動的數據產品與服務。

以上就是 OPPO 數據中臺的整個體系，而數據倉庫在其中處于非常基礎與核心的位置。

1.3. 構建 OPPO 離線數倉 

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

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

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

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

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

1.6. 構建 OPPO 實時數倉

 

 

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

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

二. 基于 Flink SQL 的擴展工作

2.1.Why Flink SQL

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

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

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

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

2.2. 基于 WEB 的開發 IDE 

2.5.Flink SQL 對接外部數據源

搞清楚了 Flink SQL 注冊庫表的過程，給我們帶來這樣一個思路：如果外部元數據創建的表也能被轉換成 TableFactory 可識別的 map，那么就能被無縫地注冊到 TableEnvironment。基于這個思路，我們實現了 Flink SQL 與已有元數據中心的對接，大致過程參見下圖： 

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

2.6. 實時表 - 維表關聯

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

假定我們有一個 SQL 分析，需要將點擊流表與用戶維表進行關聯，這個目前在 Flink SQL 中應該怎么來實現?我們有兩種實現方式，一個基于 UDF，一個基于 SQL 轉換。 

三.構建實時數倉的應用案例

下面分享幾個典型的應用案例，都是在我們的平臺上用 Flink SQL 來實現的。

3.1. 實時 ETL 拆分

這里是一個典型的實時 ETL 鏈路，從大表中拆分出各業務對應的小表： 

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

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

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

3.2. 實時指標統計

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

3.3. 實時標簽導入

這里是一個實時標簽導入的案例，手機端實時感知到當前用戶的經緯度，轉換成具體 POI 后導入 ES，最終在標簽系統上做用戶定向。

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

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

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

4.1. 端到端的實時流處理

什么是端到端?一端是采集到的原始數據，另一端是報表 / 標簽 / 接口這些對數據的呈現與應用，連接兩端的是中間實時流。當前我們基于 SQL 的實時流處理，源表是 Kafka，目標表也是 Kafka，統一經過 Kafka 后再導入到 Druid/ES/HBase。

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

然后，上述的端到端其實是由割裂的三個步驟來完成的，每一步可能需要由不同角色人去負責處理：數據處理需要數據開發人員，數據導入需要引擎開發人員，數據資產化需要產品開發人員。 

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

4.2. 實時流的血緣分析

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

4.3. 離線 - 實時數倉一體化

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