醫療線上OLAP場景下基於Apache Hudi 模式演變的改造與應用
背景
在 Apache Hudi支援完整的Schema演變的方案中（https://mp.weixin.qq.com/s/rSW864o2YEbHw6oQ4Lsq0Q）， 讀取方面，只完成了SQL on Spark的支援（Spark3以上，用於離線分析場景），Presto（用於線上OLAP場景）及Apache Hive（Hudi的bundle包）的支援，在正式釋出版本中（Hudi 0.12.1, PrestoDB 0.277）還未支援。在當前的醫療場景下，Schema變更發生次數較多，且經常使用Presto讀取Hudi資料進行線上OLAP分析，在讀到Schema變更過的表時很可能會產生錯誤結果，造成不可預知的損失，所以必須完善Presto在讀取方面對Schema完整演變的支援。
另外使用者對使用presto對Hudi讀取的實時性要求較高，之前的方案裡Presto只支援Hudi的讀優化方式讀取。讀優化的情況下，由於預設的布隆索引有如下行為：
- insert 操作的資料，每次寫入提交後能夠查詢到；
- update，delete操作的資料必須在發生資料合併後才能讀取到；
- insert與（update，delete）操作 presto 能夠查詢到的時間不一致；
- 所以必須增加presto對hudi的快照查詢支援。
由於Presto分為兩個分支（Trino和PrestoDB），其中PrestoDB的正式版本已經支援快照查詢模式，而Trino主線還不存在這個功能，所以優先考慮在PrestoDB上實現，我們基於Trino的方案也在開發中。
計劃基於Prestodb的Presto-Hudi模組改造，設計自 RFC-44: Hudi Connector for Presto。單獨的Hudi聯結器可以拋開當前程式碼的限制，高效地進行特定優化、新增新功能、整合高階功能並隨著上游專案快速發展。
術語說明
-
read_optimized（讀優化）：COW表和MOR表的ro表，只讀取parquet檔案的查詢模式
-
snapshot（快照）：MOR表的rt表，讀取log檔案和parquet並計算合併結果的查詢模式
現狀：
Hudi的Schema演變過程中多種引擎的表現
其中trino是以官方360版本為基礎開發的本地版本，部分參考某開啟狀態的pr，使其支援了快照查詢
Hive對Hudi支援的情況
hive使用hudi提供的hudi-hadoop-mr模組的InputFormat介面，支援完整schema的功能在10月28日合入Hudi主線。
Trino對Hudi支援的情況
Trino版本主線分支無法用快照模式查詢。Hudi聯結器最終於22年9月28日合入主線，仍沒有快照查詢的功能。本地版本基於trino360主動合入社群中開啟狀態的pr（Hudi MOR changes），基於hive聯結器完成了快照查詢能力。
PrestoDB對Hudi支援的情況
PrestoDB版本主線分支支援Hudi聯結器，本身沒有按列位置獲取列值的功能，所以沒有串列問題，並且支援快照查詢模式。
改造方案
版本
-
Hudi: 0.12.1
-
Presto: 0.275
該模組的設計如下
讀優化
Presto 會使用它自己優化的方式讀parquet檔案。在presto-hudi的HudiPageSourceProvider -> HudiParquetPageSources -> 最終使用presto-parquet 的 ParquetReader讀取
快照
Presto 針對mor表的快照讀，會使用hudi提供的huid-hadoop-mr的InputFormat介面。在presto-hudi的HudiPageSourceProvider -> HudiRecordCursors裡建立 HoodieParquetRealtimeInputFormat -> 獲取RealtimeCompactedRecordReader，基礎檔案使用HoodieParquetInputFormat的getRecordReader，日誌檔案使用HoodieMergedLogRecordScanner掃描
讀優化的改造
基本思想：在presto-hudi模組的HudiParquetPageSources中，獲取檔案和查詢的 InternalSchema ，merge後與presto裡的schema列資訊轉換，進行查詢。
具體步驟：
- 使用TableSchemaResolver的getTableInternalSchemaFromCommitMetadata方法獲取最新的完整InternalSchema
- 使用HudiParquetPageSources類的createParquetPageSource方法傳入引數regularColumns（List
），與完整InternalSchema通過InternalSchemaUtils.pruneInternalSchema方法獲取剪枝後的InternalSchema - 通過FSUtils.getCommitTime方法利用檔名的時間戳獲取commitInstantTime，再利用InternalSchemaCache.getInternalSchemaByVersionId方法獲取檔案的InternalSchema
- 使用InternalSchemaMerger的mergeSchema方法，獲取剪枝後的查詢InternalSchema和檔案InternalSchema進行merge的InternalSchema
- 使用merge後的InternalSchema的列名list，轉換為HudiParquetPageSources的requestedSchema，改變HudiParquetPageSources的getDescriptors和getColumnIO等方法邏輯的結果
實現為 https://github.com/prestodb/presto/pull/18557 （開啟狀態）
快照的改造
基本思想：改造huid-hadoop-mr模組的InputFormat，獲取資料和查詢的 InternalSchema ，將merge後的schema列資訊設定為hive任務所需的屬性，進行查詢。
具體步驟：
1.基礎檔案支援完整schema演變，spark-sql的實現此處無法複用，新增轉換類，在HoodieParquetInputFormat中使用轉換類，根據commit獲取檔案schema，根據查詢schema和檔案schema進行merge，將列名和屬性設定到job的屬性裡serdeConstants.LIST_COLUMNS，ColumnProjectionUtils.READ_COLUMN_NAMES_CONF_STR，serdeConstants.LIST_COLUMN_TYPES；
2.日誌檔案支援完整schema演變，spark-sql的實現此處可以複用。HoodieParquetRealtimeInputFormat的RealtimeCompactedRecordReader中，使用轉換類設定reader物件的幾個schema屬性，使其複用現有的merge資料schema與查詢schema的邏輯。
已經存在pr可以達到目標 https://github.com/apache/hudi/pull/6989 （合入master，0.13）
Presto的配置
${presto_home}/etc/catalog/hudi.properties，基本複製hive.properties；主要修改為
connector.name=hudi
Presto的部署
此處分別為基於hudi0.12.1和prestodb的release0.275合入pr後打的包，改動涉及檔案不同版本間差異不大，無需關注版本問題
分別將mor表改造涉及的包：
hudi-presto-bundle-0.12.1.jar
以及cow表改造涉及的包：
presto-hudi-0.275.1-SNAPSHOT.jar
放入${presto_home}/etc/catalog/hudi.propertiesplugin/hudi
重啟presto服務
開發過程遇到的問題及解決
總結
當前已經實現PrestoDB對Hudi的快照讀，以及對schema完整演變的支援，滿足了大批量表以MOR的表格式快速寫入資料湖，且頻繁變更表結構的同時，能夠準確實時地進行OLAP分析的功能。但由於Trino社群更加活躍，以前的很多功能基於Trino開發，下一步計劃改造Trino，使其完整支援快照讀與兩種查詢模式下的schema完整演變。