Hudi 實(shí)踐 | Robinhood 基于 Apache Hudi 的下一代數(shù)據(jù)湖實(shí)踐
1. 摘要
Robinhood 的使命是使所有人的金融民主化。Robinhood 內(nèi)部不同級(jí)別的持續(xù)數(shù)據(jù)分析和數(shù)據(jù)驅(qū)動(dòng)決策是實(shí)現(xiàn)這一使命的基礎(chǔ)。我們有各種數(shù)據(jù)源——OLTP 數(shù)據(jù)庫(kù)、事件流和各種第 3 方數(shù)據(jù)源。需要快速、可靠、安全和以隱私為中心的數(shù)據(jù)湖攝取服務(wù)來(lái)支持各種報(bào)告、關(guān)鍵業(yè)務(wù)管道和儀表板。不僅在數(shù)據(jù)存儲(chǔ)規(guī)模和查詢(xún)方面，也在我們?cè)跀?shù)據(jù)湖支持的用例方面，我們從最初的數(shù)據(jù)湖版本[1]都取得了很大的進(jìn)展。在這篇博客中，我們將描述如何使用各種開(kāi)源工具構(gòu)建基于變更數(shù)據(jù)捕獲的增量攝取，以將我們核心數(shù)據(jù)集的數(shù)據(jù)新鮮延遲從 1 天減少到 15 分鐘以下。我們還將描述大批量攝取模型中的局限性，以及在大規(guī)模操作增量攝取管道時(shí)學(xué)到的經(jīng)驗(yàn)教訓(xùn)。
2. 數(shù)據(jù)湖和生態(tài)系統(tǒng)
Robinhood 的數(shù)據(jù)湖存儲(chǔ)和計(jì)算基礎(chǔ)架構(gòu)是為我們的許多數(shù)據(jù)驅(qū)動(dòng)功能提供支持的基石，例如業(yè)務(wù)分析儀表板和產(chǎn)品改進(jìn)見(jiàn)解。它也是為業(yè)務(wù)和臨時(shí)報(bào)告和分析運(yùn)行大規(guī)模數(shù)據(jù)處理的數(shù)據(jù)源。此外，生態(tài)系統(tǒng)會(huì)影響以隱私為中心的原語(yǔ)，例如旨在保護(hù)用戶(hù)隱私的匿名化和訪(fǎng)問(wèn)控制。
主要的 OLTP（在線(xiàn)事務(wù)處理）數(shù)據(jù)庫(kù)由 Postgres RDS 管理；Amazon S3 是 Data Lake 存儲(chǔ)，它為我們的 Data Lake 提供經(jīng)濟(jì)高效且可擴(kuò)展的存儲(chǔ)層；我們主要使用 Apache Spark 運(yùn)行生產(chǎn)批處理管道；我們的儀表板由 Trino 分布式 SQL 查詢(xún)引擎提供支持；Apache Hadoop Yarn 管理用于運(yùn)行 Apache Spark 作業(yè)的計(jì)算集群；Apache Hive Metastore 為查詢(xún)引擎管理和提供表模式；Apache Airflow 是工作流編排服務(wù)。下圖是具有計(jì)算生態(tài)系統(tǒng)的數(shù)據(jù)湖：
在整篇文章中我們使用指標(biāo)“數(shù)據(jù)新鮮度”來(lái)比較下面不同的數(shù)據(jù)攝取架構(gòu)，此指標(biāo)為源數(shù)據(jù)庫(kù)中的表中發(fā)生的更改在相應(yīng)的 Data Lake 表中可見(jiàn)提供了時(shí)間延遲。
3. 大批量攝取的限制
作為數(shù)據(jù)湖演進(jìn)的第一步，我們首先使用在線(xiàn)數(shù)據(jù)庫(kù)的只讀副本獲取在線(xiàn)數(shù)據(jù)庫(kù)的每日快照。攝取這些表的完整快照會(huì)導(dǎo)致數(shù)據(jù)湖表的寫(xiě)入放大率很高。即使對(duì)于一個(gè)有數(shù)十億行的表來(lái)說(shuō)，一天只有幾十萬(wàn)行的變化，攝取該表的完整快照也會(huì)導(dǎo)致讀取和寫(xiě)入整個(gè)表。此外當(dāng)使用實(shí)時(shí)副本（而不是作為上游的數(shù)據(jù)庫(kù)備份）時(shí)，在只讀副本 I/O 性能方面會(huì)出現(xiàn)瓶頸，這會(huì)導(dǎo)致快照時(shí)間過(guò)長(zhǎng)，從而導(dǎo)致較大的攝取延遲。即使采用了諸如通過(guò)分區(qū)讀取并行化 I/O 之類(lèi)的技術(shù)，這種攝取架構(gòu)也無(wú)法在一小時(shí)內(nèi)交付數(shù)據(jù)。Robinhood 確實(shí)需要保持?jǐn)?shù)據(jù)湖的低數(shù)據(jù)新鮮度。許多過(guò)去在市場(chǎng)交易時(shí)間之后或之前以每日節(jié)奏運(yùn)行的批處理管道必須以每小時(shí)或更高的頻率運(yùn)行，以支持不斷發(fā)展的用例。很明顯我們需要更快的攝取管道將在線(xiàn)數(shù)據(jù)庫(kù)復(fù)制到數(shù)據(jù)湖。
4. 新架構(gòu)
實(shí)現(xiàn) Data Lake 較低數(shù)據(jù)新鮮度的更好方法是增量攝取。增量攝取是一種眾所周知的技術(shù)，用于為數(shù)據(jù)湖構(gòu)建有效的攝取管道。在這里攝取管道不是拍攝快照并將它們作為一個(gè)整體轉(zhuǎn)儲(chǔ)到 Data Lake，而是以流方式使用 OLTP 數(shù)據(jù)庫(kù)的預(yù)寫(xiě)日志并將它們攝取到 Data Lake 表中，就像數(shù)據(jù)庫(kù)到數(shù)據(jù)庫(kù)復(fù)制的方式一樣。從概念上講，我們有一個(gè)兩階段管道。
?變更數(shù)據(jù)捕獲 (CDC) 服務(wù)使用 OLTP 數(shù)據(jù)庫(kù)中的預(yù)寫(xiě)日志 (WAL) 數(shù)據(jù)并將它們緩沖在變更日志隊(duì)列中。?數(shù)據(jù)攝取作業(yè)定期或以連續(xù)方式拖尾隊(duì)列并更新數(shù)據(jù)湖“原始”表。
下圖是增量攝取組件
中間更改日志隊(duì)列允許分離兩個(gè)階段之間的關(guān)注點(diǎn)，這兩個(gè)階段將能夠獨(dú)立運(yùn)行，并且每個(gè)階段都可以暫停而不影響另一個(gè)階段。隊(duì)列提供了必要的隔離，以便將數(shù)據(jù)攝取到數(shù)據(jù)湖的任何延遲都不會(huì)對(duì) CDC 造成背壓。在第一階段，我們選擇 Debezium 作為變更數(shù)據(jù)捕獲 (CDC) 提供商。Debezium 是一個(gè)構(gòu)建在 Kafka Connect 之上的開(kāi)源分布式變更數(shù)據(jù)捕獲平臺(tái)，Debezium 帶有一個(gè)經(jīng)過(guò)充分證明的一流 Postgres CDC 連接器。根據(jù)我們的基準(zhǔn)測(cè)試，我們發(fā)現(xiàn) Debezium 可以輕松處理我們預(yù)計(jì)的負(fù)載量，我們已經(jīng)設(shè)置 Debezium 使用開(kāi)源的 Confluent Schema Registry 以 avro 編碼格式將更改記錄寫(xiě)入 Kafka，與 json 編碼相比，Avro 編碼提供了更好的性能。在第二階段，我們使用 Apache Hudi 從 Kafka 增量攝取變更日志，以創(chuàng)建數(shù)據(jù)湖表。Apache Hudi 是一個(gè)統(tǒng)一的數(shù)據(jù)湖平臺(tái)，用于在數(shù)據(jù)湖上執(zhí)行批處理和流處理，Apache Hudi 帶有一個(gè)功能齊全的基于 Spark 的開(kāi)箱即用的攝取系統(tǒng)，稱(chēng)為 Deltastreamer，具有一流的 Kafka 集成和一次性寫(xiě)入功能，與不可變數(shù)據(jù)不同，我們的 CDC 數(shù)據(jù)有相當(dāng)大比例的更新和刪除，Hudi Deltastreamer 利用其可插入的記錄級(jí)索引在 Data Lake 表上執(zhí)行快速高效的 upserts，Hudi 通過(guò)自動(dòng)清理舊文件版本、數(shù)據(jù)Clustering、Hive表模式同步和文件大小調(diào)整來(lái)自我管理其表，以寫(xiě)入大小合適的文件，原始表當(dāng)前以 Hudi 的寫(xiě)時(shí)復(fù)制模式存儲(chǔ)，該模式提供原生列式讀取性能。
5. 效果總結(jié)
我們已經(jīng)部署了增量攝取管道，以將 1000 個(gè) Postgres 表攝取到數(shù)據(jù)湖中。在新架構(gòu)之前，由于快照的限制和所涉及的成本，這些表只能保證能夠以每天的節(jié)奏進(jìn)行快照。?使用這種新架構(gòu)，Data Lake 用戶(hù)很高興看到關(guān)鍵表的數(shù)據(jù)新鮮度從 24 小時(shí)縮短到 15 分鐘以下。?大批量快照運(yùn)行時(shí)間顯示快照表的運(yùn)行時(shí)間長(zhǎng)。請(qǐng)注意由于只讀副本 I/O 瓶頸，其中許多表的快照需要按順序運(yùn)行。
顯示大批量快照的大批量快照運(yùn)行計(jì)劃每天僅運(yùn)行一次，這是因?yàn)閺臄?shù)據(jù)庫(kù)中快照所有表的周轉(zhuǎn)時(shí)間很長(zhǎng)。
新的增量攝取數(shù)據(jù)新鮮度顯示新攝取系統(tǒng)的端到端數(shù)據(jù)新鮮度約為 5 分鐘。
6. 經(jīng)驗(yàn)教訓(xùn)
在本節(jié)中我們將分享在大規(guī)模構(gòu)建增量攝取管道時(shí)學(xué)到的經(jīng)驗(yàn)教訓(xùn)。我們希望這對(duì)任何希望為他們的數(shù)據(jù)湖踏上類(lèi)似旅程的人來(lái)說(shuō)都是有價(jià)值的。
7. 可縮放的初始引導(dǎo)程序
對(duì)數(shù)據(jù)湖的增量攝取仍然需要源表的初始快照。Debezium 確實(shí)提供了初始快照模式，但需要查詢(xún)主 RDS 實(shí)例，我們不想查詢(xún)主 RDS 實(shí)例以進(jìn)行快照，以避免生產(chǎn) OLTP 查詢(xún)與初始快照查詢(xún)之間的任何資源競(jìng)爭(zhēng)。此外，我們需要通過(guò)以無(wú)鎖方式運(yùn)行并發(fā)分區(qū)查詢(xún)以及從數(shù)據(jù)庫(kù)備份中獲取快照來(lái)優(yōu)化初始快照時(shí)間的能力。出于這些原因，我們?cè)?Apache Hudi Deltastreamer 之上提供了專(zhuān)用的只讀副本并實(shí)現(xiàn)了一個(gè)自定義快照器，它利用 Spark 運(yùn)行并發(fā)分區(qū)快照查詢(xún)來(lái)獲取表的初始快照，Apache Hudi 的可插拔源框架允許我們用幾行代碼無(wú)縫實(shí)現(xiàn)這一點(diǎn)。對(duì)于帶外初始快照，我們需要在增量攝取和快照之間切換時(shí)仔細(xì)跟蹤 CDC 流中的正確水印，使用 Kafka，數(shù)據(jù)攝取作業(yè)的 CDC 水印轉(zhuǎn)換為 Kafka 偏移量，這標(biāo)志著要應(yīng)用于快照表的開(kāi)始更改日志事件，如果我們選擇一個(gè)任意的 Kafka 偏移量，我們最終可能會(huì)錯(cuò)過(guò)一些應(yīng)用到 Data Lake 表的更改事件。
從概念上講，我們需要 3 個(gè)階段來(lái)執(zhí)行正確的快照并過(guò)渡到增量攝?。?/p>
?保存最新的 Kafka 偏移量，以在切換到增量攝取時(shí)用于重播變更日志。設(shè)“T?”為最新事件的源時(shí)間。?確保只讀副本在時(shí)間“T? + Δ”時(shí)是最新的，其中 Δ 表示捕獲 kafka 偏移量以及額外緩沖時(shí)間時(shí)的 Debezium 延遲。否則，整個(gè)方程式將無(wú)法保證 0% 的數(shù)據(jù)丟失。從只讀副本中獲取表的初始快照并創(chuàng)建 Data Lake 表?從之前存儲(chǔ)的 kafka 偏移量開(kāi)始消費(fèi)并執(zhí)行表的增量攝取。一旦增量攝取開(kāi)始發(fā)生，將配置單元表定義同步到數(shù)據(jù)的最新位置，下游消費(fèi)者現(xiàn)在將能夠查詢(xún)新引導(dǎo)的表。
下圖是使用引導(dǎo)架構(gòu)的增量攝取架構(gòu)
從專(zhuān)用只讀副本進(jìn)行快照具有局限性，例如副本端的 I/O 瓶頸以及 24 * 7 在線(xiàn)維護(hù)只讀副本的成本開(kāi)銷(xiāo)。我們正在探索一種對(duì) OLTP 數(shù)據(jù)庫(kù)進(jìn)行按需備份并使用 AWS S3 導(dǎo)出發(fā)布到 S3 的方法。然后我們可以依靠大規(guī)模處理這些 S3 導(dǎo)出并構(gòu)建初始快照，這種機(jī)制可能允許更快的快照并克服只讀副本端的一些 I/O 瓶頸。
8. 使用 Postgres 邏輯復(fù)制監(jiān)控背壓風(fēng)險(xiǎn)
Postgres 邏輯復(fù)制需要 CDC 連接器直連主 RDS。Postgres 邏輯復(fù)制協(xié)議保證保留 WAL 日志文件，直到 Debezium 完全處理它們。如果 Debezium 卡住或無(wú)法跟上消耗 WAL 日志的速度，這可能會(huì)導(dǎo)致 WAL 日志文件累積并耗盡可用磁盤(pán)空間，Debezium 社區(qū)建議密切監(jiān)視滯后消息，我們的 Debezium 負(fù)載測(cè)試也讓我們對(duì) Debezium 能夠處理預(yù)計(jì)的變更速度增加充滿(mǎn)信心。
9. 自動(dòng)化恢復(fù)
從每日快照切換到增量攝取的副作用之一是攝取工作流變得有狀態(tài)。管道可能處于快照或增量攝取狀態(tài)。此外，還需要執(zhí)行架構(gòu)升級(jí)、監(jiān)控和數(shù)據(jù)質(zhì)量驗(yàn)證等其他操作，新表和數(shù)據(jù)庫(kù)需要定期地加入。端到端管道涉及不同的系統(tǒng)——在線(xiàn) CDC 世界和數(shù)據(jù)湖的批處理/流攝取。為 1000 個(gè)表執(zhí)行入職和常規(guī)操作需要適當(dāng)?shù)臓顟B(tài)管理和自動(dòng)化。我們意識(shí)到我們需要在內(nèi)部構(gòu)建一流的編排服務(wù)，該服務(wù)將利用 Apache Airflow 來(lái)管理攝取管道、跟蹤載入和表狀態(tài)并自動(dòng)處理狀態(tài)轉(zhuǎn)換和其他維護(hù)，這有助于我們大規(guī)模運(yùn)營(yíng)管道。
10. 并非所有表都是平等的
當(dāng)談到這些表對(duì)我們的關(guān)鍵用例的重要性時(shí)，pareto原則是有效的，我們有一小部分關(guān)鍵表需要在 15 分鐘內(nèi)保證數(shù)據(jù)新鮮度，我們采取了一種方法，根據(jù)表的重要性將表分類(lèi)為不同的層，高度關(guān)鍵的表被標(biāo)記為第 0 層，對(duì)于這些表，我們提供了一個(gè)單獨(dú)的 CDC 復(fù)制槽，以將這些關(guān)鍵表的 CDC 通道與其他表的通道隔離。此外我們?yōu)?Hudi deltastreamer 提供了專(zhuān)門(mén)的資源，以持續(xù)攝取增量更改日志，并能夠在 5 -15 分鐘內(nèi)保持?jǐn)?shù)據(jù)最新。對(duì)于較低優(yōu)先級(jí)的表，Hudi deltastreamer 配置為以批處理模式每 15 分鐘運(yùn)行一次。
11. 管理 Postgres 模式更新
我們的業(yè)務(wù)是將表從在線(xiàn) OLTP 世界復(fù)制到 Data Lake 世界，復(fù)制的數(shù)據(jù)不是不透明的，而是具有適當(dāng)?shù)哪Ｊ?，并且?fù)制管道保證了將在線(xiàn)表模式轉(zhuǎn)換為數(shù)據(jù)湖的模式的明確定義的行為。鑒于 Data Lakes 還能夠存儲(chǔ)數(shù)據(jù)更改的整個(gè)歷史，因此在線(xiàn)和 Data Lake 世界的向后兼容性意味著什么不同。例如，在在線(xiàn)世界中，向 postgres 添加一個(gè)不可為空的列是非常好的，但不會(huì)遵守用于存儲(chǔ)動(dòng)態(tài)變更日志的 Avro（或 Protobuf）的模式演變規(guī)則。擁有明確定義的架構(gòu)演化合約有助于保持?jǐn)?shù)據(jù)湖管道更加穩(wěn)定。我們發(fā)現(xiàn)大多數(shù)時(shí)候，Schema更改涉及添加新列，我們正在使用 Debezium 功能來(lái)凍結(jié)我們從 Postgres 表中讀取的列集，并依靠重新引導(dǎo)表來(lái)處理模式升級(jí)，我們計(jì)劃為端到端管道添加模式兼容性檢測(cè)機(jī)制，以減少重新引導(dǎo)的次數(shù)。
12. 未來(lái)規(guī)劃
我們看到使用增量攝取的原始數(shù)據(jù)湖表的采用速度更快，并且我們正在不斷努力提高管道的可靠性。以下是我們正在著手的一些后續(xù)步驟：
?數(shù)據(jù)質(zhì)量保證：我們實(shí)施了以不同頻率運(yùn)行的通用和自定義數(shù)據(jù)質(zhì)量和完整性檢查，以發(fā)現(xiàn)復(fù)制數(shù)據(jù)中的差異，我們正在努力利用 Apache Hudi 的預(yù)提交驗(yàn)證支持在每批提交之前運(yùn)行自定義驗(yàn)證。?進(jìn)一步減少數(shù)據(jù)新鮮度滯后：我們目前使用的是 Apache Hudi Copy-On-Write 格式。在這種模式下，我們可以看到大約 5-15 分鐘范圍內(nèi)的數(shù)據(jù)新鮮度，我們計(jì)劃探索 Apache Hudi 的 Merge-On-Read 格式，以進(jìn)一步降低數(shù)據(jù)新鮮度。?流式數(shù)據(jù)湖：Apache Hudi 提供增量處理能力，就像數(shù)據(jù)庫(kù)變更日志一樣，我們未來(lái)的工作涉及使用這種原語(yǔ)并構(gòu)建端到端流管道以有效地將更改滲透到下游表，這也將使我們能夠以實(shí)時(shí)流媒體的方式執(zhí)行隱私保護(hù)操作，例如屏蔽和匿名化。?用于服務(wù)間數(shù)據(jù)交換的 CDC 服務(wù)：CDC 已在 Robinhood 中用于為數(shù)據(jù)湖的增量攝取提供更改流，我們正在研究使用 CDC 流在各種在線(xiàn)微服務(wù)之間進(jìn)行可靠的數(shù)據(jù)交換。?數(shù)據(jù)計(jì)算：我們一直致力于提高基于 Apache Spark 和 Trino 構(gòu)建的數(shù)據(jù)計(jì)算平臺(tái)的可用性、效率和性能，以支持關(guān)鍵數(shù)據(jù)計(jì)算工作負(fù)載。
這些是在 Robinhood 數(shù)據(jù)基礎(chǔ)設(shè)施團(tuán)隊(duì)工作的激動(dòng)人心的時(shí)刻，因?yàn)槲覀円呀?jīng)開(kāi)始構(gòu)建下一代 Robinhood 數(shù)據(jù)湖。
引用鏈接
[1]?最初的數(shù)據(jù)湖版本:?[https://robinhood.engineering/data-lake-at-robinhood-3e9cdf963368](https://robinhood.engineering/data-lake-at-robinhood-3e9cdf963368)