Arctic 基于 Hive 的流批一體實(shí)踐

　　背景

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

　　什么是 Arctic

　　Arctic 是搭建在 Apache Iceberg 之上的流式湖倉(cāng)服務(wù) ( Streaming LakeHouse Service )。相比 Iceberg、Hudi、Delta 等數(shù)據(jù)湖，Arctic 提供了更加優(yōu)化的 CDC，流式更新，OLAP 等功能，并且結(jié)合了 Iceberg 高效的離線處理能力，Arctic 能服務(wù)于更多的流批混用場(chǎng)景。Arctic 還提供了包括結(jié)構(gòu)自優(yōu)化、并發(fā)沖突解決、標(biāo)準(zhǔn)化的湖倉(cāng)管理功能等，可以有效減少數(shù)據(jù)湖在管理和優(yōu)化上負(fù)擔(dān)。

　　Arctic Table 依賴 Iceberg 作為基礎(chǔ)表格式，但是 Arctic 沒(méi)有傾入 Iceberg 的實(shí)現(xiàn)，而是將 Iceberg 做為 lib 使用，同時(shí) Arctic 作為專門為流批一體計(jì)算設(shè)計(jì)的流式湖倉(cāng)，Arctic Table 還封裝了消息隊(duì)列作為表的一部分，在流式計(jì)算場(chǎng)景下可以提供更低的消息延遲，并且提供了流式更新，主鍵唯一性保證等功能。

　　流體一批的解決方案

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

　　這套架構(gòu)最明顯的問(wèn)題就是多套系統(tǒng)帶來(lái)的運(yùn)維成本和重復(fù)開(kāi)發(fā)帶來(lái)的低效率，其次就是兩套系統(tǒng)同時(shí)建模帶來(lái)的語(yǔ)義二義性問(wèn)題，并且真實(shí)生產(chǎn)場(chǎng)景中，還會(huì)出現(xiàn)實(shí)時(shí)和離線視圖合并的需求，或者引入 KV 的實(shí)時(shí)維表關(guān)聯(lián)的需求。

　　Arctic 的核心目標(biāo)之一，就是為業(yè)務(wù)提供基于數(shù)據(jù)湖的去 Lambda 化，業(yè)務(wù)系統(tǒng)使用 Arctic 替代 Kafka 和Hive，實(shí)現(xiàn)存儲(chǔ)底座的流批一體。

　　為此 Arctic 提供了以下功能：

　　Message Queue 的封裝：Arctic 通過(guò)將 MessageQueue 和數(shù)據(jù)湖封裝成一張表，實(shí)現(xiàn)了 Spark、Flink、Trino 等不同計(jì)算引擎訪問(wèn)時(shí)不需要區(qū)分流表和批表，實(shí)現(xiàn)了計(jì)算指標(biāo)上的統(tǒng)一。

　　毫秒級(jí)流計(jì)算延遲：Message Queue 提供了毫秒級(jí)的讀延遲，并且提供了數(shù)據(jù)寫(xiě)入和讀取的一致性保障。

　　分鐘級(jí)的 OLAP 延遲：Arctic 支持流式寫(xiě)入以及流式更新，在查詢時(shí)通過(guò) Merge on Read 實(shí)現(xiàn)分鐘級(jí)的 OLAP 查詢。

　　Table Store

　　Arctic Table 由不同的 Table Store 組成，TableStore 是 Arctic 在存儲(chǔ)系統(tǒng)中定義的表格式實(shí)體，Tablestore 類似于數(shù)據(jù)庫(kù)中的 cluster index，代表獨(dú)立的存儲(chǔ)結(jié)構(gòu)，目前分為三種 TableStore。

　　ChangeStore

　　ChangeStroe 是一張 Iceberg 表，它代表了表上的增量數(shù)據(jù)，或者說(shuō)最新的數(shù)據(jù)變更，通常由 Apache Flink 任務(wù)實(shí)時(shí)寫(xiě)入，并用于下游任務(wù)近實(shí)時(shí)的消費(fèi)。

　　BaseStore

　　BaseStore 也是張 Iceberg 表，它代表了表上的存量數(shù)據(jù)。通常來(lái)自批計(jì)算的全量初始化，或者通過(guò)Optimizer 定時(shí)將來(lái)自 ChangeStore 的數(shù)據(jù)合并入 BaseStore。在對(duì)Arctic 表執(zhí)行查詢時(shí)， BaseStore 的數(shù)據(jù)會(huì)聯(lián)合 ChangeStore 的數(shù)據(jù)一起通過(guò)Merge-On-Read 返回。

　　LogStore

　　盡管 Changestore 已經(jīng)能夠?yàn)楸硖峁┙鼘?shí)時(shí)的 CDC 能力，但在對(duì)延遲有更高要求的場(chǎng)景仍然需要諸如 Apache Kafka 這樣的消息隊(duì)列提供毫秒級(jí)的 CDC 數(shù)據(jù)分發(fā)能力。而消息隊(duì)列在 Arctic 表中被封裝為 Logstore。它由 Flink 任務(wù)實(shí)時(shí)寫(xiě)入，并用于下游 Flink 任務(wù)進(jìn)行實(shí)時(shí)消費(fèi)。

　　Arctic 對(duì) Hive 的兼容

　　在真實(shí)業(yè)務(wù)實(shí)踐中，Hive 有著非常龐大的存量用戶以及圍繞其構(gòu)建的中臺(tái)體系，要想一步直接完成從 Hive 到湖倉(cāng)系統(tǒng)的轉(zhuǎn)換難度非常大，因此如何利用已有的 Hive 生態(tài)是 Arctic 實(shí)現(xiàn)流批一體首先需要解決的問(wèn)題。為此 Arctic 提供了 Hive 兼容的能力，以幫助 Hive 用戶可以平滑的遷移到流式數(shù)倉(cāng)中。具體到細(xì)節(jié)，Arctic 提供了以下 Hive 兼容能力：

　　數(shù)據(jù)訪問(wèn)層面的兼容：Arctic 與 Hive原生的讀寫(xiě)方式保持兼容，即通過(guò) Arctic 寫(xiě)入的數(shù)據(jù)，Hive 可以讀;Hive 寫(xiě)入的數(shù)據(jù)，Arctic 可以讀。

　　元數(shù)據(jù)層面的兼容：Arctic 表可以在 HMS 上注冊(cè)并管理，用戶直接對(duì) Hive 表執(zhí)行 DDL 可以被 Arctic 感知到。

　　Hive 生態(tài)的兼容：Arctic 表可以復(fù)用目前圍繞 Hive 的生態(tài)，比如可以直接通過(guò) ranger 對(duì) Hive 進(jìn)行權(quán)限管理的方式對(duì) Arctic 表進(jìn)行授權(quán)。

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

　　Hive 兼容的 Table Store

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

　　重點(diǎn)我們關(guān)注 Basestore 下的結(jié)構(gòu)，其中區(qū)分了兩個(gè)目錄空間：

　　hive location: Hive 表(或分區(qū))的目錄空間，會(huì)記錄在 Hive Meta Store 中，用原生的 Hive reader 會(huì)讀到這部分?jǐn)?shù)據(jù)。

　　iceberg location: 存儲(chǔ)近實(shí)時(shí)寫(xiě)入數(shù)據(jù)的目錄空間，用 Iceberg 管理，包含 insert file 與 delete file，原生的 Hive reader 無(wú)法讀取到其中的數(shù)據(jù)， Arctic reader 能讀取到。

　　兩個(gè)目錄空間的設(shè)計(jì)保障了支持 Arctic 完整特性的基礎(chǔ)之上仍然兼容 Hive 原生讀取。

　　Hive 數(shù)據(jù)同步

　　Hive location 的劃分實(shí)現(xiàn)了 Arctic 寫(xiě)入數(shù)據(jù)對(duì) Hive 查詢引擎讀的兼容，但是通過(guò) Hive 查詢引擎寫(xiě)入的數(shù)據(jù)或者 schema 變更卻無(wú)法讓 Arctic 立即識(shí)別，為此 Arctic 引入了 Hive Syncer 用于識(shí)別通過(guò) Hive 查詢引擎對(duì)表結(jié)構(gòu)和數(shù)據(jù)的變更。Hive Syncer 包括 2 個(gè)目標(biāo)：

　　Hive 表結(jié)構(gòu)變更同步到 Arctic

　　Hive 表數(shù)據(jù)變更同步到 Arctic

　　Table Metadata Sync

　　Hive 表結(jié)構(gòu)信息的同步是通過(guò)對(duì)比 Arctic Table Schema 和 Hive Table Schema 的差異實(shí)現(xiàn)的，由于對(duì)比代價(jià)較小，Arctic 采取的方式是在所有的讀取/寫(xiě)入/schema 查詢/變更 執(zhí)行前都會(huì)執(zhí)行 Metadata Sync 操作。通過(guò)對(duì) Schema 的對(duì)比，Arctic 可以自動(dòng)識(shí)別在 Hive 表上的 DDL 變更。Hive Schema 的同步能力使得 Arctic 的數(shù)據(jù)開(kāi)發(fā)可以繼續(xù)復(fù)用Hive生態(tài)下的數(shù)據(jù)建模工具，數(shù)據(jù)開(kāi)發(fā)只需要如同對(duì) Hive 表建模一樣即可完成對(duì) Arctic 表的建模。

　　Table Data Sync

　　Hive 表數(shù)據(jù)的變更的檢查是通過(guò)分區(qū)下的 transient_lastDdlTime 字段識(shí)別的，讀取 Hive 分區(qū)下數(shù)據(jù)時(shí)會(huì)對(duì)比分區(qū)的修改時(shí)間是否和 Arctic 的 metadata 中記載是否一致，如果不一致就通過(guò) HDFS 的 listDir 接口獲取分區(qū)下的全部文件，并對(duì)比 Arctic 表最新 snapshot 對(duì)應(yīng)的文件，如果文件列表有差異，說(shuō)明有通過(guò)非 Arctic 的途徑對(duì) Hive 表的數(shù)據(jù)進(jìn)行了修改，此時(shí) Arctic 會(huì)生成一個(gè)新的快照，對(duì) Arctic 表的文件信息進(jìn)行修正。

　　由于 HDFS 的 listDir 操作是一個(gè)比較重的操作，默認(rèn)情況下是通過(guò) AMS 定時(shí)觸發(fā) DataSync 檢查，如果對(duì)數(shù)據(jù)一致性要求更高，可以通過(guò)參數(shù) base.hive.auto-sync-data-write 配置為每次查詢前進(jìn)行 Data Sync 檢查。

　　Hive 數(shù)據(jù)同步的能力使得用戶從離線開(kāi)發(fā)鏈路遷移到實(shí)時(shí)開(kāi)發(fā)鏈接的過(guò)程中保留離線數(shù)據(jù)開(kāi)發(fā)的邏輯，通過(guò)離線完成對(duì)實(shí)時(shí)的數(shù)據(jù)修正，并且保證了實(shí)時(shí)和離線建模的統(tǒng)一以及指標(biāo)的統(tǒng)一。

　　存量 Hive 表原地升級(jí)

　　Arctic 不僅支持創(chuàng)建 Hive 兼容表，還支持直接將已經(jīng)存在的 Hive 表升級(jí)為一張 Arctic 下的 Hive 兼容表。在 AMS 上導(dǎo)入 HMS 對(duì)應(yīng)的 hive-site.xml 即可看到 HMS 上對(duì)應(yīng)的表，在對(duì)應(yīng)的 Hive 表上點(diǎn)擊 Upgrade 按鈕即可對(duì) Hive 表進(jìn)行原地升級(jí)。

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

　　Hive 的原地升級(jí)操作是非常輕量級(jí)的，在執(zhí)行 Upgrade 操作的背后，AMS 僅僅是新建一個(gè)空的 Arctic Table，然后掃描 Hive 目錄，并創(chuàng)建一個(gè)包括所有 Hive 下的 Parquet 文件的 Snapshot 即可，整個(gè)過(guò)程并不涉及到數(shù)據(jù)文件的復(fù)制和重寫(xiě)。

　　兼容 Hive 表的權(quán)限管理

　　圍繞著 Hive 已經(jīng)有了一套完整的大數(shù)據(jù)生態(tài)，其中對(duì)于表的權(quán)限管理和數(shù)據(jù)脫敏極為重要，當(dāng)前 Arctic的 Hive 兼容表已經(jīng)適配了 incubator-kyuubi 項(xiàng)目下的 spark-auth 插件 https://github.com/apache/incubator-kyuubi 通過(guò)該插件 Arctic 完成了對(duì) Ranger 的適配，在實(shí)際應(yīng)用中，通過(guò) Ranger 對(duì) Arctic 對(duì)應(yīng)的 Hive 進(jìn)行授權(quán)，在 SparkJob 中即可完成對(duì) Arctic 表的鑒權(quán)。

　　基于Hive 的流批一體實(shí)踐

　　Arctic 的 Hive 兼容模式是為了幫助適應(yīng)了 Hive 的用戶快速上手 Arctic，對(duì)于 Hive 用戶來(lái)說(shuō)，如果滿足以下其中一點(diǎn)：

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

　　2. 在離線場(chǎng)景下有成熟的產(chǎn)品構(gòu)建，并且希望為離線賦予部分實(shí)時(shí)的能力，但是又不想對(duì)離線平臺(tái)做過(guò)多的改造

　　即可嘗試通過(guò) Arctic Hive 兼容表解決你的痛點(diǎn)。

　　實(shí)踐案例：網(wǎng)易云音樂(lè)特征生產(chǎn)工程實(shí)時(shí)化

　　網(wǎng)易云音樂(lè)的推薦業(yè)務(wù)圍繞著 Spark+Hive 已經(jīng)構(gòu)建了一套成熟的大數(shù)據(jù)+AI 開(kāi)發(fā)體系，隨著業(yè)務(wù)的增長(zhǎng)，業(yè)務(wù)對(duì)整套系統(tǒng)的實(shí)時(shí)性要求在不斷增強(qiáng)，但是直接通過(guò) Flink + Kafka 構(gòu)建的實(shí)時(shí)鏈路并不夠完善。在離線鏈路中圍繞著 Hive 有著完善的基礎(chǔ)設(shè)施和方法論，數(shù)據(jù)開(kāi)發(fā)和算法工程師通過(guò)模型設(shè)計(jì)中心完成表的設(shè)計(jì)，數(shù)據(jù)開(kāi)發(fā)負(fù)責(zé)數(shù)據(jù)的攝取，清洗，打?qū)�，聚合等基礎(chǔ)處理，算法工程師負(fù)責(zé)在 DWS 層的數(shù)據(jù)上實(shí)現(xiàn)特征生產(chǎn)算法，分析師通過(guò)對(duì) ODS 層、DWD 層以及 DWS 層的表執(zhí)行Ad Hoc 式的查詢并構(gòu)建分析報(bào)表以評(píng)估特征數(shù)據(jù)質(zhì)量。整套鏈路層次分明、分工清晰，即最大限度的復(fù)用了計(jì)算結(jié)果，又比較好的統(tǒng)一了指標(biāo)口徑，是典型的 T+1 的數(shù)倉(cāng)建設(shè)。但是在實(shí)時(shí)鏈路中，數(shù)據(jù)開(kāi)發(fā)僅僅協(xié)助完成原始數(shù)據(jù)到 Kafka 的攝取，算法工程師需要從 ODS 層數(shù)據(jù)進(jìn)行加工，整個(gè)鏈路缺乏數(shù)據(jù)分層，既不能復(fù)用離線計(jì)算結(jié)果，也無(wú)法保證指標(biāo)的一致性。

　　整個(gè)特征工程的生產(chǎn)路線的現(xiàn)狀如下圖所示：

　　由于存在大量的存量 Hive 表，并且還有來(lái)自 Presto 和 Impala 的查詢鏈路需要復(fù)用 ODS 和 DWD 層的 Hive 表，整個(gè)特征工程想直接使用 Iceberg 或 Hudi 這樣的系統(tǒng)其切換代價(jià)還是很大的，系統(tǒng)切換期間對(duì)系統(tǒng)整體 SLA 要求較高，新系統(tǒng)磨合過(guò)程中如果造成數(shù)據(jù)產(chǎn)出延遲，對(duì)于業(yè)務(wù)來(lái)說(shuō)是不可接受的。最終我們采用了 Arctic Hive 兼容表的模式， 分階段的將 Hive 表升級(jí)為 Arctic 下的 Hive 兼容表，升級(jí)后的數(shù)據(jù)生產(chǎn)鏈路如下圖所示：

　　升級(jí)后Arctic 為整個(gè)特征工程帶來(lái)了以下好處：

　　1. Arctic 以無(wú)感知的方式完成了約 2PB 級(jí)別的 Hive 表實(shí)時(shí)化，由于做到 Hive 的讀寫(xiě)兼容，本身 T+1 的全量數(shù)據(jù)回補(bǔ)以及分析師的報(bào)表查詢 SQL 不用做任何修改，升級(jí)過(guò)程中保證了不影響離線鏈路開(kāi)發(fā)。

　　2. 實(shí)時(shí)特征的生產(chǎn)復(fù)用了數(shù)倉(cāng) DWS 層數(shù)據(jù)，不需要從 ODS 層直接構(gòu)建特征算法，而數(shù)倉(cāng)的清洗、聚合均由數(shù)據(jù)開(kāi)發(fā)完成，提升了算法工程師的人效，使得算法工程師可以更好的專注于特征算法本身。平均下來(lái)每個(gè)算法節(jié)省人效約 1 天。

　　3. 完成了實(shí)時(shí)鏈路和離線鏈路的統(tǒng)一，在數(shù)據(jù)血緣，數(shù)據(jù)指標(biāo)，模型設(shè)計(jì)上可以做到更好的數(shù)據(jù)治理。

　　4. Arctic 本身可以為 ODS 和 DWD 層的表配置更激進(jìn)的 Optimize 策略，以 10 分鐘的頻率對(duì) Hive Table 的數(shù)據(jù)進(jìn)行 Overwrite, 分析師可以享受到更加實(shí)時(shí)的分析報(bào)表。

　　總結(jié)

　　本文介紹了網(wǎng)易數(shù)帆開(kāi)源的新一代流式湖倉(cāng) Arctic 以及其基于 Hive 的流批一體實(shí)踐。希望讀者可以經(jīng)此文章了解 Arctic 并對(duì)業(yè)務(wù)構(gòu)建流批一體的數(shù)據(jù)湖有幫助。