引言

近些年，企業(yè)對(duì)數(shù)據(jù)服務(wù)實(shí)時(shí)化服務(wù)的需求日益增多。本文整理了常見(jiàn)實(shí)時(shí)數(shù)據(jù)組件的性能特點(diǎn)和適用場(chǎng)景，介紹了美團(tuán)如何通過(guò) Flink 引擎構(gòu)建實(shí)時(shí)數(shù)據(jù)倉(cāng)庫(kù)，從而提供高效、穩(wěn)健的實(shí)時(shí)數(shù)據(jù)服務(wù)。本文主要闡述使用 Flink 在實(shí)際數(shù)據(jù)生產(chǎn)上的經(jīng)驗(yàn)。

實(shí)時(shí)平臺(tái)初期架構(gòu)

在實(shí)時(shí)數(shù)據(jù)系統(tǒng)建設(shè)初期，由于對(duì)實(shí)時(shí)數(shù)據(jù)的需求較少，形成不了完整的數(shù)據(jù)體系。我們采用的是“一路到底”的開(kāi)發(fā)模式：通過(guò)在實(shí)時(shí)計(jì)算平臺(tái)上部署 Storm 作業(yè)處理實(shí)時(shí)數(shù)據(jù)隊(duì)列來(lái)提取數(shù)據(jù)指標(biāo)，直接推送到實(shí)時(shí)應(yīng)用服務(wù)中。

圖1 初期實(shí)時(shí)數(shù)據(jù)架構(gòu)

但是，隨著產(chǎn)品和業(yè)務(wù)人員對(duì)實(shí)時(shí)數(shù)據(jù)需求的不斷增多，新的挑戰(zhàn)也隨之發(fā)生。

1. 數(shù)據(jù)指標(biāo)越來(lái)越多，“煙囪式”的開(kāi)發(fā)導(dǎo)致代碼耦合問(wèn)題嚴(yán)重。
2. 需求越來(lái)越多，有的需要明細(xì)數(shù)據(jù)，有的需要 OLAP 分析。單一的開(kāi)發(fā)模式難以應(yīng)付多種需求。
3. 缺少完善的監(jiān)控系統(tǒng)，無(wú)法在對(duì)業(yè)務(wù)產(chǎn)生影響之前發(fā)現(xiàn)并修復(fù)問(wèn)題。

實(shí)時(shí)數(shù)據(jù)倉(cāng)庫(kù)的構(gòu)建

為解決以上問(wèn)題，我們根據(jù)生產(chǎn)離線數(shù)據(jù)的經(jīng)驗(yàn)，選擇使用分層設(shè)計(jì)方案來(lái)建設(shè)實(shí)時(shí)數(shù)據(jù)倉(cāng)庫(kù)，其分層架構(gòu)如下圖所示：

圖2 實(shí)時(shí)數(shù)倉(cāng)數(shù)據(jù)分層架構(gòu)

該方案由以下四層構(gòu)成：

1. ODS 層：Binlog 和流量日志以及各業(yè)務(wù)實(shí)時(shí)隊(duì)列。
2. 數(shù)據(jù)明細(xì)層：業(yè)務(wù)領(lǐng)域整合提取事實(shí)數(shù)據(jù)，離線全量和實(shí)時(shí)變化數(shù)據(jù)構(gòu)建實(shí)時(shí)維度數(shù)據(jù)。
3. 數(shù)據(jù)匯總層：使用寬表模型對(duì)明細(xì)數(shù)據(jù)補(bǔ)充維度數(shù)據(jù)，對(duì)共性指標(biāo)進(jìn)行匯總。
4. App 層：為了具體需求而構(gòu)建的應(yīng)用層，通過(guò) RPC 框架對(duì)外提供服務(wù)。

通過(guò)多層設(shè)計(jì)我們可以將處理數(shù)據(jù)的流程沉淀在各層完成。比如在數(shù)據(jù)明細(xì)層統(tǒng)一完成數(shù)據(jù)的過(guò)濾、清洗、規(guī)范、脫敏流程;在數(shù)據(jù)匯總層加工共性的多維指標(biāo)匯總數(shù)據(jù)。提高了代碼的復(fù)用率和整體生產(chǎn)效率。同時(shí)各層級(jí)處理的任務(wù)類型相似，可以采用統(tǒng)一的技術(shù)方案優(yōu)化性能，使數(shù)倉(cāng)技術(shù)架構(gòu)更簡(jiǎn)潔。

技術(shù)選型

1.存儲(chǔ)引擎的調(diào)研

實(shí)時(shí)數(shù)倉(cāng)在設(shè)計(jì)中不同于離線數(shù)倉(cāng)在各層級(jí)使用同種儲(chǔ)存方案，比如都存儲(chǔ)在 Hive 、DB 中的策略。首先對(duì)中間過(guò)程的表，采用將結(jié)構(gòu)化的數(shù)據(jù)通過(guò)消息隊(duì)列存儲(chǔ)和高速 KV 存儲(chǔ)混合的方案。實(shí)時(shí)計(jì)算引擎可以通過(guò)監(jiān)聽(tīng)消息消費(fèi)消息隊(duì)列內(nèi)的數(shù)據(jù)，進(jìn)行實(shí)時(shí)計(jì)算。而在高速 KV 存儲(chǔ)上的數(shù)據(jù)則可以用于快速關(guān)聯(lián)計(jì)算，比如維度數(shù)據(jù)。 其次在應(yīng)用層上，針對(duì)數(shù)據(jù)使用特點(diǎn)配置存儲(chǔ)方案直接寫(xiě)入。避免了離線數(shù)倉(cāng)應(yīng)用層同步數(shù)據(jù)流程帶來(lái)的處理延遲。 為了解決不同類型的實(shí)時(shí)數(shù)據(jù)需求，合理的設(shè)計(jì)各層級(jí)存儲(chǔ)方案，我們調(diào)研了美團(tuán)內(nèi)部使用比較廣泛的幾種存儲(chǔ)方案。

表1 存儲(chǔ)方案列表

根據(jù)不同業(yè)務(wù)場(chǎng)景，實(shí)時(shí)數(shù)倉(cāng)各個(gè)模型層次使用的存儲(chǔ)方案大致如下：

圖3 實(shí)時(shí)數(shù)倉(cāng)存儲(chǔ)分層架構(gòu)

1. 數(shù)據(jù)明細(xì)層 對(duì)于維度數(shù)據(jù)部分場(chǎng)景下關(guān)聯(lián)的頻率可達(dá) 10w+ TPS，我們選擇 Cellar(美團(tuán)內(nèi)部存儲(chǔ)系統(tǒng)) 作為存儲(chǔ)，封裝維度服務(wù)為實(shí)時(shí)數(shù)倉(cāng)提供維度數(shù)據(jù)。
2. 數(shù)據(jù)匯總層 對(duì)于通用的匯總指標(biāo)，需要進(jìn)行歷史數(shù)據(jù)關(guān)聯(lián)的數(shù)據(jù)，采用和維度數(shù)據(jù)一樣的方案通過(guò) Cellar 作為存儲(chǔ)，用服務(wù)的方式進(jìn)行關(guān)聯(lián)操作。
3. 數(shù)據(jù)應(yīng)用層 應(yīng)用層設(shè)計(jì)相對(duì)復(fù)雜，再對(duì)比了幾種不同存儲(chǔ)方案后。我們制定了以數(shù)據(jù)讀寫(xiě)頻率 1000 QPS 為分界的判斷依據(jù)。對(duì)于讀寫(xiě)平均頻率高于 1000 QPS 但查詢不太復(fù)雜的實(shí)時(shí)應(yīng)用，比如商戶實(shí)時(shí)的經(jīng)營(yíng)數(shù)據(jù)。采用 Cellar 為存儲(chǔ)，提供實(shí)時(shí)數(shù)據(jù)服務(wù)。對(duì)于一些查詢復(fù)雜的和需要明細(xì)列表的應(yīng)用，使用 Elasticsearch 作為存儲(chǔ)則更為合適。而一些查詢頻率低，比如一些內(nèi)部運(yùn)營(yíng)的數(shù)據(jù)。 Druid 通過(guò)實(shí)時(shí)處理消息構(gòu)建索引，并通過(guò)預(yù)聚合可以快速的提供實(shí)時(shí)數(shù)據(jù) OLAP 分析功能。對(duì)于一些歷史版本的數(shù)據(jù)產(chǎn)品進(jìn)行實(shí)時(shí)化改造時(shí)，也可以使用 MySQL 存儲(chǔ)便于產(chǎn)品迭代。

2.計(jì)算引擎的調(diào)研

在實(shí)時(shí)平臺(tái)建設(shè)初期我們使用 Storm 引擎來(lái)進(jìn)行實(shí)時(shí)數(shù)據(jù)處理。Storm 引擎雖然在靈活性和性能上都表現(xiàn)不錯(cuò)。但是由于 API 過(guò)于底層，在數(shù)據(jù)開(kāi)發(fā)過(guò)程中需要對(duì)一些常用的數(shù)據(jù)操作進(jìn)行功能實(shí)現(xiàn)。比如表關(guān)聯(lián)、聚合等，產(chǎn)生了很多額外的開(kāi)發(fā)工作，不僅引入了很多外部依賴比如緩存，而且實(shí)際使用時(shí)性能也不是很理想。同時(shí) Storm 內(nèi)的數(shù)據(jù)對(duì)象 Tuple 支持的功能也很簡(jiǎn)單，通常需要將其轉(zhuǎn)換為 Java 對(duì)象來(lái)處理。對(duì)于這種基于代碼定義的數(shù)據(jù)模型，通常我們只能通過(guò)文檔來(lái)進(jìn)行維護(hù)。不僅需要額外的維護(hù)工作，同時(shí)在增改字段時(shí)也很麻煩。綜合來(lái)看使用 Storm 引擎構(gòu)建實(shí)時(shí)數(shù)倉(cāng)難度較大。我們需要一個(gè)新的實(shí)時(shí)處理方案，要能夠?qū)崿F(xiàn)：

1. 提供高級(jí) API，支持常見(jiàn)的數(shù)據(jù)操作比如關(guān)聯(lián)聚合，***是能支持 SQL。
2. 具有狀態(tài)管理和自動(dòng)支持久化方案，減少對(duì)存儲(chǔ)的依賴。
3. 便于接入元數(shù)據(jù)服務(wù)，避免通過(guò)代碼管理數(shù)據(jù)結(jié)構(gòu)。
4. 處理性能至少要和 Storm 一致。

我們對(duì)主要的實(shí)時(shí)計(jì)算引擎進(jìn)行了技術(shù)調(diào)研?？偨Y(jié)了各類引擎特性如下表所示：

表2 實(shí)時(shí)計(jì)算方案列表

從調(diào)研結(jié)果來(lái)看，F(xiàn)link 和 Spark Streaming 的 API 、容錯(cuò)機(jī)制與狀態(tài)持久化機(jī)制都可以解決一部分我們目前使用 Storm 中遇到的問(wèn)題。但 Flink 在數(shù)據(jù)延遲上和 Storm 更接近，對(duì)現(xiàn)有應(yīng)用影響最小。而且在公司內(nèi)部的測(cè)試中 Flink 的吞吐性能對(duì)比 Storm 有十倍左右提升。綜合考量我們選定 Flink 引擎作為實(shí)時(shí)數(shù)倉(cāng)的開(kāi)發(fā)引擎。

更加引起我們注意的是，F(xiàn)link 的 Table 抽象和 SQL 支持。雖然使用 Strom 引擎也可以處理結(jié)構(gòu)化數(shù)據(jù)。但畢竟依舊是基于消息的處理 API ，在代碼層層面上不能完全享受操作結(jié)構(gòu)化數(shù)據(jù)的便利。而 Flink 不僅支持了大量常用的 SQL 語(yǔ)句，基本覆蓋了我們的開(kāi)發(fā)場(chǎng)景。而且 Flink 的 Table 可以通過(guò) TableSchema 進(jìn)行管理，支持豐富的數(shù)據(jù)類型和數(shù)據(jù)結(jié)構(gòu)以及數(shù)據(jù)源?？梢院苋菀椎暮同F(xiàn)有的元數(shù)據(jù)管理系統(tǒng)或配置管理系統(tǒng)結(jié)合。通過(guò)下圖我們可以清晰的看出 Storm 和 Flink 在開(kāi)發(fā)統(tǒng)過(guò)程中的區(qū)別。

圖4 Flink - Storm 對(duì)比圖

在使用 Storm 開(kāi)發(fā)時(shí)處理邏輯與實(shí)現(xiàn)需要固化在 Bolt 的代碼。Flink 則可以通過(guò) SQL 進(jìn)行開(kāi)發(fā)，代碼可讀性更高，邏輯的實(shí)現(xiàn)由開(kāi)源框架來(lái)保證可靠高效，對(duì)特定場(chǎng)景的優(yōu)化只要修改 Flink SQL 優(yōu)化器功能實(shí)現(xiàn)即可，而不影響邏輯代碼。使我們可以把更多的精力放到到數(shù)據(jù)開(kāi)發(fā)中，而不是邏輯的實(shí)現(xiàn)。當(dāng)需要離線數(shù)據(jù)和實(shí)時(shí)數(shù)據(jù)口徑統(tǒng)一的場(chǎng)景時(shí)，我們只需對(duì)離線口徑的 SQL 腳本稍加改造即可，極大地提高了開(kāi)發(fā)效率。同時(shí)對(duì)比圖中 Flink 和 Storm 使用的數(shù)據(jù)模型，Storm 需要通過(guò)一個(gè) Java 的 Class 去定義數(shù)據(jù)結(jié)構(gòu)，F(xiàn)link Table 則可以通過(guò)元數(shù)據(jù)來(lái)定義?？梢院芎玫暮蛿?shù)據(jù)開(kāi)發(fā)中的元數(shù)據(jù)，數(shù)據(jù)治理等系統(tǒng)結(jié)合，提高開(kāi)發(fā)效率。

Flink使用心得

在利用 Flink-Table 構(gòu)建實(shí)時(shí)數(shù)據(jù)倉(cāng)庫(kù)過(guò)程中。我們針對(duì)一些構(gòu)建數(shù)據(jù)倉(cāng)庫(kù)的常用操作，比如數(shù)據(jù)指標(biāo)的維度擴(kuò)充，數(shù)據(jù)按主題關(guān)聯(lián)，以及數(shù)據(jù)的聚合運(yùn)算通過(guò) Flink 來(lái)實(shí)現(xiàn)總結(jié)了一些使用心得。

1.維度擴(kuò)充

數(shù)據(jù)指標(biāo)的維度擴(kuò)充，我們采用的是通過(guò)維度服務(wù)獲取維度信息。雖然基于 Cellar 的維度服務(wù)通常的響應(yīng)延遲可以在 1ms 以下。但是為了進(jìn)一步優(yōu)化 Flink 的吞吐，我們對(duì)維度數(shù)據(jù)的關(guān)聯(lián)全部采用了異步接口訪問(wèn)的方式，避免了使用 RPC 調(diào)用影響數(shù)據(jù)吞吐。

對(duì)于一些數(shù)據(jù)量很大的流，比如流量日志數(shù)據(jù)量在 10W 條/秒這個(gè)量級(jí)。在關(guān)聯(lián) UDF 的時(shí)候內(nèi)置了緩存機(jī)制，可以根據(jù)***率和時(shí)間對(duì)緩存進(jìn)行淘汰，配合用關(guān)聯(lián)的 Key 值進(jìn)行分區(qū)，顯著減少了對(duì)外部服務(wù)的請(qǐng)求次數(shù)，有效的減少了處理延遲和對(duì)外部系統(tǒng)的壓力。

2.數(shù)據(jù)關(guān)聯(lián)

數(shù)據(jù)主題合并，本質(zhì)上就是多個(gè)數(shù)據(jù)源的關(guān)聯(lián)，簡(jiǎn)單的來(lái)說(shuō)就是 Join 操作。Flink 的 Table 是建立在***流這個(gè)概念上的。在進(jìn)行 Join 操作時(shí)并不能像離線數(shù)據(jù)一樣對(duì)兩個(gè)完整的表進(jìn)行關(guān)聯(lián)。采用的是在窗口時(shí)間內(nèi)對(duì)數(shù)據(jù)進(jìn)行關(guān)聯(lián)的方案，相當(dāng)于從兩個(gè)數(shù)據(jù)流中各自截取一段時(shí)間的數(shù)據(jù)進(jìn)行 Join 操作。有點(diǎn)類似于離線數(shù)據(jù)通過(guò)限制分區(qū)來(lái)進(jìn)行關(guān)聯(lián)。同時(shí)需要注意 Flink 關(guān)聯(lián)表時(shí)必須有至少一個(gè)“等于”關(guān)聯(lián)條件，因?yàn)榈忍?hào)兩邊的值會(huì)用來(lái)分組。

由于 Flink 會(huì)緩存窗口內(nèi)的全部數(shù)據(jù)來(lái)進(jìn)行關(guān)聯(lián)，緩存的數(shù)據(jù)量和關(guān)聯(lián)的窗口大小成正比。因此 Flink 的關(guān)聯(lián)查詢，更適合處理一些可以通過(guò)業(yè)務(wù)規(guī)則限制關(guān)聯(lián)數(shù)據(jù)時(shí)間范圍的場(chǎng)景。比如關(guān)聯(lián)下單用戶購(gòu)買之前 30 分鐘內(nèi)的瀏覽日志。過(guò)大的窗口不僅會(huì)消耗更多的內(nèi)存，同時(shí)會(huì)產(chǎn)生更大的 Checkpoint ，導(dǎo)致吞吐下降或 Checkpoint 超時(shí)。在實(shí)際生產(chǎn)中可以使用 RocksDB 和啟用增量保存點(diǎn)模式，減少 Checkpoint 過(guò)程對(duì)吞吐產(chǎn)生影響。對(duì)于一些需要關(guān)聯(lián)窗口期很長(zhǎng)的場(chǎng)景，比如關(guān)聯(lián)的數(shù)據(jù)可能是幾天以前的數(shù)據(jù)。對(duì)于這些歷史數(shù)據(jù)，我們可以將其理解為是一種已經(jīng)固定不變的"維度"。可以將需要被關(guān)聯(lián)的歷史數(shù)據(jù)采用和維度數(shù)據(jù)一致的處理方法："緩存 + 離線"數(shù)據(jù)方式存儲(chǔ)，用接口的方式進(jìn)行關(guān)聯(lián)。另外需要注意 Flink 對(duì)多表關(guān)聯(lián)是直接順序鏈接的，因此需要注意先進(jìn)行結(jié)果集小的關(guān)聯(lián)。

3.聚合運(yùn)算

使用聚合運(yùn)算時(shí)，F(xiàn)link 對(duì)常見(jiàn)的聚合運(yùn)算如求和、極值、均值等都有支持。美中不足的是對(duì)于 Distinct 的支持，F(xiàn)link-1.6 之前的采用的方案是通過(guò)先對(duì)去重字段進(jìn)行分組再聚合實(shí)現(xiàn)。對(duì)于需要對(duì)多個(gè)字段去重聚合的場(chǎng)景，只能分別計(jì)算再進(jìn)行關(guān)聯(lián)處理效率很低。為此我們開(kāi)發(fā)了自定義的 UDAF，實(shí)現(xiàn)了 MapView 精確去重、BloomFilter 非精確去重、 HyperLogLog 超低內(nèi)存去重方案應(yīng)對(duì)各種實(shí)時(shí)去重場(chǎng)景。但是在使用自定義的 UDAF 時(shí)，需要注意 RocksDBStateBackend 模式對(duì)于較大的 Key 進(jìn)行更新操作時(shí)序列化和反序列化耗時(shí)很多。可以考慮使用 FsStateBackend 模式替代。另外要注意的一點(diǎn) Flink 框架在計(jì)算比如 Rank 這樣的分析函數(shù)時(shí)，需要緩存每個(gè)分組窗口下的全部數(shù)據(jù)才能進(jìn)行排序，會(huì)消耗大量?jī)?nèi)存。建議在這種場(chǎng)景下優(yōu)先轉(zhuǎn)換為 TopN 的邏輯，看是否可以解決需求。

下圖展示一個(gè)完整的使用 Flink 引擎生產(chǎn)一張實(shí)時(shí)數(shù)據(jù)表的過(guò)程：

圖5 實(shí)時(shí)計(jì)算流程圖

實(shí)時(shí)數(shù)倉(cāng)成果

通過(guò)使用實(shí)時(shí)數(shù)倉(cāng)代替原有流程，我們將數(shù)據(jù)生產(chǎn)中的各個(gè)流程抽象到實(shí)時(shí)數(shù)倉(cāng)的各層當(dāng)中。實(shí)現(xiàn)了全部實(shí)時(shí)數(shù)據(jù)應(yīng)用的數(shù)據(jù)源統(tǒng)一，保證了應(yīng)用數(shù)據(jù)指標(biāo)、維度的口徑的一致。在幾次數(shù)據(jù)口徑發(fā)生修改的場(chǎng)景中，我們通過(guò)對(duì)倉(cāng)庫(kù)明細(xì)和匯總進(jìn)行改造，在完全不用修改應(yīng)用代碼的情況下就完成全部應(yīng)用的口徑切換。在開(kāi)發(fā)過(guò)程中通過(guò)嚴(yán)格的把控?cái)?shù)據(jù)分層、主題域劃分、內(nèi)容組織標(biāo)準(zhǔn)規(guī)范和命名規(guī)則。使數(shù)據(jù)開(kāi)發(fā)的鏈路更為清晰，減少了代碼的耦合。再配合上使用 Flink SQL 進(jìn)行開(kāi)發(fā)，代碼加簡(jiǎn)潔。單個(gè)作業(yè)的代碼量從平均 300+ 行的 JAVA 代碼 ，縮減到幾十行的 SQL 腳本。項(xiàng)目的開(kāi)發(fā)時(shí)長(zhǎng)也大幅減短，一人日開(kāi)發(fā)多個(gè)實(shí)時(shí)數(shù)據(jù)指標(biāo)情況也不少見(jiàn)。

除此以外我們通過(guò)針對(duì)數(shù)倉(cāng)各層級(jí)工作內(nèi)容的不同特點(diǎn)，可以進(jìn)行針對(duì)性的性能優(yōu)化和參數(shù)配置。比如 ODS 層主要進(jìn)行數(shù)據(jù)的解析、過(guò)濾等操作，不需要 RPC 調(diào)用和聚合運(yùn)算。 我們針對(duì)數(shù)據(jù)解析過(guò)程進(jìn)行優(yōu)化，減少不必要的 JSON 字段解析，并使用更高效的 JSON 包。在資源分配上，單個(gè) CPU 只配置 1GB 的內(nèi)存即可滿需求。而匯總層主要?jiǎng)t主要進(jìn)行聚合與關(guān)聯(lián)運(yùn)算，可以通過(guò)優(yōu)化聚合算法、內(nèi)外存共同運(yùn)算來(lái)提高性能、減少成本。資源配置上也會(huì)分配更多的內(nèi)存，避免內(nèi)存溢出。通過(guò)這些優(yōu)化手段，雖然相比原有流程實(shí)時(shí)數(shù)倉(cāng)的生產(chǎn)鏈路更長(zhǎng)，但數(shù)據(jù)延遲并沒(méi)有明顯增加。同時(shí)實(shí)時(shí)數(shù)據(jù)應(yīng)用所使用的計(jì)算資源也有明顯減少。

展望

我們的目標(biāo)是將實(shí)時(shí)倉(cāng)庫(kù)建設(shè)成可以和離線倉(cāng)庫(kù)數(shù)據(jù)準(zhǔn)確性，一致性媲美的數(shù)據(jù)系統(tǒng)。為商家，業(yè)務(wù)人員以及美團(tuán)用戶提供及時(shí)可靠的數(shù)據(jù)服務(wù)。同時(shí)作為到餐實(shí)時(shí)數(shù)據(jù)的統(tǒng)一出口，為集團(tuán)其他業(yè)務(wù)部門(mén)助力。未來(lái)我們將更加關(guān)注在數(shù)據(jù)可靠性和實(shí)時(shí)數(shù)據(jù)指標(biāo)管理。建立完善的數(shù)據(jù)監(jiān)控，數(shù)據(jù)血緣檢測(cè)，交叉檢查機(jī)制。及時(shí)對(duì)異常數(shù)據(jù)或數(shù)據(jù)延遲進(jìn)行監(jiān)控和預(yù)警。同時(shí)優(yōu)化開(kāi)發(fā)流程，降低開(kāi)發(fā)實(shí)時(shí)數(shù)據(jù)學(xué)習(xí)成本。讓更多有實(shí)時(shí)數(shù)據(jù)需求的人，可以自己動(dòng)手解決問(wèn)題。