應用案例 | 美團點評基於 Flink 的實時數倉建設實踐

本文是美團技術團隊分享的美團點評基於 Flink 的實時數倉建設實踐，Apache Flink 社區公衆號（ Ververica ）受權轉載，文章主要從常見實時數據組件的性能特色和適用場景以及美團經過 Flink 構建實時數據倉庫的過程分享其經驗，供社區參考。

引言

近些年，企業對數據服務實時化服務需求日益增多。本文整理了常見實時數據組件的性能特色和適用場景，介紹了美團如何經過 Flink 引擎構建實時數據倉庫，從而提供高效、穩健的實時數據服務。此前咱們美團技術博客發佈過一篇文章《流計算框架 Flink 與 Storm 的性能對比》，對 Flink 和 Storm 兩個引擎的計算性能進行了比較。本文主要闡述使用 Flink 在實際數據生產上的經驗。

實時平臺初期架構

在實時數據系統建設初期，業務需求也相對較少，尚未造成完整的數據體系。咱們採用的是「一路到底」的開發模式：經過在實時計算平臺上部署 Storm 做業處理實時數據隊列來提取數據指標，直接推送到實時應用服務中。

可是，隨着產品和業務人員對實時數據需求的不斷增多，新的挑戰也隨之發生。

數據指標愈來愈多，「煙囪式」的開發致使代碼耦合問題嚴重。 需求愈來愈多，有的須要明細數據，有的須要 OLAP 分析。單一的開發模式難以應付多種需求。 缺乏完善的監控系統，沒法在對業務產生影響以前發現並修復問題。

實時數據倉庫的構建

爲解決以上問題，咱們根據生產離線數據的經驗，選擇使用分層設計方案來建設實時數據倉庫，其分層架構以下圖所示：

該方案由如下四層構成：

1. ODS 層：Binlog 和流量日誌以及各業務實時隊列。

2. 數據明細層：業務領域整合提取事實數據，離線全量和實時變化數據構建實時維度數據。

3. 數據彙總層：使用寬表模型對明細數據補充維度數據，對共性指標進行彙總。

4. App 層：爲了具體需求而構建的應用層，經過 RPC 框架對外提供服務。

經過多層設計咱們能夠將處理數據的流程沉澱在各層完成。好比在數據明細層統一完成數據的過濾、清洗、規範、脫敏流程；在數據彙總層加工共性的多維指標彙總數據，提升了代碼的複用率和總體生產效率。同時各層級處理的任務類型類似，能夠採用統一的技術方案優化性能，使數倉技術架構更簡潔。

技術選型

1. 存儲引擎的調研

實時數倉在設計中不一樣於離線數倉在各層級使用同種儲存方案，好比都存儲在 Hive 、DB 中的策略。首先對中間過程的表，採用將結構化的數據經過消息隊列存儲和高速 KV 存儲混合的方案。實時計算引擎能夠經過監聽消息消費消息隊列內的數據，進行實時計算；而在高速 KV 存儲上的數據則能夠用於快速關聯計算，好比維度數據。

其次在應用層上，針對數據使用特色配置存儲方案直接寫入，避免了離線數倉應用層同步數據流程帶來的處理延遲。爲了解決不一樣類型的實時數據需求，合理的設計各層級存儲方案，咱們調研了美團內部使用比較普遍的幾種存儲方案。

根據不一樣業務場景，實時數倉各個模型層次使用的存儲方案大體以下：

• 數據明細層：對於維度數據部分場景下關聯的頻率可達 10萬多TPS，咱們選擇 Cellar（美團內部基於Tair開發的KV存儲） 做爲存儲，封裝維度服務爲實時數倉提供維度數據。
• 數據彙總層：對於通用的彙總指標，須要進行歷史數據關聯的數據，採用和維度數據同樣的方案經過 Cellar 做爲存儲，用服務的方式進行關聯操做。
• 數據應用層：應用層設計相對複雜，在對比了幾種不一樣存儲方案後。咱們制定了以數據讀寫頻率 1000 QPS 爲分界的判斷依據。對於讀寫平均頻率高於 1000 QPS 但查詢不太複雜的實時應用，好比商戶實時的經營數據。採用 Cellar 爲存儲，提供實時數據服務。對於一些查詢複雜的和須要明細列表的應用，使用 Elasticsearch 做爲存儲則更爲合適；而一些查詢頻率低，好比一些內部運營的數據， Druid 經過實時處理消息構建索引，並經過預聚合能夠快速的提供實時數據 OLAP 分析功能。對於一些歷史版本的數據產品進行實時化改造時，也可使用 MySQL 存儲便於產品迭代。

2. 計算引擎的調研

在實時平臺建設初期咱們使用 Storm 引擎來進行實時數據處理。Storm 引擎雖然在靈活性和性能上都表現不錯，可是因爲 API 過於底層，在數據開發過程當中須要對一些經常使用的數據操做進行功能實現。好比表關聯、聚合等，產生了不少額外的開發工做，不只引入了不少外部依賴好比緩存，並且實際使用時性能也不是很理想。同時， Storm 內的數據對象 Tuple 支持的功能也很簡單，一般須要將其轉換爲 Java 對象來處理。

對於這種基於代碼定義的數據模型，一般咱們只能經過文檔來進行維護，不只須要額外的維護工做，同時在增改字段時也很麻煩。綜合來看，使用 Storm 引擎構建實時數倉難度較大。咱們須要一個新的實時處理方案，要可以實現：

1. 提供高級 API，支持常見的數據操做好比關聯聚合，最好是能支持 SQL。
2. 具備狀態管理和自動支持久化方案，減小對存儲的依賴。
3. 便於接入元數據服務，避免經過代碼管理數據結構。
4. 處理性能至少要和 Storm 一致。

咱們對主要的實時計算引擎進行了技術調研。總結了各種引擎特性以下表所示：

從調研結果來看，Flink 和 Spark Streaming 的 API 、容錯機制與狀態持久化機制均可以解決一部分，咱們目前使用 Storm 中遇到的問題；但 Flink 在數據延遲上和 Storm 更接近，對現有應用影響最小，並且在公司內部的測試中 Flink 的吞吐性能對比 Storm 有十倍左右提高。綜合考量，咱們選定 Flink 引擎做爲實時數倉的開發引擎。

更加引發咱們注意的是，Flink 的 Table 抽象和 SQL 支持，雖然使用 Strom 引擎也能夠處理結構化數據，但畢竟依舊是基於消息的處理 API ，在代碼層層面上不能徹底享受操做結構化數據的便利。而 Flink 不只支持了大量經常使用的 SQL 語句，基本覆蓋了咱們的開發場景，而且 Flink 的 Table 能夠經過 TableSchema 進行管理，支持豐富的數據類型和數據結構以及數據源，能夠很容易的和現有的元數據管理系統或配置管理系統結合。經過下圖咱們能夠清晰的看出 Storm 和 Flink 在開發統過程當中的區別。

在使用 Storm 開發時處理邏輯與實現須要固化在 Bolt 的代碼。Flink 則能夠經過 SQL 進行開發，代碼可讀性更高，邏輯的實現由開源框架來保證可靠高效，對特定場景的優化只要修改 Flink SQL 優化器功能實現便可，而不影響邏輯代碼，使咱們能夠把更多的精力放到數據開發中，而不是邏輯的實現。當須要離線數據和實時數據口徑統一的場景時，咱們只需對離線口徑的 SQL 腳本稍加改造便可，極大地提升了開發效率。同時，對比圖中 Flink 和 Storm 使用的數據模型，Storm 須要經過一個 Java 的 Class 去定義數據結構，Flink Table 則能夠經過元數據來定義。能夠很好的和數據開發中的元數據，數據治理等系統結合，提升開發效率。

Flink使用心得

在利用 Flink-Table 構建實時數據倉庫過程當中，咱們針對一些構建數據倉庫的經常使用操做，好比數據指標的維度擴充，數據按主題關聯，以及數據的聚合運算經過 Flink 來實現總結了一些使用心得。

1. 維度擴充

數據指標的維度擴充，咱們採用的是經過維度服務獲取維度信息。雖然基於 Cellar 的維度服務一般的響應延遲能夠在 1ms 如下，可是爲了進一步優化 Flink 的吞吐，咱們對維度數據的關聯所有采用了異步接口訪問的方式，避免了使用 RPC 調用影響數據吞吐。

對於一些數據量很大的流，好比流量日誌數據量在 10萬秒/條這個量級，在關聯 UDF 的時候內置了緩存機制，能夠根據命中率和時間對緩存進行淘汰，配合用關聯的 Key 值進行分區，顯著減小了對外部服務的請求次數，有效的減小了處理延遲和對外部系統的壓力。

2. 數據關聯

數據主題合併，本質上就是多個數據源的關聯，簡單的來講就是 Join 操做。Flink 的 Table 是創建在無限流這個概念上的，在進行 Join 操做時並不能像離線數據同樣對兩個完整的表進行關聯，採用的是在窗口時間內對數據進行關聯的方案，至關於從兩個數據流中各自截取一段時間的數據進行 Join 操做，有點相似於離線數據經過限制分區來進行關聯。同時須要注意 Flink 關聯表時必須有至少一個「等於」關聯條件，由於等號兩邊的值會用來分組。

因爲 Flink 會緩存窗口內的所有數據來進行關聯，緩存的數據量和關聯的窗口大小成正比。所以 Flink 的關聯查詢，更適合處理一些能夠經過業務規則限制關聯數據時間範圍的場景，好比關聯下單用戶購買以前 30 分鐘內的瀏覽日誌。過大的窗口不只會消耗更多的內存，同時會產生更大的 Checkpoint ，致使吞吐降低或 Checkpoint 超時。在實際生產中可使用 RocksDB 和啓用增量保存點模式，減小 Checkpoint 過程對吞吐產生影響。

對於一些須要關聯窗口期很長的場景，好比關聯的數據多是幾天之前的數據，對於這些歷史數據，咱們能夠將其理解爲是一種已經固定不變的"維度"。能夠將須要被關聯的歷史數據採用和維度數據一致的處理方法："緩存 + 離線"數據方式存儲，用接口的方式進行關聯。另外，須要注意 Flink 對多表關聯是直接順序連接的，所以須要注意先進行結果集小的關聯。

3. 聚合運算

使用聚合運算時，Flink 對常見的聚合運算如求和、極值、均值等都有支持。美中不足的是對於 Distinct 的支持，Flink-1.6 以前的採用的方案是經過先對去重字段進行分組再聚合實現。對於須要對多個字段去重聚合的場景，只能分別計算再進行關聯處理效率很低，爲此咱們開發了自定義的 UDAF，實現了 MapView 精確去重、BloomFilter 非精確去重、 HyperLogLog 超低內存去重方案應對各類實時去重場景。

可是在使用自定義的 UDAF 時，須要注意 RocksDBStateBackend 模式對於較大的 Key 進行更新操做時序列化和反序列化耗時不少。能夠考慮使用 FsStateBackend 模式替代。另外要注意的一點 Flink 框架在計算好比 Rank 這樣的分析函數時，須要緩存每一個分組窗口下的所有數據才能進行排序，會消耗大量內存，建議在這種場景下優先轉換爲 TopN 的邏輯，看是否能夠解決需求。

下圖展現一個完整的使用 Flink 引擎生產一張實時數據表的過程：

實時數倉成果

經過使用實時數倉代替原有流程，咱們將數據生產中的各個流程抽象到實時數倉的各層當中，實現了所有實時數據應用的數據源統一，保證了應用數據指標、維度的口徑的一致。在幾回數據口徑發生修改的場景中，咱們經過對倉庫明細和彙總進行改造，在徹底不用修改應用代碼的狀況下就完成所有應用的口徑切換。在開發過程當中經過嚴格的把控數據分層、主題域劃分、內容組織標準規範和命名規則，使數據開發的鏈路更爲清晰，減小了代碼的耦合；再配合上使用 Flink SQL 進行開發，代碼更加簡潔，單個做業的代碼量從平均 300+ 行的 Java 代碼 ，縮減到幾十行的 SQL 腳本；項目的開發時長也大幅減短，一人日開發多個實時數據指標狀況也很多見。

除此之外咱們經過針對數倉各層級工做內容的不一樣特色，能夠進行鍼對性的性能優化和參數配置，好比 ODS 層主要進行數據的解析、過濾等操做，不須要 RPC 調用和聚合運算。 咱們針對數據解析過程進行優化，減小沒必要要的 JSON 字段解析，並使用更高效的 JSON 包，在資源分配上，單個 CPU 只配置 1GB 的內存便可滿需求。

而彙總層主要則主要進行聚合與關聯運算，能夠經過優化聚合算法、內外存共同運算來提升性能、減小成本；資源配置上也會分配更多的內存，避免內存溢出。經過這些優化手段，雖然相比原有流程實時數倉的生產鏈路更長，但數據延遲並無明顯增長，同時實時數據應用所使用的計算資源也有明顯減小。

展望

咱們的目標是將實時倉庫建設成能夠和離線倉庫數據準確性，一致性媲美的數據系統，爲商家，業務人員以及美團用戶提供及時可靠的數據服務，同時做爲到餐實時數據的統一出口，爲集團其餘業務部門助力。

將來咱們將更加關注在數據可靠性和實時數據指標管理，創建完善的數據監控，數據血緣檢測，交叉檢查機制。及時對異常數據或數據延遲進行監控和預警；同時，優化開發流程，下降開發實時數據學習成本，讓更多有實時數據需求的人，能夠本身動手解決問題。

參考內容

流計算框架 Flink 與 Storm 的性能對比

關於做者

偉倫， 美團到店餐飲技術部實時數據負責人，2017年加入美團，長期從事數據平臺、實時數據計算、數據架構方面的開發工做。在使用 Flink 進行實時數據生產和提升生產效率上，有一些心得和產出。同時也在積極推廣 Flink 在實時數據處理中的實戰經驗。