1.前置知識

ODPS（Open Data Platform and Service）是阿里云自研的一體化大數據計算平臺和數據倉庫產品，在集團內部離線作為離線數據處理和存儲的產品。離線計算任務節點叫做Odps節點，存儲的離線表叫做Odps表；

Flink: 實時計算引擎，本文代碼開發和測試均基于集團內部實時計算平臺，代碼細節可能會和Flink 官方社區文檔有些許不同，假如用于生產環境測試，參考Apache Flink 官方文檔為準，但是技術方案是通用的哈；

https://flink.apache.org/posts/

2.項目背景

現有業務需求是 “根據用戶注冊以來的累計跑步里程，給用戶發放勛章”，需要實時的計算出用戶【歷史～此時刻】的累計跑步數據。

比如說，某個用戶20210101首次上傳跑步記錄，之后又多次上傳跑步記錄，我們需要實時的計算出，在20210101～當前時刻 期間，該用戶累計跑了多少公里，累計跑了多少次等指標。上述指標的計算涉及用戶歷史至今的所有數據(2018~至今該用戶所有數據)，考慮使用批流結合的方式進行統計。參考批流結合的常用 lambda 方案：

圖片

我們將其拆分到“實時+離線”兩條鏈路分別計算，離線鏈路計算用戶歷史至昨日的累計數據data1，實時鏈路計算當日實時累計數據data2。然后在對兩條鏈路的數據進行匯總，data1+data2即為用戶歷史至今日此時刻的累計數據。

圖片

這里，離線鏈路使用odps來做，實時計算使用Flink來做，數據存儲涉及 hbase、odps，所用消息中間件是MQ。

3.解決方案

3.1 方案描述 

離線鏈路設計

離線鏈路計算目的：為了計算出全量用戶【歷史至昨日】的累計數據。

任務初始化時，先將歷史的存量數據全量計算一次，得到存量累計值；以后每日計算用戶昨日的新增數據，即新增累計值 ；兩者相加即為用戶歷史至昨日的累計數據；循環往復，即可每日更新歷史累計數據。

對應的數據鏈路應該長這樣：

圖片

離線鏈路計算流程如下：

step1：用戶歷史數據初始化。假設該計算任務發布的時間為20231010，首先要對用戶 歷史～20231009 期間的歷史數據進行匯總，得到一個 歷史存量累計數據 history_data；

step2：從20231010起，對用戶每日的增量跑步數據進行匯總，得到該日的增量累計數據 day_data；

step3：將每日的增量累計數據day_data 與 歷史存量累計數據history_data 進行求和，作為新的歷史存量累計數據 history_data(T-1) = day_data(T-1) + history_data(T-2) ；

step4：重復 step2 和step3 ，每日更新歷史存量累計數據 history_data 。

該方案的優點是，歷史全量數據只用計算一次，每日只需計算增量部分后再與存量合并即可，節省計算資源。

實時鏈路設計

實時鏈路計算目的：實時計算出用戶【當日零點至此刻】的累計數據

實時鏈路的計算邏輯比較簡單，對應的計算鏈路示意圖如下：

圖片

實時鏈路計算流程如下：

step1：用戶新增的跑步記錄通過MQ發送給Flink任務；

step2：Flink節點1對數據去重；

step3：Flink節點2對實時匯總統計 當日零點至此刻 用戶的跑步累計數據；step4：將計算結果輸出給下游。

實時離線鏈路融合

實時離線鏈路融合目的：實時得到用戶歷史至此時刻的匯總數據

從上述的離線、實時鏈路中，我們分別得到了用戶【歷史～昨日】累計數據，和【當日凌晨～此刻】累計數據，只需將兩者相加即可實時得到用戶【歷史～此刻】的累計數據：

1. ODPS 計算出用戶 [非當日的歷史累計數據]，為使用方便，會每天更新全量用戶歷史累計數據；
2. 使用Flink節點1 實時計算用戶當日上傳的跑步累計數據；
3. 使用 Flink節點2 實時的將離線數據和實時數據匯總起來；
4. 將匯總結果寫入Hbase結果表，同時發送個MQ消息給下游業務方。

圖片

這里需要有兩點需要注意：

1、根據業務特點，這里將離線計算結果作為維表使用：

Flink任務的下游業務方更關注當日上傳過跑步記錄的用戶的數據更新情況，ODPS結果表作為維表用，Flink任務只對當日上傳跑步記錄的用戶進行查詢，得到“非當日歷史統計數據”，在與“當日新增跑步數據”相加，即可得到該歷史至今的最終的統計數據（更新hbase結果表），符合需求；

我們的跑步用戶中大部分的用戶不會每天都上傳跑步記錄，這些人的結果數據不會發生改變。若將ODPS表作為源表，則依舊會為這些用戶更新數據，浪費計算資源。

【優化】odps表作為維表，不適合大數據量的情況，大數據量使用hbase表作為維表比較合適。這里將odps表數據同步到hbase表中，再拿該hbase表作為維表。

2、初始化下游結果表：在整個任務跑起來前，需要先使用ODPS表的bizdate分區數據初始化hbase結果表，然后再由實時任務對結果表進行更新；

最終的方案示意圖如下：

圖片

3.2 存在的問題

上面的lambda方案有個問題，每日凌晨零點過后，實時任務已開始計算新的一天數據，而離線任務計算尚未結束，這時會出現一個離線數據缺失的窗口期。重點分析一下框圖中“實時數據+離線數據”的部分：

圖片

正常情況

當一個用戶在T日實時上傳了自己的跑步記錄，Flink節點1會計算出其 [當日0點起至此刻] 的跑步累計數據data1，Flink節點2會根據該用戶id取hbase維表里查詢其 [歷史～T-1日] 的累計數據 data2 (hbase表里數據由odps每日更新，即T-1日的存量累計匯總數據)，將data1和data2二者匯總，就可得到 用戶歷史至此時刻的匯總數據；

異常情況 

在凌晨(比如說，在00:00～00:30），ODPS正在計算最新分區數據(T-1日的數據)的期間，新的分區還沒生成完，或者ODPS計算已經完成，但odps表同步base表同步任務還未完成，此時若發生了查詢，會發生什么？

會使用老分區的數據(T-2日的數據，而不是期望的T-1日數據)，導致數據不準。

【問題描述】

在凌晨時分，ODPS計算T-1日數據期間，如果發生了對T-1日的數據查詢，則無法獲取到期望的T-1日數據，會繼續使用T-2日的數據

這里“無法獲取正確數據”的時間長度 = ODPS計算時間 + ODPS同步數據到Hbase的時間

【原因】

Flink查詢維表時 使用維表當前的數據快照，本次查詢完成后再發生的維表更新不會對已有查詢造成影響。

【舉例】

case1（ODPS計算未完成）：

27號，Flink任務計算27號當天的用戶累計數據，同時查詢odps維表的 26號分區 中該用戶的歷史累計數據，兩者相加，得到27號的實時累計結果；

28號凌晨，ODPS正在計算27號分區的數據，任務還未結束，27號分區數據尚不可用；而Flink任務已經開始計算28號當天的用戶累計數據，此刻發生了一次維表查詢，期望從維表中查到該用戶27號統計的歷史累計數據，然而由于27號數據未準備好，則維表會返回26號的歷史累計數據，這會導致數據計算錯誤，相當于丟失了該用戶27號的數據。

case2（ODPS計算完成，但odps表同步habse表任務未完成）：

28號凌晨，ODPS的計算已完成，odps表正在同步數據到hbase表期間，如果Flink發生了查詢，期望獲取用戶27號的最新數據，但由于還沒有更新完成，還是會用26號的數據，會造成類似的錯誤結果。

本文轉載自微信公眾號「滌生大數據」，作者「滌生-莫哥」，可以通過以下二維碼關注。

轉載本文請聯系「滌生大數據」公眾號。