美團點評基於 Flink 的實時數倉建設實踐

引言

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

實時平臺初期架構

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

圖1 初期實時數據架構

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

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

實時數據倉庫的構建

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

圖2 實時數倉數據分層架構

該方案由如下四層構成：

1. ODS 層：Binlog 和流量日誌以及各業務實時隊列。
2. 數據明細層：業務領域整合提取事實數據，離線全量和實時變化數據構建實時維度數據。
3. 數據彙總層：使用寬表模型對明細數據補充維度數據，對共性指標進行彙總。
4. App 層：爲了具體需求而構建的應用層，經過 RPC 框架對外提供服務。

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

技術選型

1.存儲引擎的調研

實時數倉在設計中不一樣於離線數倉在各層級使用同種儲存方案，好比都存儲在 Hive 、DB 中的策略。首先對中間過程的表，採用將結構化的數據經過消息隊列存儲和高速 KV 存儲混合的方案。實時計算引擎能夠經過監聽消息消費消息隊列內的數據，進行實時計算。而在高速 KV 存儲上的數據則能夠用於快速關聯計算，好比維度數據。 其次在應用層上，針對數據使用特色配置存儲方案直接寫入。避免了離線數倉應用層同步數據流程帶來的處理延遲。 爲了解決不一樣類型的實時數據需求，合理的設計各層級存儲方案，咱們調研了美團內部使用比較普遍的幾種存儲方案。

表1 存儲方案列表

方案	優點	劣勢
MySQL	1. 具備完備的事務功能，能夠對數據進行更新。2. 支持 SQL，開發成本低。	1. 橫向擴展成本大，存儲容易成爲瓶頸； 2. 實時數據的更新和查詢頻率都很高，線上單個實時應用請求就有 1000+ QPS；使用 MySQL 成本過高。
Elasticsearch	1. 吞吐量大，單個機器能夠支持 2500+ QPS，而且集羣能夠快速橫向擴展。2. Term 查詢時響應速度很快，單個機器在 2000+ QPS時，查詢延遲在 20 ms之內。	1. 沒有原生的 SQL 支持，查詢 DSL 有必定的學習門檻；2. 進行聚合運算時性能降低明顯。
Druid	1. 支持超大數據量，經過 Kafka 獲取實時數據時，單個做業可支持 6W+ QPS；2. 能夠在數據導入時經過預計算對數據進行彙總，減小的數據存儲。提升了實際處理數據的效率；3. 有不少開源 OLAP 分析框架。實現如 Superset。	1. 預聚合致使沒法支持明細的查詢；2. 沒法支持 Join 操做；3. Append-only 不支持數據的修改。只能以 Segment 爲單位進行替換。
Cellar	1. 支持超大數據量，採用內存加分佈式存儲的架構，存儲性價比很高；2. 吞吐性能好，經測試處理 3W+ QPS 讀寫請求時，平均延遲在 1ms左右；經過異步讀寫線上最高支持 10W+ QPS。	1. 接口僅支持 KV，Map，List 以及原子加減等；2. 單個 Key 值不得超過 1KB ，而 Value 的值超過 100KB 時則性能降低明顯。

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

圖3 實時數倉存儲分層架構

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

2.計算引擎的調研

在實時平臺建設初期咱們使用 Storm 引擎來進行實時數據處理。Storm 引擎雖然在靈活性和性能上都表現不錯。可是因爲 API 過於底層，在數據開發過程當中須要對一些經常使用的數據操做進行功能實現。好比表關聯、聚合等，產生了不少額外的開發工做，不只引入了不少外部依賴好比緩存，並且實際使用時性能也不是很理想。同時 Storm 內的數據對象 Tuple 支持的功能也很簡單，一般須要將其轉換爲 Java 對象來處理。對於這種基於代碼定義的數據模型，一般咱們只能經過文檔來進行維護。不只須要額外的維護工做，同時在增改字段時也很麻煩。綜合來看使用 Storm 引擎構建實時數倉難度較大。咱們須要一個新的實時處理方案，要可以實現：

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

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

表2 實時計算方案列表

項目/引擎	Storm	Flink	spark-treaming
API	靈活的底層 API 和具備事務保證的 Trident API	流 API 和更加適合數據開發的 Table API 和 Flink SQL 支持	流 API 和 Structured-Streaming API 同時也可使用更適合數據開發的 Spark SQL
容錯機制	ACK 機制	State 分佈式快照保存點	RDD 保存點
狀態管理	Trident State狀態管理	Key State 和 Operator State兩種 State 可使用，支持多種持久化方案	有 UpdateStateByKey 等 API 進行帶狀態的變動，支持多種持久化方案
處理模式	單條流式處理	單條流式處理	Mic batch處理
延遲	毫秒級	毫秒級	秒級
語義保障	At Least Once，Exactly Once	Exactly Once，At Least Once	At Least Once
從調研結果來看，Flink 和 Spark Streaming 的 API 、容錯機制與狀態持久化機制均可以解決一部分咱們目前使用 Storm 中遇到的問題。但 Flink 在數據延遲上和 Storm 更接近，對現有應用影響最小。並且在公司內部的測試中 Flink 的吞吐性能對比 Storm 有十倍左右提高。綜合考量咱們選定 Flink 引擎做爲實時數倉的開發引擎。

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

圖4 Flink - Storm 對比圖

在使用 Storm 開發時處理邏輯與實現須要固化在 Bolt 的代碼。Flink 則能夠經過 SQL 進行開發，代碼可讀性更高，邏輯的實現由開源框架來保證可靠高效，對特定場景的優化只要修改 Flink SQL 優化器功能實現便可，而不影響邏輯代碼。使咱們能夠把更多的精力放到到數據開發中，而不是邏輯的實現。

當須要離線數據和實時數據口徑統一的場景時，咱們只需對離線口徑的 SQL 腳本稍加改造便可，極大地提升了開發效率。同時對比圖中 Flink 和 Storm 使用的數據模型，Storm 須要經過一個 Java 的 Class 去定義數據結構，Flink Table 則能夠經過元數據來定義。能夠很好的和數據開發中的元數據，數據治理等系統結合，提升開發效率。

Flink使用心得

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

1.維度擴充

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

對於一些數據量很大的流，好比流量日誌數據量在 10W 條/秒這個量級。在關聯 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 引擎生產一張實時數據表的過程：

圖5 實時計算流程圖

實時數倉成果

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

除此之外咱們經過針對數倉各層級工做內容的不一樣特色，能夠進行鍼對性的性能優化和參數配置。好比 ODS 層主要進行數據的解析、過濾等操做，不須要 RPC 調用和聚合運算。 咱們針對數據解析過程進行優化，減小沒必要要的 JSON 字段解析，並使用更高效的 JSON 包。在資源分配上，單個 CPU 只配置 1GB 的內存便可滿需求。而彙總層主要則主要進行聚合與關聯運算，能夠經過優化聚合算法、內外存共同運算來提升性能、減小成本。資源配置上也會分配更多的內存，避免內存溢出。經過這些優化手段，雖然相比原有流程實時數倉的生產鏈路更長，但數據延遲並無明顯增長。同時實時數據應用所使用的計算資源也有明顯減小。

展望

咱們的目標是將實時倉庫建設成能夠和離線倉庫數據準確性，一致性媲美的數據系統。爲商家，業務人員以及美團用戶提供及時可靠的數據服務。同時做爲到餐實時數據的統一出口，爲集團其餘業務部門助力。將來咱們將更加關注在數據可靠性和實時數據指標管理。創建完善的數據監控，數據血緣檢測，交叉檢查機制。及時對異常數據或數據延遲進行監控和預警。同時優化開發流程，下降開發實時數據學習成本。讓更多有實時數據需求的人，能夠本身動手解決問題。

參考文獻

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

關於做者

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

招聘信息

對數據工程和將數據經過服務業務釋放價值感興趣的同窗，能夠發送簡歷到 huangweilun@meituan.com。咱們在實時數據倉庫、實時數據治理、實時數據產品開發框架、面向銷售和商家側的數據型創新產品層面，都有不少未知但有意義的領域等你來開拓。