https://mp.weixin.qq.com/s/hx-q13QteNvtXRpNsE5Y0A
作者 | 知乎數據工程團隊編輯 | VincentAI 前線導讀：“數據智能” (Data Intelligence) 有一個必須且基礎的環節，就是數據倉庫的建設，同時，數據倉庫也是公司數據發展到一定規模后必然會提供的一種基礎服務。從智能商業的角度來講，數據的結果代表了用戶的反饋，獲取結果的及時性就顯得尤為重要，快速的獲取數據反饋能夠幫助公司更快的做出決策，更好的進行產品迭代，實時數倉在這一過程中起到了不可替代的作用。
更多優質內容請關注微信公眾號“AI 前線”（ID：ai-front）
本文主要講述知乎的實時數倉實踐以及架構的演進，這包括以下幾個方面：
-
實時數倉 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 配合前端完成數據可視化。
其中第一、三部分的相關內容請分別參考：知乎客戶端埋點流程、模型和平台技術，Druid 與知乎數據分析平台，此處我們詳細介紹第二部分。由於實時數據流的穩定性不如離線數據流，當實時流出現問題后需要離線數據重刷歷史數據，因此實時處理部分我們采用了 lambda 架構。
Lambda 架構有高容錯、低延時和可擴展的特點，為了實現這一設計，我們將 ETL 工作分為兩部分：Streaming ETL 和 Batch ETL。
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 參數解析、新老用戶識別。
動態配置 Streaming
由於 Streaming 任務需要 7 * 24 小時運行，但有些業務邏輯，比如：存在一個元數據信息中心，當這個元數據發生變化時，需要將這種變化映射到數據流上方便下游使用數據，這種變化可能需要停止 Streaming 任務以更新業務邏輯，但元數據變化的頻率非常高，且在元數據變化后如何及時通知程序的維護者也很難。動態配置 Streaming 為我們提供了一個解決方案，該方案如下圖所示。
我們可以把經常變化的元數據作為 Streaming Broadcast 變量，該變量扮演的角色類似於只讀緩存，同時針對該變量可設置 TTL，緩存過期后 Executor 節點會重新向 Driver 請求最新的變量。通過這種機制可以非常自然的將元數據的變化映射到數據流上，無需重啟任務也無需通知程序的維護者。
UTM 參數解析
UTM 的全稱是 Urchin Tracking Module，是用於追蹤網站流量來源的利器，關於 UTM 背景知識介紹可以參考網上其他內容，這里不再贅述。下圖是我們解析 UTM 信息的完整邏輯。
流量數據通過 UTM 參數解析后，我們可以很容易滿足以下需求：
-
查看各搜索引擎導流情況以及這些流量來自於哪些熱門搜索詞。
-
市場部某次活動帶來的流量大小，如：頁面瀏覽數、獨立訪問用戶數等。
-
從站內分享出去的鏈接在各分享平台（如：微信、微博）被瀏覽的情況。
新老用戶識別
對於互聯網公司而言，增長是一個永恆的話題，實時拿到新增用戶量，對於增長運營十分重要。例如：一次投放 n 個渠道，如果能拿到每個渠道的實時新增用戶數，就可以快速判斷出那些渠道更有價值。我們用下圖來表達 Streaming ETL 中是如何識別新老用戶的。
判斷一個用戶是不是新用戶，最簡單的辦法就是維護一個歷史用戶池，對每條日志判斷該用戶是否存在於用戶池中。 由於日志量巨大，為了不影響 Streaming 任務的處理速度，我們設計了兩層緩存：Thread Local Cache 和 Redis Cache，同時用 HBase 做持久化存儲以保存歷史用戶。訪問速度：本地內存 > 遠端內存 > 遠端磁盤，對於我們這個任務來說，只有 1% 左右的請求會打到 HBase，日志高峰期 26w/s，完全不會影響任務的實時性。當然本地緩存 LruCache 的容量大小和 Redis 的性能也是影響實時性的兩個因素。
Streaming ETL 除了上述幾個通用場景外，還有一些其他邏輯，這些邏輯的存在有的是為了滿足下游更方便的使用數據的需求，有的是對某些錯誤埋點的修復，總之 Streaming ETL 在整個實時數倉中處於指標計算的上游，有着不可替代的作用。
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 任務。
Batch ETL
接下來要介紹的是 Lambda 架構的第二個部分：Batch ETL，此部分我們需要解決數據落地、離線 ETL、數據批量導入 Druid 等問題。針對數據落地我們自研了 map reduce 任務 Batch Loader，針對數據修復我們自研了離線任務 Repair ETL，離線修復邏輯和實時邏輯共用一套 ETL Lib，針對批量導入 ProtoParquet 數據到 Druid，我們擴展了 Druid 的導入插件。
Repair ETL
數據架構圖中有兩個 Kafka，第一個 Kafka 存放的是原始日志，第二個 Kafka 存放的是實時 ETL 后的日志，我們將兩個 Kafka 的數據全部落地，這樣做的目的是為了保證數據鏈路的穩定性。因為實時 ETL 中有大量的業務邏輯，未知需求的邏輯也許會給整個流量數據帶來安全隱患，而上游的 Log Collect Server 不存在任何業務邏輯只負責收發日志，相比之下第一個 Kafka 的數據要安全和穩定的多。Repair ETL 並不是經常啟用，只有當實時 ETL 丟失數據或者出現邏輯錯誤時，才會啟用該程序用於修復日志。
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 數據。
Batch Load 2 Druid
采用 Tranquility 實時導入 Druid，這種方式強制需要一個時間窗口，當上游數據延遲超過窗值后會丟棄窗口之外的數據，這種情況會導致實時報表出現指標錯誤。為了修復這種錯誤，我們通過 Druid 發起一個離線 Map Reduce 任務定期重導上一個時間段的數據。通過這里的 Batch 導入和前面的實時導入，實現了實時數倉的 Lambda 架構。
實時數倉 1.0 的幾個不足之處
到目前為止我們已經介紹完 Lambda 架構實時數倉的幾個模塊，1.0 版本的實時數倉有以下幾個不足：
-
所有的流量數據存放在同一個 Kafka Topic 中，如果下游每個業務線都要消費，這會導致全量數據被消費多次，Kafka 出流量太高無法滿足該需求。
-
所有的指標計算全部由 Druid 承擔，Druid 同時兼顧實時數據源和離線數據源的查詢，隨着數據量的暴漲 Druid 穩定性急劇下降，這導致各個業務的核心報表不能穩定產出。
-
由於每個業務使用同一個流量數據源配置報表，導致查詢效率低下，同時無法對業務做數據隔離和成本計算。
實時數倉 2.0 版本
隨着數據量的暴漲，Druid 中的流量數據源經常查詢超時同時各業務消費實時數據的需求也開始增多，如果繼續沿用實時數倉 1.0 架構，需要付出大量的額外成本。於是，在實時數倉 1.0 的基礎上，我們建立起了實時數倉 2.0，梳理出了新的架構設計並開始着手建立實時數倉體系，新的架構如下圖所示。
原始層
實時數倉 1.0 我們只對流量數據做 ETL 處理，在 2.0 版本中我們加入了對業務庫的變更日志 Binlog 的處理，Binlog 日志在原始層為庫級別或者 Mysql 實例級別，即：一個庫或者實例的變更日志存放在同一個 Kafka Topic 中。同時隨着公司業務的發展不斷有新 App 產生，在原始層不僅采集「知乎」日志，像知乎極速版以及內部孵化項目的埋點數據也需要采集，不同 App 的埋點數據仍然使用同一套 PB Schema。
明細層
明細層是我們的 ETL 層，這一層數據是由原始層經過 Streaming ETL 后得到。其中對 Binlog 日志的處理主要是完成庫或者實例日志到表日志的拆分，對流量日志主要是做一些通用 ETL 處理，由於我們使用的是同一套 PB 結構，對不同 App 數據處理的邏輯代碼可以完全復用，這大大降低了我們的開發成本。
匯總層之明細匯總
明細匯總層是由明細層通過 ETL 得到，主要以寬表形式存在。業務明細匯總是由業務事實明細表和維度表 Join 得到，流量明細匯總是由流量日志按業務線拆分和流量維度 Join 得到。流量按業務拆分后可以滿足各業務實時消費的需求，我們在流量拆分這一塊做到了自動化，下圖演示了流量數據自動切分的過程。
