Arctic 基于 Hive 的流批一體實踐

2022年11月07日 16:08:43 張永翔、胡溢勝來源：網易杭州研究院

　　背景

　　隨著大數據業務的發展，基于 Hive 的數倉體系逐漸難以滿足日益增長的業務需求，一方面已有很大體量的用戶，但是在實時性，功能性上嚴重缺失;另一方面 Hudi，Iceberg 這類系統在事務性，快照管理上帶來巨大提升，但是對已經存在的 Hive 用戶有較大的遷移成本，并且難以滿足流式計算毫秒級延遲的需求。為了滿足網易內外部客戶對于流批一體業務的需求，網易數帆基于 Apache Iceberg 研發了新一代流式湖倉，相較于 Hudi，Iceberg 等傳統湖倉，它提供了流式更新，維表 Join，partial upsert 等功能，并且將 Hive，Iceberg，消息隊列整合為一套流式湖倉服務，實現了開箱即用的流批一體，能幫助業務平滑地從 Hive 過渡到 Streaming Lakehouse。

　　什么是 Arctic

　　Arctic 是搭建在 Apache Iceberg 之上的流式湖倉服務 ( Streaming LakeHouse Service )。相比 Iceberg、Hudi、Delta 等數據湖，Arctic 提供了更加優化的 CDC，流式更新，OLAP 等功能，并且結合了 Iceberg 高效的離線處理能力，Arctic 能服務于更多的流批混用場景。Arctic 還提供了包括結構自優化、并發沖突解決、標準化的湖倉管理功能等，可以有效減少數據湖在管理和優化上負擔。

　　Arctic Table 依賴 Iceberg 作為基礎表格式，但是 Arctic 沒有傾入 Iceberg 的實現，而是將 Iceberg 做為 lib 使用，同時 Arctic 作為專門為流批一體計算設計的流式湖倉，Arctic Table 還封裝了消息隊列作為表的一部分，在流式計算場景下可以提供更低的消息延遲，并且提供了流式更新，主鍵唯一性保證等功能。

　　流體一批的解決方案

　　在實時計算中，由于低延遲的要求，業務通常采用 Kafka 這類消息隊列作為流表方案，但是在離線計算中，通常采用 Hive 作為離線表，并且由于消息隊列不支持 AP 查詢，通常還需要額外的 OLAP 系統如 Kudu 以支持實時計算鏈接的最終數據輸出。這就是典型的 Lambda 架構：

　　這套架構最明顯的問題就是多套系統帶來的運維成本和重復開發帶來的低效率，其次就是兩套系統同時建模帶來的語義二義性問題，并且真實生產場景中，還會出現實時和離線視圖合并的需求，或者引入 KV 的實時維表關聯的需求。

　　Arctic 的核心目標之一，就是為業務提供基于數據湖的去 Lambda 化，業務系統使用 Arctic 替代 Kafka 和Hive，實現存儲底座的流批一體。

　　為此 Arctic 提供了以下功能：

　　Message Queue 的封裝：Arctic 通過將 MessageQueue 和數據湖封裝成一張表，實現了 Spark、Flink、Trino 等不同計算引擎訪問時不需要區分流表和批表，實現了計算指標上的統一。

　　毫秒級流計算延遲：Message Queue 提供了毫秒級的讀延遲，并且提供了數據寫入和讀取的一致性保障。

　　分鐘級的 OLAP 延遲：Arctic 支持流式寫入以及流式更新，在查詢時通過 Merge on Read 實現分鐘級的 OLAP 查詢。

　　Table Store

　　Arctic Table 由不同的 Table Store 組成，TableStore 是 Arctic 在存儲系統中定義的表格式實體，Tablestore 類似于數據庫中的 cluster index，代表獨立的存儲結構，目前分為三種 TableStore。

　　ChangeStore

　　ChangeStroe 是一張 Iceberg 表，它代表了表上的增量數據，或者說最新的數據變更，通常由 Apache Flink 任務實時寫入，并用于下游任務近實時的消費。

　　BaseStore

　　BaseStore 也是張 Iceberg 表，它代表了表上的存量數據。通常來自批計算的全量初始化，或者通過Optimizer 定時將來自 ChangeStore 的數據合并入 BaseStore。在對Arctic 表執行查詢時， BaseStore 的數據會聯合 ChangeStore 的數據一起通過Merge-On-Read 返回。

　　LogStore

　　盡管 Changestore 已經能夠為表提供近實時的 CDC 能力，但在對延遲有更高要求的場景仍然需要諸如 Apache Kafka 這樣的消息隊列提供毫秒級的 CDC 數據分發能力。而消息隊列在 Arctic 表中被封裝為 Logstore。它由 Flink 任務實時寫入，并用于下游 Flink 任務進行實時消費。

　　Arctic 對 Hive 的兼容

　　在真實業務實踐中，Hive 有著非常龐大的存量用戶以及圍繞其構建的中臺體系，要想一步直接完成從 Hive 到湖倉系統的轉換難度非常大，因此如何利用已有的 Hive 生態是 Arctic 實現流批一體首先需要解決的問題。為此 Arctic 提供了 Hive 兼容的能力，以幫助 Hive 用戶可以平滑的遷移到流式數倉中。具體到細節，Arctic 提供了以下 Hive 兼容能力：

　　數據訪問層面的兼容：Arctic 與 Hive原生的讀寫方式保持兼容，即通過 Arctic 寫入的數據，Hive 可以讀;Hive 寫入的數據，Arctic 可以讀。

　　元數據層面的兼容：Arctic 表可以在 HMS 上注冊并管理，用戶直接對 Hive 表執行 DDL 可以被 Arctic 感知到。

　　Hive 生態的兼容：Arctic 表可以復用目前圍繞 Hive 的生態，比如可以直接通過 ranger 對 Hive 進行權限管理的方式對 Arctic 表進行授權。

　　存量 Hive 表的兼容：海量的存量 Hive 表，如果有實時化的需求，可以以很低的代價將 Hive 表升級為 Arctic 表。

　　Hive 兼容的 Table Store

　　解決 Hive 兼容的首要問題是需要解決 Hive 和 Arctic 文件分布上的不同，在 Arctic 表中被分為 ChangeStore、BaseStore、LogStore 三個不同的 Table Store，從定義上，BaseStore 代表著表的存量數據，這與 Hive 的離線數倉定位是一致的，但是在實現上，Arctic 并未直接將 BaseStore 替換為 Hive Table ，而是仍然保留 Iceberg Table 作為 BaseStore 的實現以提供 ACID 等特性，并通過目錄劃分的方式，劃分出對 Hive 兼容的目錄空間，具體結構如下圖所示：

　　重點我們關注 Basestore 下的結構，其中區分了兩個目錄空間：

　　hive location: Hive 表(或分區)的目錄空間，會記錄在 Hive Meta Store 中，用原生的 Hive reader 會讀到這部分數據。

　　iceberg location: 存儲近實時寫入數據的目錄空間，用 Iceberg 管理，包含 insert file 與 delete file，原生的 Hive reader 無法讀取到其中的數據， Arctic reader 能讀取到。

　　兩個目錄空間的設計保障了支持 Arctic 完整特性的基礎之上仍然兼容 Hive 原生讀取。

　　Hive 數據同步

　　Hive location 的劃分實現了 Arctic 寫入數據對 Hive 查詢引擎讀的兼容，但是通過 Hive 查詢引擎寫入的數據或者 schema 變更卻無法讓 Arctic 立即識別，為此 Arctic 引入了 Hive Syncer 用于識別通過 Hive 查詢引擎對表結構和數據的變更。Hive Syncer 包括 2 個目標：

　　Hive 表結構變更同步到 Arctic

　　Hive 表數據變更同步到 Arctic

　　Table Metadata Sync

　　Hive 表結構信息的同步是通過對比 Arctic Table Schema 和 Hive Table Schema 的差異實現的，由于對比代價較小，Arctic 采取的方式是在所有的讀取/寫入/schema 查詢/變更執行前都會執行 Metadata Sync 操作。通過對 Schema 的對比，Arctic 可以自動識別在 Hive 表上的 DDL 變更。Hive Schema 的同步能力使得 Arctic 的數據開發可以繼續復用Hive生態下的數據建模工具，數據開發只需要如同對 Hive 表建模一樣即可完成對 Arctic 表的建模。

　　Table Data Sync

　　Hive 表數據的變更的檢查是通過分區下的 transient_lastDdlTime 字段識別的，讀取 Hive 分區下數據時會對比分區的修改時間是否和 Arctic 的 metadata 中記載是否一致，如果不一致就通過 HDFS 的 listDir 接口獲取分區下的全部文件，并對比 Arctic 表最新 snapshot 對應的文件，如果文件列表有差異，說明有通過非 Arctic 的途徑對 Hive 表的數據進行了修改，此時 Arctic 會生成一個新的快照，對 Arctic 表的文件信息進行修正。

　　由于 HDFS 的 listDir 操作是一個比較重的操作，默認情況下是通過 AMS 定時觸發 DataSync 檢查，如果對數據一致性要求更高，可以通過參數 base.hive.auto-sync-data-write 配置為每次查詢前進行 Data Sync 檢查。

　　Hive 數據同步的能力使得用戶從離線開發鏈路遷移到實時開發鏈接的過程中保留離線數據開發的邏輯，通過離線完成對實時的數據修正，并且保證了實時和離線建模的統一以及指標的統一。

　　存量 Hive 表原地升級

　　Arctic 不僅支持創建 Hive 兼容表，還支持直接將已經存在的 Hive 表升級為一張 Arctic 下的 Hive 兼容表。在 AMS 上導入 HMS 對應的 hive-site.xml 即可看到 HMS 上對應的表，在對應的 Hive 表上點擊 Upgrade 按鈕即可對 Hive 表進行原地升級。

　　Arctic 還支持在進行原地升級時指定主鍵，這樣可以將 Hive 表升級為有主鍵的 Arctic 表。

　　Hive 的原地升級操作是非常輕量級的，在執行 Upgrade 操作的背后，AMS 僅僅是新建一個空的 Arctic Table，然后掃描 Hive 目錄，并創建一個包括所有 Hive 下的 Parquet 文件的 Snapshot 即可，整個過程并不涉及到數據文件的復制和重寫。

　　兼容 Hive 表的權限管理

　　圍繞著 Hive 已經有了一套完整的大數據生態，其中對于表的權限管理和數據脫敏極為重要，當前 Arctic的 Hive 兼容表已經適配了 incubator-kyuubi 項目下的 spark-auth 插件 https://github.com/apache/incubator-kyuubi 通過該插件 Arctic 完成了對 Ranger 的適配，在實際應用中，通過 Ranger 對 Arctic 對應的 Hive 進行授權，在 SparkJob 中即可完成對 Arctic 表的鑒權。

　　基于Hive 的流批一體實踐

　　Arctic 的 Hive 兼容模式是為了幫助適應了 Hive 的用戶快速上手 Arctic，對于 Hive 用戶來說，如果滿足以下其中一點：

　　1. 有大量的存量 Hive 表，并且其中部分 Hive 表有流式寫入、訂閱的需求

　　2. 在離線場景下有成熟的產品構建，并且希望為離線賦予部分實時的能力，但是又不想對離線平臺做過多的改造

　　即可嘗試通過 Arctic Hive 兼容表解決你的痛點。

　　實踐案例：網易云音樂特征生產工程實時化

　　網易云音樂的推薦業務圍繞著 Spark+Hive 已經構建了一套成熟的大數據+AI 開發體系，隨著業務的增長，業務對整套系統的實時性要求在不斷增強，但是直接通過 Flink + Kafka 構建的實時鏈路并不夠完善。在離線鏈路中圍繞著 Hive 有著完善的基礎設施和方法論，數據開發和算法工程師通過模型設計中心完成表的設計，數據開發負責數據的攝取，清洗，打寬，聚合等基礎處理，算法工程師負責在 DWS 層的數據上實現特征生產算法，分析師通過對 ODS 層、DWD 層以及 DWS 層的表執行Ad Hoc 式的查詢并構建分析報表以評估特征數據質量。整套鏈路層次分明、分工清晰，即最大限度的復用了計算結果，又比較好的統一了指標口徑，是典型的 T+1 的數倉建設。但是在實時鏈路中，數據開發僅僅協助完成原始數據到 Kafka 的攝取，算法工程師需要從 ODS 層數據進行加工，整個鏈路缺乏數據分層，既不能復用離線計算結果，也無法保證指標的一致性。

　　整個特征工程的生產路線的現狀如下圖所示：

　　由于存在大量的存量 Hive 表，并且還有來自 Presto 和 Impala 的查詢鏈路需要復用 ODS 和 DWD 層的 Hive 表，整個特征工程想直接使用 Iceberg 或 Hudi 這樣的系統其切換代價還是很大的，系統切換期間對系統整體 SLA 要求較高，新系統磨合過程中如果造成數據產出延遲，對于業務來說是不可接受的。最終我們采用了 Arctic Hive 兼容表的模式，分階段的將 Hive 表升級為 Arctic 下的 Hive 兼容表，升級后的數據生產鏈路如下圖所示：