引言

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

實時平臺初期架構(gòu)

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

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

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

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

實時數(shù)據(jù)倉庫的構(gòu)建

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

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

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

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

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

技術(shù)選型

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

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

表1 存儲方案列表

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

圖3 實時數(shù)倉存儲分層架構(gòu)

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

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

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

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

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

表2 實時計算方案列表

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

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

圖4 Flink - Storm 對比圖

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

Flink使用心得

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

1.維度擴充

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

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

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

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

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

3.聚合運算

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

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

圖5 實時計算流程圖

實時數(shù)倉成果

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

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

展望

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