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

OPPO 作為手機(jī)廠商,，基于 Android 定制了自己的 ColorOS 系統(tǒng),，當(dāng)前日活躍用戶超過(guò) 2 億。圍繞 ColorOS,，OPPO 構(gòu)建了很多互聯(lián)網(wǎng)應(yīng)用,，比如應(yīng)用商店、瀏覽器,、信息流等,。在運(yùn)營(yíng)這些互聯(lián)網(wǎng)應(yīng)用的過(guò)程中，OPPO 積累了大量的數(shù)據(jù),，上圖右邊是整體數(shù)據(jù)規(guī)模的演進(jìn)：從 2012 年開(kāi)始每年都是 2~3 倍的增長(zhǎng)速度,，截至目前總數(shù)據(jù)量已經(jīng)超過(guò) 100PB，日增數(shù)據(jù)量超過(guò) 200TB,。

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

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

• 最下層是統(tǒng)一工具體系,，涵蓋了'接入 - 治理 - 開(kāi)發(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ù)用起來(lái),，需要場(chǎng)景驅(qū)動(dòng)的數(shù)據(jù)產(chǎn)品與服務(wù),。

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

過(guò)往 2,、3 年，我們的重點(diǎn)聚焦在離線數(shù)倉(cāng)的構(gòu)建,。上圖大致描述了整個(gè)構(gòu)建過(guò)程：首先,，數(shù)據(jù)來(lái)源基本是手機(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)烈,。

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

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

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

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

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

因此,，這里的核心問(wèn)題是如何基于 Flink 構(gòu)建出這個(gè) pipeline,，下面就介紹下我們基于 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ù),，包括常見(jiàn)的算術(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,；接下來(lái)是注冊(cè) UDF,，篇幅原因這里沒(méi)有列出 UDF 的定義；最后是才是執(zhí)行真正的 SQL,?？梢钥吹?，為了執(zhí)行 SQL，需要做這么多的編碼工作,，這并不是我們希望暴露給用戶的接口,。

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

在技術(shù)調(diào)研過(guò)程中,，我們發(fā)現(xiàn)了 Uber 在 2017 年開(kāi)源的 AthenaX 框架,。

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

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

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

AthenaX 雖然定義好了 TableCatalog 接口,，但并沒(méi)有提供可直接使用的實(shí)現(xiàn)。那么,，我們?cè)趺磥?lái)實(shí)現(xiàn),，以便對(duì)接到我們已有的元數(shù)據(jù)系統(tǒng)呢？

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

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

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

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

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

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

到目前為止,，我們的平臺(tái)已經(jīng)具備了元數(shù)據(jù)管理與 SQL 作業(yè)管理的能力,，但是要真正開(kāi)放給用戶使用，還有一點(diǎn)基本特性存在缺失。通過(guò)我們?nèi)?gòu)建數(shù)倉(cāng),，星型模型是無(wú)法避免的,。這里有一個(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)該怎么來(lái)實(shí)現(xiàn),？我們有兩種實(shí)現(xiàn)方式,，一個(gè)基于 UDF，一個(gè)基于 SQL 轉(zhuǎn)換,，下面分別展開(kāi)來(lái)講一下,。

首先是基于 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ù)對(duì)每行數(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ù)量很大,，不適合全量的加載與緩存,，這里不做展開(kāi)了。

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

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

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

Flink 中對(duì)于 Stream 的 flatmap 操作,，實(shí)際上是執(zhí)行一個(gè) RichFlatmapFunciton,，每來(lái)一行數(shù)據(jù)就調(diào)用其 flatmap() 方法做轉(zhuǎn)換。那么,，我們可以定制一個(gè) RichFlatmapFunction,，來(lái)實(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 解析，可以將維表的信息獲取出來(lái)（比如維表名,、關(guān)聯(lián)字段,、select 字段等），再封裝成 JoinContext 傳遞給 RichFlatmapFunciton,，使得的表達(dá)能力就具備通用性了,。

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

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

OPPO 的最大數(shù)據(jù)來(lái)源是手機(jī)端埋點(diǎn)，從手機(jī) APP 過(guò)來(lái)的數(shù)據(jù)有一個(gè)特點(diǎn),，所有的數(shù)據(jù)是通過(guò)統(tǒng)一的幾個(gè)通道上報(bào)過(guò)來(lái),。因?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)單,，無(wú)非是根據(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。

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

這里是一個(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)于未來(lái)工作,，我們的一些思考與規(guī)劃，還不是太成熟,，拋出來(lái)和大家探討一下,。

什么是端到端？一端是采集到的原始數(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)過(guò) 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 語(yǔ)義是比較成熟的,，一致性上有保證。

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

我們的平臺(tái)能否把端到端給自動(dòng)化起來(lái)，只需要一次 SQL 提交就能打通處理,、導(dǎo)入,、資產(chǎn)化這三步？在這個(gè)思路下,，數(shù)據(jù)開(kāi)發(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)表模板。

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

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

—   The End    —