實用好文：知乎實時數倉架構實踐及演進

ETL ApacheKafka · 發表 2019-03-01 11:14:34

摘要：從智慧商業的角度來講，資料的結果代表了使用者的反饋，獲取結果的及時性就顯得尤為重要，快速的獲取資料反饋能夠幫助公司更快的做出決策，更好的進行產品迭代，實時數倉在這一過程中起到了不可替代的作用。本文主要講述知乎的實時數倉實踐以及架構的演進，這包括以下幾個方面：實時數倉 1....

從智慧商業的角度來講，資料的結果代表了使用者的反饋，獲取結果的及時性就顯得尤為重要，快速的獲取資料反饋能夠幫助公司更快的做出決策，更好的進行產品迭代，實時數倉在這一過程中起到了不可替代的作用。

本文主要講述知乎的實時數倉實踐以及架構的演進，這包括以下幾個方面：

實時數倉 1.0 版本，主題： ETL 邏輯實時化，技術方案：Spark Streaming。
實時數倉 2.0 版本，主題：資料分層，指標計算實時化，技術方案：Flink Streaming。
實時數倉未來展望：Streaming SQL 平臺化，元資訊管理系統化，結果驗收自動化。

一、實時數倉 1.0 版本

1.0 版本的實時數倉主要是對流量資料做實時 ETL，並不計算實時指標，也未建立起實時數倉體系，實時場景比較單一，對實時資料流的處理主要是為了提升資料平臺的服務能力。

實時資料的處理向上依賴資料的收集，向下關係到資料的查詢和視覺化，下圖是實時數倉 1.0 版本的整體資料架構圖：

第一部分是資料採集，由三端 SDK 採集資料並通過 Log Collector Server 傳送到 Kafka。
第二部分是資料 ETL，主要完成對原始資料的清洗和加工並分實時和離線匯入 Druid。
第三部分是資料視覺化，由 Druid 負責計算指標並通過 Web Server 配合前端完成資料視覺化。

其中第一、三部分的相關內容請分別參考：

知乎客戶端埋點流程、模型和平臺技術：

https://zhuanlan.zhihu.com/p/40985361
不管你使用哪種OLAP引擎，都可以這樣搞定查詢優化

此處我們詳細介紹第二部分。由於實時資料流的穩定性不如離線資料流，當實時流出現問題後需要離線資料重刷歷史資料，因此實時處理部分我們採用了 Lambda 架構。

Lambda 架構有高容錯、低延時和可擴充套件的特點，為了實現這一設計，我們將 ETL 工作分為兩部分：Streaming ETL 和 Batch ETL。

1、Streaming ETL

這一部分我會介紹實時計算框架的選擇、資料正確性的保證、以及 Streaming 中一些通用的 ETL 邏輯，最後還會介紹 Spark Streaming 在實時 ETL 中的穩定性實踐。

計算框架選擇

在 2016 年年初，業界用的比較多的實時計算框架有 Storm 和 Spark Streaming。Storm 是純流式框架，Spark Streaming 用 Micro Batch 模擬流式計算，前者比後者更實時，後者比前者吞吐量大且生態系統更完善，考慮到知乎的日誌量以及初期對實時性的要求，我們選擇了 Spark Streaming 作為實時資料的處理框架。

資料正確性保證

Spark Streaming 的端到端 Exactly-once 需要下游支援冪等、上游支援流量重放，這裡我們在 Spark Streaming 這一層做到了 At-least-once，正常情況下資料不重不少，但在程式重啟時可能會重發部分資料，為了實現全域性的 Exactly-once，我們在下游做了去重邏輯，關於如何去重後面我會講到。

通用 ETL 邏輯

ETL 邏輯和埋點的資料結構息息相關，我們所有的埋點共用同一套 Proto Buffer Schema，大致如下所示：

message LogEntry {
  optional BaseInfo base = 1;
  optional DetailInfo detail = 2;
  optional ExtraInfo extra = 3;
}

BaseInfo：日誌中最基本的資訊，包括使用者資訊、客戶端資訊、時間資訊、網路資訊等日誌傳送時的必要資訊。
DetailInfo：日誌中的檢視資訊，包括當前檢視、上一個檢視等用於定位使用者所在位置的資訊。
ExtraInfo：日誌中與特定業務相關的額外資訊。

針對上述三種資訊我們將 ETL 邏輯分為通用和非通用兩類，通用邏輯和各個業務相關，主要應用於 Base 和 Detail 資訊，非通用邏輯則是由需求方針對某次需求提出，主要應用於 Extra 資訊。

這裡我們列舉 3 個通用邏輯進行介紹，這包括：動態配置 Streaming、UTM 引數解析、新老使用者識別。

1）動態配置 Streaming

由於 Streaming 任務需要 7 * 24 小時執行，但有些業務邏輯，比如：存在一個元資料資訊中心，當這個元資料發生變化時，需要將這種變化對映到資料流上方便下游使用資料，這種變化可能需要停止 Streaming 任務以更新業務邏輯，但元資料變化的頻率非常高，且在元資料變化後如何及時通知程式的維護者也很難。

動態配置 Streaming 為我們提供了一個解決方案，該方案如下圖所示：

我們可以把經常變化的元資料作為 Streaming Broadcast 變數，該變數扮演的角色類似於只讀快取，同時針對該變數可設定 TTL，快取過期後 Executor 節點會重新向 Driver 請求最新的變數。

通過這種機制可以非常自然的將元資料的變化對映到資料流上，無需重啟任務也無需通知程式的維護者。

2）UTM 引數解析

UTM 的全稱是 Urchin Tracking Module，是用於追蹤網站流量來源的利器，關於 UTM 背景知識介紹可以參考網上其他內容，這裡不再贅述。

下圖是我們解析 UTM 資訊的完整邏輯：

流量資料通過 UTM 引數解析後，我們可以很容易滿足以下需求：

檢視各搜尋引擎導流情況以及這些流量來自於哪些熱門搜尋詞。
市場部某次活動帶來的流量大小，如：頁面瀏覽數、獨立訪問使用者數等。
從站內分享出去的連結在各分享平臺（如：微信、微博）被瀏覽的情況。

3）新老使用者識別

對於網際網路公司而言，增長是一個永恆的話題，實時拿到新增使用者量，對於增長運營十分重要。例如：一次投放 n 個渠道，如果能拿到每個渠道的實時新增使用者數，就可以快速判斷出那些渠道更有價值。

我們用下圖來表達 Streaming ETL 中是如何識別新老使用者的：

判斷一個使用者是不是新使用者，最簡單的辦法就是維護一個歷史使用者池，對每條日誌判斷該使用者是否存在於使用者池中。

由於日誌量巨大，為了不影響 Streaming 任務的處理速度，我們設計了兩層快取：Thread Local Cache 和 Redis Cache，同時用 HBase 做持久化儲存以儲存歷史使用者。

訪問速度：本地記憶體 > 遠端記憶體 > 遠端磁碟，對於我們這個任務來說，只有 1% 左右的請求會打到 HBase，日誌高峰期 26w/s，完全不會影響任務的實時性。

當然本地快取 LruCache 的容量大小和 Redis 的效能也是影響實時性的兩個因素。

Streaming ETL 除了上述幾個通用場景外，還有一些其他邏輯，這些邏輯的存在有的是為了滿足下游更方便的使用資料的需求，有的是對某些錯誤埋點的修復，總之 Streaming ETL 在整個實時數倉中處於指標計算的上游，有著不可替代的作用。

2、Spark Streaming在實時數倉1.0中的穩定性實踐

Spark Streaming 消費 Kafka 資料推薦使用 Direct 模式

我們早期使用的是 High Level 或者叫 Receiver 模式並使用了 checkpoint 功能，這種方式在更新程式邏輯時需要刪除 checkpoint 否則新的程式邏輯就無法生效。

另外，由於使用了 checkpoint 功能，Streaming 任務會保持和 HDFS 通訊，可能會因為 NameNode 的抖動導致 Streaming 任務抖動。因此，推薦使用 Direct 模式，關於這種模式和 Receiver 模式的詳細對比，可以參考官方文件。

保證 Spark Streaming 任務的資源穩定

以 Yarn 為例，執行 Streaming 任務的佇列能夠分配到的最小資源小於了任務所需要的資源，任務會出現頻繁丟失 Executor 的情況，這會導致 Streaming 任務變慢，因為丟失的 Executor 所對應的資料需要重新計算，同時還需要重新分配 Executor。

Spark Streaming 消費 Kafka 時需要做資料流限速

預設情況下 Spark Streaming 以儘可能大的速度讀取訊息佇列，當 Streaming 任務掛了很久之後再次被啟動時，由於拉取的資料量過大可能會導致上游的 Kafka 叢集 IO 被打爆進而出現 Kafka 叢集長時間阻塞。可以使用 Streaming Conf 引數做限速，限定每秒拉取的最大速度。

Spark Streaming 任務失敗後需要自動拉起

長時間執行發現，Spark Streaming 並不能 7 * 24h 穩定執行，我們用 Supervisor 管理 Driver 程序，當任務掛掉後 Driver 程序將不復存在，此時 Supervisor 將重新拉起 Streaming 任務。

3、Batch ETL

接下來要介紹的是 Lambda 架構的第二個部分：Batch ETL，此部分我們需要解決資料落地、離線 ETL、資料批量匯入 Druid 等問題。

針對資料落地我們自研了 map reduce 任務 Batch Loader。
針對資料修復我們自研了離線任務 Repair ETL，離線修復邏輯和實時邏輯共用一套 ETL Lib。
針對批量匯入 ProtoParquet 資料到 Druid，我們擴充套件了 Druid 的匯入外掛。

Batch Load 2 HDFS

前面已經介紹過，我們所有的埋點共用同一套 Proto Buffer Schema，資料傳輸格式全部為二進位制。我們自研了落地 Kafka PB 資料到 Hdfs 的 Map Reduce 任務 BatchLoader，該任務除了落地資料外，還負責對資料去重。

在 Streaming ETL 階段我們做到了 At-least-once，通過此處的BatchLoader 去重我們實現了全域性 Exactly-once。 BatchLoader 除了支援落地資料、對資料去重外，還支援多目錄分割槽（p_date/p_hour/p_plaform/p_logtype）、資料回放、自依賴管理（早期沒有統一的排程器）等。

截止到目前，BatchLoader 落地了 40+ 的 Kakfa Topic 資料。

Repair ETL

資料架構圖中有兩個 Kafka，第一個 Kafka 存放的是原始日誌，第二個 Kafka 存放的是實時 ETL 後的日誌，我們將兩個 Kafka 的資料全部落地，這樣做的目的是為了保證資料鏈路的穩定性。

因為實時 ETL 中有大量的業務邏輯，未知需求的邏輯也許會給整個流量資料帶來安全隱患，而上游的 Log Collect Server 不存在任何業務邏輯只負責收發日誌，相比之下第一個 Kafka 的資料要安全和穩定的多。

Repair ETL 並不是經常啟用，只有當實時 ETL 丟失資料或者出現邏輯錯誤時，才會啟用該程式用於修復日誌。

Batch Load 2 Druid

採用 Tranquility 實時匯入 Druid，這種方式強制需要一個時間視窗，當上遊資料延遲超過窗值後會丟棄視窗之外的資料，這種情況會導致實時報表出現指標錯誤。

為了修復這種錯誤，我們通過 Druid 發起一個離線 Map Reduce 任務定期重導上一個時間段的資料。通過這裡的 Batch 匯入和前面的實時匯入，實現了實時數倉的 Lambda 架構。

4、實時數倉 1.0 的幾個不足之處

到目前為止我們已經介紹完 Lambda 架構實時數倉的幾個模組，1.0 版本的實時數倉有以下幾個不足

所有的流量資料存放在同一個 Kafka Topic 中，如果下游每個業務線都要消費，這會導致全量資料被消費多次，Kafka 出流量太高無法滿足該需求。
所有的指標計算全部由 Druid 承擔，Druid 同時兼顧實時資料來源和離線資料來源的查詢，隨著資料量的暴漲 Druid 穩定性急劇下降，這導致各個業務的核心報表不能穩定產出。
由於每個業務使用同一個流量資料來源配置報表，導致查詢效率低下，同時無法對業務做資料隔離和成本計算。

二、實時數倉 2.0 版本

隨著資料量的暴漲，Druid 中的流量資料來源經常查詢超時，同時各業務消費實時資料的需求也開始增多，如果繼續沿用實時數倉 1.0 架構，需要付出大量的額外成本。

於是，在實時數倉 1.0 的基礎上，我們建立起了實時數倉 2.0，梳理出了新的架構設計並開始著手建立實時數倉體系，新的架構如下圖所示：

1、原始層

實時數倉 1.0 我們只對流量資料做 ETL 處理，在 2.0 版本中我們加入了對業務庫的變更日誌 Binlog 的處理，Binlog 日誌在原始層為庫級別或者 Mysql 例項級別，即：一個庫或者例項的變更日誌存放在同一個 Kafka Topic 中。

同時隨著公司業務的發展不斷有新 App 產生，在原始層不僅採集「知乎」日誌，像知乎極速版以及內部孵化專案的埋點資料也需要採集，不同 App 的埋點資料仍然使用同一套 PB Schema。

2、明細層

明細層是我們的 ETL 層，這一層資料是由原始層經過 Streaming ETL 後得到。

其中對 Binlog 日誌的處理主要是完成庫或者例項日誌到表日誌的拆分，對流量日誌主要是做一些通用 ETL 處理，由於我們使用的是同一套 PB 結構，對不同 App 資料處理的邏輯程式碼可以完全複用，這大大降低了我們的開發成本。

3、彙總層

彙總層之明細彙總

明細彙總層是由明細層通過 ETL 得到，主要以寬表形式存在。

業務明細彙總是由業務事實明細表和維度表 Join 得到，流量明細彙總是由流量日誌按業務線拆分和流量維度 Join 得到。

流量按業務拆分後可以滿足各業務實時消費的需求，我們在流量拆分這一塊做到了自動化，下圖演示了流量資料自動切分的過程：

Streaming Proxy 是流量分發模組，它消費上游 ETL 後的全量資料並定期讀取埋點元資訊，通過將流量資料與元資訊資料進行「Join」完成按業務進行流量拆分的邏輯，同時也會對切分後的流量按業務做 ETL 處理。

只要埋點元資訊中新增一個埋點，那麼這個埋點對應的資料就會自動切分到該業務的 Kafka 中，最終業務 Kafka 中的資料是獨屬於當前業務的且已經被通用 ETL 和業務 ETL 處理過，這大大降低了各個業務使用資料的成本。

彙總層之指標彙總

指標彙總層是由明細層或者明細彙總層通過聚合計算得到，這一層產出了絕大部分的實時數倉指標，這也是與實時數倉 1.0 最大的區別。

知乎是一個生產內容的平臺，對業務指標的彙總我們可以從內容角度和使用者角度進行彙總：

從內容角度我們可以實時統計內容（內容可以是答案、問題、文章、視訊、想法）的被點贊數、被關注數、被收藏數等指標，
從使用者角度我可以實時統計使用者的粉絲數、回答數、提問數等指標。

對流量指標的彙總我們分為各業務指標彙總和全域性指標彙總：

對各業務指標彙總，我們可以實時統計首頁、搜尋、視訊、想法等業務的卡片曝光數、卡片點選數、CTR 等。
對全域性指標彙總我們主要以實時會話為主，實時統計一個會話內的 PV 數、卡片曝光數、點選數、瀏覽深度、會話時長等指標。

指標彙總層的儲存選型

不同於明細層和明細彙總層，指標彙總層需要將實時計算好的指標儲存起來以供應用層使用。我們根據不同的場景選用了 HBase 和 Redis 作為實時指標的儲存引擎。

HBase 的場景主要是滿足高頻 Append 操作、低頻隨機讀取且指標列較多的需求，例如：每分鐘統計一次所有內容的被點贊數、被關注數、被收藏數等指標，將每分鐘聚合後的結果行 Append 到 HBase 並不會帶來效能和儲存量的問題，但這種情況下 Redis 在儲存量上可能會出現瓶頸。

Redis 的場景主要是滿足帶 Update 操作且 OPS 較高的需求，例如：實時統計全站所有內容（問題、答案、文章等）的累計 PV 數，由於瀏覽內容產生大量的 PV 日誌，可能高達幾萬或者幾十萬每秒，需要對每一條內容的 PV 進行實時累加，這種場景下選用 Redis 更為合適。

指標計算打通指標系統和視覺化系統

指標口徑管理依賴指標系統，指標視覺化依賴視覺化系統，我們通過下圖的需求開發過程來講解如何將三者聯絡起來：

1）需求方整理好需求文件後向數倉工程師提出需求並約會議評審需求，需求文件中必須包含指標的計算口徑和指標對應的維度。

2）數倉工程師根據需求文件對需求進行評審，評審不通過則返回需求方進一步整理需求並重新提需。

3）在需求評審通過後，數倉工程師開始排期開發：

首先在視覺化系統中建立一個數據源，這個資料來源是後期配置實時報表的資料來源，建立資料來源也即在 HBase 中建立一張 HBase 表。
針對該資料來源建立指標列，建立指標列也即在 HBase 列族中建立列，建立指標列的同時會將該指標資訊錄入指標管理系統。
針對該資料來源繫結維表，這個維表是後期配置多維報表時選用維度值要用的，如果要繫結的維表已經存在，則直接繫結，否則需要匯入維表。
一個完整的資料來源建立後，數倉工程師才能開發實時應用程式，通過應用程式將多維指標實時寫入已建立的資料來源中。

4）需求方根據已建立的資料來源直接配置實時報表。

4、應用層

應用層主要是使用匯總層資料以滿足業務需求。應用層主要分三塊：

通過直接讀取指標彙總資料做實時視覺化，滿足固化的實時報表需求，這部分由實時大盤服務承擔；
推薦演算法等業務直接消費明細彙總資料做實時推薦；
通過 Tranquility 程式實時攝入明細彙總資料到 Druid，滿足實時多維即席分析需求。

5、實時數倉 2.0 中的技術實現

相比實時數倉 1.0 以 Spark Streaming 作為主要實現技術，在實時數倉 2.0 中，我們將 Flink 作為指標彙總層的主要計算框架。

Flink 相比 Spark Streaming 有更明顯的優勢，主要體現在：低延遲、Exactly-once 語義支援、Streaming SQL 支援、狀態管理、豐富的時間型別和視窗計算、CEP 支援等。

我們在實時數倉 2.0 中主要以 Flink 的 Streaming SQL 作為實現方案。使用 Streaming SQL 有以下優點：易於平臺化、開發效率高、維度成本低等。

目前 Streaming SQL 使用起來也有一些缺陷：語法和 Hive SQL 有一定區別，初使用時需要適應；UDF 不如 Hive 豐富，寫 UDF 的頻率高於 Hive。

6、實時數倉 2.0 取得的進展

在明細彙總層通過流量切分滿足了各個業務實時消費日誌的需求。目前完成流量切分的業務達到 14+，由於各業務消費的是切分後的流量，Kafka 出流量下降了一個數量級。
各業務核心實時報表可以穩定產出。由於核心報表的計算直接由數倉負責，視覺化系統直接讀取實時結果，保證了實時報表的穩定性，目前多個業務擁有實時大盤，實時報表達 40+。
提升了即席查詢的穩定性。核心報表的指標計算轉移到數倉，Druid 只負責即席查詢，多維分析類的需求得到了滿足。
成本計算需求得到了解決。由於各業務擁有了獨立的資料來源且各核心大盤由不同的實時程式負責，可以方便的統計各業務使用的儲存資源和計算資源。

三、實時數倉未來展望

從實時數倉 1.0 到 2.0，不管是資料架構還是技術方案，我們在深度和廣度上都有了更多的積累。隨著公司業務的快速發展以及新技術的誕生，實時數倉也會不斷的迭代優化。短期可預見的我們會從以下方面進一步提升實時數倉的服務能力。

1、Streaming SQL 平臺化

目前 Streaming SQL 任務是以程式碼開發 maven 打包的方式提交任務，開發成本高，後期隨著 Streaming SQL 平臺的上線，實時數倉的開發方式也會由 Jar 包轉變為 SQL 檔案。

2、實時資料元資訊管理系統化

對數倉元資訊的管理可以大幅度降低使用資料的成本，離線數倉的元資訊管理已經基本完善，實時數倉的元資訊管理才剛剛開始。

3、實時數倉結果驗收自動化

對實時結果的驗收只能藉助與離線資料指標對比的方式，以 Hive 和 Kafka 資料來源為例，分別執行 Hive SQL 和 Flink SQL，統計結果並對比是否一致實現實時結果驗收的自動化。

作者：xiaocui

來源： https://zhuanlan.zhihu.com/p/56807637

dbaplus社群歡迎廣大技術人員投稿，投稿郵箱：[email protected]

實用好文：知乎實時數倉架構實踐及演進

您可能也會喜歡…