摘要：實(shí)際上，本身就預(yù)留了與外部元數(shù)據(jù)對接的能力，分別提供了和這兩個(gè)抽象。對接外部數(shù)據(jù)源搞清楚了注冊庫表的過程，給我們帶來這樣一個(gè)思路如果外部元數(shù)據(jù)創(chuàng)建的表也能被轉(zhuǎn)換成可識別的，那么就能被無縫地注冊到。

本文整理自 2019 年 4 月 13 日在深圳舉行的 Flink Meetup 會(huì)議，分享嘉賓張俊，目前擔(dān)任 OPPO 大數(shù)據(jù)平臺(tái)研發(fā)負(fù)責(zé)人，也是 Apache Flink contributor。本文主要內(nèi)容如下：

OPPO 實(shí)時(shí)數(shù)倉的演進(jìn)思路；

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

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

未來工作的思考和展望。

一.OPPO 實(shí)時(shí)數(shù)倉的演進(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)營這些互聯(lián)網(wǎng)應(yīng)用的過程中，OPPO 積累了大量的數(shù)據(jù)，上圖右邊是整體數(shù)據(jù)規(guī)模的演進(jìn)：從 2012 年開始每年都是 2~3 倍的增長速度，截至目前總數(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ù)倉庫，劃分成"原始層 - 明細(xì)層 - 匯總層 - 應(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ū)動(dòng)的數(shù)據(jù)產(chǎn)品與服務(wù)。

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

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

過往 2、3 年，我們的重點(diǎn)聚焦在離線數(shù)倉的構(gòu)建。上圖大致描述了整個(gè)構(gòu)建過程：首先，數(shù)據(jù)來源基本是手機(jī)、日志文件以及 DB 數(shù)據(jù)庫，我們基于 Apache NiFi 打造了高可用、高吞吐的接入系統(tǒng)，將數(shù)據(jù)統(tǒng)一落入 HDFS，形成原始層；緊接著，基于 Hive 的小時(shí)級 ETL 與天級匯總 Hive 任務(wù)，分別負(fù)責(zé)計(jì)算生成明細(xì)層與匯總層；最后，應(yīng)用層是基于 OPPO 內(nèi)部研發(fā)的數(shù)據(jù)產(chǎn)品，主要是報(bào)表分析、用戶畫像以及接口服務(wù)。此外，中間的明細(xì)層還支持基于 Presto 的即席查詢與自助提數(shù)。
伴隨著離線數(shù)倉的逐步完善，業(yè)務(wù)對實(shí)時(shí)數(shù)倉的訴求也愈發(fā)強(qiáng)烈。

1.4. 數(shù)倉實(shí)時(shí)化的訴求

對于數(shù)倉實(shí)時(shí)化的訴求，大家通常都是從業(yè)務(wù)視角來看，但其實(shí)站在平臺(tái)的角度，實(shí)時(shí)化也能帶來切實(shí)的好處。首先，從業(yè)務(wù)側(cè)來看，報(bào)表、標(biāo)簽、接口等都會(huì)有實(shí)時(shí)的應(yīng)用場景，分別參見上圖左邊的幾個(gè)案例；其次，對平臺(tái)側(cè)來說，我們可以從三個(gè)案例來看：第一，OPPO 大量的批量任務(wù)都是從 0 點(diǎn)開始啟動(dòng)，都是通過 T+1 的方式去做數(shù)據(jù)處理，這會(huì)導(dǎo)致計(jì)算負(fù)載集中爆發(fā)，對集群的壓力很大；第二，標(biāo)簽導(dǎo)入也屬于一種 T+1 批量任務(wù)，每次全量導(dǎo)入都會(huì)耗費(fèi)很長的時(shí)間；第三，數(shù)據(jù)質(zhì)量的監(jiān)控也必須是 T+1 的，導(dǎo)致沒辦法及時(shí)發(fā)現(xiàn)數(shù)據(jù)的一些問題。

既然業(yè)務(wù)側(cè)和平臺(tái)側(cè)都有實(shí)時(shí)化的這個(gè)訴求，那 OPPO 是如何來構(gòu)建自己的實(shí)時(shí)數(shù)倉呢？

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

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

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

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

從上圖可以看到，整個(gè) pipeline 與離線數(shù)倉基本相似，只是把 Hive 替換為 Flink，把 HDFS 替換為 Kafka。從總體流程來看，基本模型是不變的，還是由原始層、明細(xì)層、匯總層、應(yīng)用層的級聯(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)勢是四個(gè)：第一，低延遲，高吞吐；第二，端到端的 Exactly-once；第三，可容錯(cuò)的狀態(tài)管理；第四，Window & Event time 的支持?；?Runtime 抽象出 3 個(gè)層次的 API，SQL 處于最上層。

Flink SQL API 有哪些優(yōu)勢呢？我們也從四個(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è)簡單的演示：

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

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

前面提到過，數(shù)倉的抽象是 Table，編程接口是 SQL+UDF。對于用戶來說，平臺(tái)提供的編程界面應(yīng)該是類似上圖的那種，有用過 HUE 做交互查詢的應(yīng)該很熟悉。左邊的菜單是 Table 列表，右邊是 SQL 編輯器，可以在上面直接寫 SQL，然后提交執(zhí)行。要實(shí)現(xiàn)這樣一種交互方式，F(xiàn)link SQL 默認(rèn)是無法實(shí)現(xiàn)的，中間存在 gap，總結(jié)下來就 2 點(diǎn)：第一，元數(shù)據(jù)的管理，怎么去創(chuàng)建庫表，怎么去上傳 UDF，使得之后在 SQL 中可直接引用；第二，SQL 作業(yè)的管理，怎么去編譯 SQL，怎么去提交作業(yè)。
在技術(shù)調(diào)研過程中，我們發(fā)現(xiàn)了 Uber 在 2017 年開源的 AthenaX 框架。

2.3.AthenaX：基于 REST 的 SQL 管理器

AthenaX 可以看作是一個(gè)基于 REST 的 SQL 管理器，它是怎么實(shí)現(xiàn) SQL 作業(yè)與元數(shù)據(jù)管理的呢？

對于 SQL 作業(yè)提交，AthenaX 中有一個(gè) Job 的抽象，封裝了要執(zhí)行的 SQL 以及作業(yè)資源等信息。所有的 Job 由一個(gè) JobStore 來托管，它定期跟 YARN 當(dāng)中處于 Running 狀態(tài)的 App 做一個(gè)匹配。如果不一致，就會(huì)向 YARN 提交對應(yīng)的 Job。

對于元數(shù)據(jù)管理，核心的問題是如何將外部創(chuàng)建的庫表注入 Flink，使得 SQL 中可以識別到。實(shí)際上，F(xiàn)link 本身就預(yù)留了與外部元數(shù)據(jù)對接的能力，分別提供了 ExternalCatalog 和 ExternalCatalogTable 這兩個(gè)抽象。AthenaX 在此基礎(chǔ)上再封裝出一個(gè) TableCatalog，在接口層面做了一定的擴(kuò)展。在提交 SQL 作業(yè)的階段，AthenaX 會(huì)自動(dòng)將 TableCatalog 注冊到 Flink，再調(diào)用 Flink SQL 的接口將 SQL 編譯為 Flink 的可執(zhí)行單元 JobGraph，并最終提交到 YARN 生成新的 App。

AthenaX 雖然定義好了 TableCatalog 接口，但并沒有提供可直接使用的實(shí)現(xiàn)。那么，我們怎么來實(shí)現(xiàn)，以便對接到我們已有的元數(shù)據(jù)系統(tǒng)呢？

2.4.Flink SQL 注冊庫表的過程

首先，我們得搞清楚 Flink SQL 內(nèi)部是如何注冊庫表的。整個(gè)過程涉及到三個(gè)基本的抽象：TableDescriptor、TableFactory 以及 TableEnvironment。

TableDescriptor 顧名思義，是對表的描述，它由三個(gè)子描述符構(gòu)成：第一是 Connector，描述數(shù)據(jù)的來源，比如 Kafka、ES 等；第二是 Format，描述數(shù)據(jù)的格式，比如 csv、json、avro 等；第三是 Schema，描述每個(gè)字段的名稱與類型。TableDescriptor 有兩個(gè)基本的實(shí)現(xiàn)——ConnectTableDescriptor 用于描述內(nèi)部表，也就是編程方式創(chuàng)建的表；ExternalCatalogTable 用于描述外部表。

有了 TableDescriptor，接下來需要 TableFactory 根據(jù)描述信息來實(shí)例化 Table。不同的描述信息需要不同的 TableFactory 來處理，F(xiàn)link 如何找到匹配的 TableFactory 實(shí)現(xiàn)呢？實(shí)際上，為了保證框架的可擴(kuò)展性，F(xiàn)link 采用了 Java SPI 機(jī)制來加載所有聲明過的 TableFactory，通過遍歷的方式去尋找哪個(gè) TableFactory 是匹配該 TableDescriptor 的。TableDescriptor 在傳遞給 TableFactory 前，被轉(zhuǎn)換成一個(gè) map，所有的描述信息都用 key-value 形式來表達(dá)。TableFactory 定義了兩個(gè)用于過濾匹配的方法——一個(gè)是 requiredContext()，用于檢測某些特定 key 的 value 是否匹配，比如 connector.type 是否為 kakfa；另一個(gè)是 supportedProperties()，用于檢測 key 是否能識別，如果出現(xiàn)不識別的 key，說明無法匹配。

匹配到了正確的 TableFactory，接下來就是創(chuàng)建真正的 Table，然后將其通過 TableEnvironment 注冊。最終注冊成功的 Table，才能在 SQL 中引用。

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

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

通過元數(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ù)倉，星型模型是無法避免的。這里有一個(gè)比較簡單的案例：中間的事實(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)換，下面分別展開來講一下。

2.7. 基于 UDF 的維表關(guān)聯(lián)

首先是基于 UDF 的實(shí)現(xiàn)，需要用戶將原始 SQL 改寫為帶 UDF 調(diào)用的 SQL，這里是 userDimFunc，上圖右邊是它的代碼實(shí)現(xiàn)。UserDimFunc 繼承了 Flink SQL 抽象的 TableFunction，它是其中一種 UDF 類型，可以將任意一行數(shù)據(jù)轉(zhuǎn)換成一行或多行數(shù)據(jù)。為了實(shí)現(xiàn)維表關(guān)聯(lián)，在 UDF 初始化時(shí)需要從 MySQL 全量加載維表的數(shù)據(jù)，緩存在內(nèi)存 cache 中。后續(xù)對每行數(shù)據(jù)的處理，TableFunction 會(huì)調(diào)用 eval() 方法，在 eval() 中根據(jù) user_id 去查找 cache，從而實(shí)現(xiàn)關(guān)聯(lián)。當(dāng)然，這里是假定維表數(shù)據(jù)比較小，如果數(shù)據(jù)量很大，不適合全量的加載與緩存，這里不做展開了。

基于 UDF 的實(shí)現(xiàn)，對用戶和平臺(tái)來說都不太友好：用戶需要寫奇怪的 SQL 語句，比如圖中的 LATERAL TABLE；平臺(tái)需要為每個(gè)關(guān)聯(lián)場景定制特定的 UDF，維護(hù)成本太高。有沒有更好的方式呢？下面我們來看看基于 SQL 轉(zhuǎn)換的實(shí)現(xiàn)。

2.8. 基于 SQL 轉(zhuǎn)換的維表關(guān)聯(lián)

我們希望解決基于 UDF 實(shí)現(xiàn)所帶來的問題，用戶不需要改寫原始 SQL，平臺(tái)不需要開發(fā)很多 UDF。有一種思路是，是否可以在 SQL 交給 Flink 編譯之前，加一層 SQL 的解析與改寫，自動(dòng)實(shí)現(xiàn)維表的關(guān)聯(lián)？經(jīng)過一定的技術(shù)調(diào)研與 POC，我們發(fā)現(xiàn)是行得通的，所以稱之為基于 SQL 轉(zhuǎn)換的實(shí)現(xiàn)。下面將該思路展開解釋下。

首先，增加的 SQL 解析是為了識別 SQL 中是否存在預(yù)先定義的維度表，比如上圖中的 user_dim。一旦識別到維表，將觸發(fā) SQL 改寫的流程，將紅框標(biāo)注的 join 語句改寫成新的 Table，這個(gè) Table 怎么得到呢？我們知道，流計(jì)算領(lǐng)域近年來發(fā)展出“流表二象性”的理念，F(xiàn)link 也是該理念的踐行者。這意味著，在 Flink 中 Stream 與 Table 之間是可以相互轉(zhuǎn)換的。我們把 ad_clicks 對應(yīng)的 Table 轉(zhuǎn)換成 Stream，再調(diào)用 flatmap 形成另一個(gè) Stream，最后再轉(zhuǎn)換回 Table，就得到了 ad_clicks_user。最后的問題是，flatmap 是如何實(shí)現(xiàn)維表關(guān)聯(lián)的？

Flink 中對于 Stream 的 flatmap 操作，實(shí)際上是執(zhí)行一個(gè) RichFlatmapFunciton，每來一行數(shù)據(jù)就調(diào)用其 flatmap() 方法做轉(zhuǎn)換。那么，我們可以定制一個(gè) RichFlatmapFunction，來實(shí)現(xiàn)維表數(shù)據(jù)的加載、緩存、查找以及關(guān)聯(lián)，功能與基于 UDF 的 TableFunction 實(shí)現(xiàn)類似。

既然 RichFlatmapFunciton 的實(shí)現(xiàn)邏輯與 TableFunction 相似，那為什么相比基于 UDF 的方式，這種實(shí)現(xiàn)能更加通用呢？核心的點(diǎn)在于多了一層 SQL 解析，可以將維表的信息獲取出來（比如維表名、關(guān)聯(lián)字段、select 字段等），再封裝成 JoinContext 傳遞給 RichFlatmapFunciton，使得的表達(dá)能力就具備通用性了。

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

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

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

這里是一個(gè)典型的實(shí)時(shí) ETL 鏈路，從大表中拆分出各業(yè)務(wù)對應(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)，都要去升級客戶端，去增加新的通道。比如我們有個(gè) sdk_log 通道，所有 APP 應(yīng)用的埋點(diǎn)都往這個(gè)通道上報(bào)數(shù)據(jù)，導(dǎo)致這個(gè)通道對應(yīng)的原始層表巨大，一天幾十個(gè) TB。但實(shí)際上，每個(gè)業(yè)務(wù)只關(guān)心它自身的那部分?jǐn)?shù)據(jù)，這就要求我們在原始層進(jìn)行 ETL 拆分。

這個(gè) SQL 邏輯比較簡單，無非是根據(jù)某些業(yè)務(wù)字段做篩選，插入到不同的業(yè)務(wù)表中去。它的特點(diǎn)是，多行 SQL 最終合并成一個(gè) SQL 提交給 Flink 執(zhí)行。大家擔(dān)心的是，包含了 4 個(gè) SQL，會(huì)不會(huì)對同一份數(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ù)倉的 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)簽 / 接口這些對數(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)該是面向場景的展示表 / 標(biāo)簽表 / 接口表。比如對于展示表，創(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ù)倉的朋友都很清楚它的重要性，它在數(shù)據(jù)治理中都起著不可或缺的關(guān)鍵作用。對于實(shí)時(shí)數(shù)倉來說也莫不如此。我們希望構(gòu)建端到端的血緣關(guān)系，從采集系統(tǒng)的接入通道開始，到中間流經(jīng)的實(shí)時(shí)表與實(shí)時(shí)作業(yè)，再到消費(fèi)數(shù)據(jù)的產(chǎn)品，都能很清晰地展現(xiàn)出來?；谘夑P(guān)系的分析，我們才能評估數(shù)據(jù)的應(yīng)用價(jià)值，核算數(shù)據(jù)的計(jì)算成本。

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

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

更多資訊請?jiān)L問 Apache Flink 中文社區(qū)網(wǎng)站