大資料技術與架構
大資料真好玩
本文将介紹Apache Hudi的基本概念、設計以及總體基礎架構。
Apache Hudi(簡稱：Hudi)允許您在現有的hadoop相容存儲之上存儲大量資料，同時提供兩種原語，使得除了經典的批處理之外，還可以在資料湖上進行流處理。
這兩種原語分别是:
1）Update/Delete記錄：Hudi使用細粒度的檔案/記錄級别索引來支援Update/Delete記錄，同時還提供寫操作的事務保證。查詢會處理最後一個送出的快照，并基于此輸出結果。
2）變更流：Hudi對擷取資料變更提供了一流的支援：可以從給定的時間點擷取給定表中已updated/inserted/deleted的所有記錄的增量流，并解鎖新的查詢姿勢（類别）。
這些原語緊密結合，解鎖了基于DFS抽象的流/增量處理能力。如果您熟悉流處理，那麼這和從kafka主題消費事件，然後使用狀态存儲逐漸累加中間結果類似。
在架構上會有以下幾點優勢：
1）效率的提升：攝取資料通常需要處理更新、删除以及強制唯一鍵限制。然而，由于缺乏像Hudi這樣能對這些功能提供标準支援的系統，資料工程師們通常會采用大批量的作業來重新處理一整天的事件，或者每次運作都重新加載整個上遊資料庫，進而導緻大量的計算資源浪費。由于Hudi支援記錄級更新，它通過隻處理有變更的記錄并且隻重寫表中已更新/删除的部分，而不是重寫整個表分區甚至整個表，為這些操作帶來一個數量級的性能提升。
2）更快的ETL/派生Pipelines：從外部系統攝入資料後，下一步需要使用Apache Spark/Apache Hive或者任何其他資料處理架構來ETL這些資料用于諸如資料倉庫、機器學習或者僅僅是資料分析等一些應用場景。通常，這些處理再次依賴以代碼或SQL表示的批處理作業，這些作業将批量處理所有輸入資料并重新計算所有輸出結果。通過使用增量查詢而不是快照查詢來查詢一個或多個輸入表，可以大大加速此類資料管道，進而再次導緻像上面一樣僅處理來自上遊表的增量更改，然後upsert或者delete目标派生表。
3）擷取新鮮資料：減少資源還能擷取性能上的提升并不是常見的事。畢竟我們通常會使用更多的資源（例如記憶體）來提升性能（例如查詢延遲）。通過從根本上擺脫資料集的傳統管理方式，Hudi将批量處理增量化的一個很好的副作用是：與以前的資料湖相比，pipeline運作的時間會更短，資料傳遞會更快。
4）統一存儲：基于以上三個優點，在現有資料湖之上進行更快速、更輕量的處理意味着僅出于通路近實時資料的目的時不再需要專門的存儲或資料集市。
Hudi是從零設計的，用于從大型資料集輸入和輸出資料，并借鑒了資料庫設計的原理。為此，Hudi提供了索引實作，可以将記錄的鍵快速映射到其所在的檔案位置。同樣，對于流式輸出資料，Hudi通過其特殊列添加并跟蹤記錄級别的中繼資料，進而可以提供所有發生變更的精确增量流。
Hudi注意到使用者可能對資料新鮮度（寫友好）與查詢性能（讀/查詢友好）有不同的期望，并支援了三種查詢類型，這些類型提供實時快照，增量流以及稍早的純列資料。在每一步，Hudi都努力做到自我管理（例如自動優化編寫程式的并行性，保持檔案大小）和自我修複（例如：自動復原失敗的送出），即使這樣做會稍微增加運作時成本（例如：在記憶體中緩存輸入資料已分析工作負載）。如果沒有這些内置的操作杠杆/自我管理功能，這些大型流水線的營運成本通常會翻倍。
Hudi還具有 append only、雲資料友好的設計，該設計使Hudi無縫管理所有雲提供商傷的資料，并實作了日志結構化存儲系統的原理。
在寫方面，Hudi表被模組化為鍵值對資料集，其中每條記錄都有一個唯一的記錄鍵。此外，一個記錄鍵還可以包括分區路徑，在該路徑下，可以對記錄進行分區和存儲。這通常有助于減少索引查詢的搜尋空間。
了解了Hudi項目的關鍵技術動機後，現在讓我們更深入地研究Hudi系統本身的設計。在較高的層次上，用于寫Hudi表的元件使用了一種受支援的方式嵌入到Apache Spark作業中，它會在支援DFS的存儲上生成代表Hudi表的一組檔案。然後，在具有一定保證的情況下，諸如Apache Spark、Presto、Apache Hive之類的查詢引擎可以查詢該表。
Hudi表的三個主要元件：
1）有序的時間軸中繼資料。類似于資料庫事務日志。
2）分層布局的資料檔案：實際寫入表中的資料。
3）索引（多種實作方式）：映射包含指定記錄的資料集。
Hudi提供了以下功能來對基礎資料進行寫入、查詢，這使其成為大型資料湖的重要子產品：
1）支援快速，可插拔索引的upsert();
2）高效、隻掃描新資料的增量查詢；
3）原子性的資料釋出和復原，支援恢複的Savepoint；
4）使用mvcc風格設計的讀和寫快照隔離；
5）使用統計資訊管理檔案大小；
6）已有記錄update/delta的自管理壓縮；
7）稽核資料修改的時間軸中繼資料；
8）滿足GDPR(通用資料保護條例)、資料删除功能。
在其核心，Hudi維護了一條包含在不同的即時時間（instant time）對資料集做的所有instant操作的timeline，進而提供表的即時視圖，同時還有效的支援按到達順序進行資料檢索。時間軸類似于資料庫的redo/transaction日志，由一組時間軸執行個體組成。Hudi保證在時間軸上執行的操作的原子性和基于即時時間的時間軸一緻性。時間軸被實作為表基礎路徑下.hoodie中繼資料檔案夾下的一組檔案。具體來說，最新的instant被儲存為單個檔案，而較舊的instant被存檔到時間軸歸檔檔案夾中，以限制writers和queries列出的檔案數量。
一個Hudi 時間軸instant由下面幾個元件構成：
1）操作類型：對資料集執行的操作類型；
2）即時時間：即時時間通常是一個時間戳(例如：20190117010349)，該時間戳按操作開始時間的順序單調增加；
3）即時狀态：instant的目前狀态;
每個instant都有avro或者json格式的中繼資料資訊，詳細的描述了該操作的狀态以及這個即時時刻instant的狀态。
關鍵的Instant操作類型有：
1）COMMIT：一次送出表示将一組記錄原子寫入到資料集中；
2）CLEAN: 删除資料集中不再需要的舊檔案版本的背景活動;
3）DELTA_COMMIT:将一批記錄原子寫入到MergeOnRead存儲類型的資料集中，其中一些/所有資料都可以隻寫到增量日志中;
4）COMPACTION: 協調Hudi中差異資料結構的背景活動，例如：将更新從基于行的日志檔案變成列格式。在内部，壓縮表現為時間軸上的特殊送出;
5）ROLLBACK: 表示送出/增量送出不成功且已復原，删除在寫入過程中産生的所有部分檔案;
6）SAVEPOINT: 将某些檔案組标記為"已儲存"，以便清理程式不會将其删除。在發生災難/資料恢複的情況下，它有助于将資料集還原到時間軸上的某個點;
任何給定的即時都會處于以下狀态之一：
1）REQUESTED:表示已排程但尚未初始化；
2）INFLIGHT: 表示目前正在執行該操作;
3）COMPLETED: 表示在時間軸上完成了該操作.
Hudi将表組織成DFS上基本路徑下的檔案夾結構中。如果表是分區的，則在基本路徑下還會有其他的分區，這些分區是包含該分區資料的檔案夾，與Hive表非常類似。每個分區均由相對于基本路徑的分區路徑唯一辨別。在每個分區内，檔案被組織成檔案組，由檔案ID唯一辨別。每個檔案組包含一個或多個檔案片，每個檔案片都包含一個base file（某個送出/壓縮即時時間生成的列式存儲檔案，例如：parquet檔案）以及一組日志檔案（包含自生成基本檔案以來對基本檔案的插入/更新）。Hudi采用了MVCC設計，壓縮操作會将日志和基本檔案合并以産生新的檔案片，而清理操作則将未使用的/較舊的檔案片删除以回收DFS上的空間。
Hudi通過索引機制提供高效的upsert操作，該機制會将一個記錄鍵+分區路徑組合一緻性的映射到一個檔案ID.這個記錄鍵和檔案組/檔案ID之間的映射自記錄被寫入檔案組開始就不會再改變。簡而言之，這個映射檔案組包含了一組檔案的所有版本。Hudi目前提供了3種索引實作（HBaseIndex,、HoodieBloomIndex（HoodieGlobalBloomIndex）、InMemoryHashIndex）來映射一個記錄鍵到包含該記錄的檔案ID。這将使我們無需掃描表中的每條記錄，就可顯著提高upsert速度。
Hudi索引可以根據其查詢分區記錄的能力進行分類：
1）全局索引：不需要分區資訊即可查詢記錄鍵映射的檔案ID。比如，寫程式可以傳入null或者任何字元串作為分區路徑（partitionPath）,但索引仍然會查找到該記錄的位置。全局索引在記錄鍵在整張表中保證唯一的情況下非常有用，但是查詢的消耗随着表的大小函數式增加。
2）非全局索引：與全局索引不同，非全局索引依賴分區路徑(partitionPath),對于給定的記錄鍵，它隻會在給定分區路徑下查找該記錄。這比較适合總是同時生成分區路徑和記錄鍵的場景，同時還能享受到更好的擴充性，因為查詢索引的消耗隻與寫入到該分區下資料集有關系。
COW表寫的時候資料直接寫入basefile,（parquet）不寫log檔案。是以COW表的檔案片隻包含basefile(一個parquet檔案構成一個檔案片)。
這種的存儲方式的Spark DAG相對簡單。關鍵目标是是使用partitioner将tagged Hudi記錄RDD（所謂的tagged是指已經通過索引查詢，标記每條輸入記錄在表中的位置）分成一些列的updates和inserts.為了維護檔案大小，我們先對輸入進行采樣，獲得一個工作負載profile,這個profile記錄了輸入記錄的insert和update、以及在分區中的分布等資訊。把資料重新打包：
1）對于updates, 該檔案ID的最新版本都将被重寫一次，并對所有已更改的記錄使用新值；
2）對于inserts.記錄首先打包到每個分區路徑中的最小檔案中，直到達到配置的最大大小。之後的所有剩餘記錄将再次打包到新的檔案組，新的檔案組也會滿足最大檔案大小要求。
MOR表寫資料時，記錄首先會被快速的寫進日志檔案，稍後會使用時間軸上的壓縮操作将其與基礎檔案合并。根據查詢是讀取日志中的合并快照流還是變更流，還是僅讀取未合并的基礎檔案，MOR表支援多種查詢類型。
在高層次上，MOR writer在讀取資料時會經曆與COW writer 相同的階段。這些更新将追加到最新檔案篇的最新日志檔案中，而不會合并。對于insert,Hudi支援兩種模式：
1）插入到日志檔案：有可索引日志檔案的表會執行此操作（HBase索引）；
2）插入parquet檔案：沒有索引檔案的表（例如布隆索引）
與寫時複制（COW）一樣，對已标記位置的輸入記錄進行分區，以便将所有發往相同檔案id的upserts分到一組。這批upsert會作為一個或多個日志塊寫入日志檔案。Hudi允許用戶端控制日志檔案大小。對于寫時複制（COW）和讀時合并（MOR）writer來說，Hudi的WriteClient是相同的。幾輪資料的寫入将會累積一個或多個日志檔案。這些日志檔案與基本的parquet檔案（如有）一起構成一個檔案片，而這個檔案片代表該檔案的一個完整版本。
這種表是用途最廣、最進階的表。為寫（可以指定不同的壓縮政策，吸收突發寫流量）和查詢（例如權衡資料的新鮮度和查詢性能）提供了很大的靈活性。同時它包含一個學習曲線，以便在操作上掌控他。
了解Hudi資料源或者deltastreamer工具提供的3種不同寫操作以及如何最好的利用他們可能會有所幫助。這些操作可以在對資料集發出的每個commit/delta commit中進行選擇/更改。
1）upsert操作：這是預設操作，在該操作中，首先通過查詢索引将資料記錄标記為插入或更新，然後再運作試探法确定如何最好地将他們打包到存儲，以對檔案大小進行優化，最終将記錄寫入。對于諸如資料庫更改捕獲之類的用例，建議在輸入幾乎肯定包含更新的情況下使用此操作。
2）insert操作：與upsert相比，insert操作也會運作試探法确定打包方式，優化檔案大小，但會完全跳過索引查詢。是以對于諸如日志重複資料删除（結合下面提到的過濾重複項選項）的用例而言，它比upsert的速度快得多。這也适用于資料集可以容忍重複項，但隻需要Hudi具有事務性寫/增量拉取/存儲管理功能的用例。
3）bulk insert操作：upsert 和insert操作都會将輸入記錄保留在記憶體中，以賈逵愛存儲啟發式計算速度，是以對于最初加載/引導Hudi資料集的用例而言可能會很麻煩。Bulk insert提供了與insert相同的語義，同時實作了基于排序的資料寫入算法，該算法可以很好的擴充數百TB的初始負載。但是這隻是在調整檔案大小方面進行的最大努力，而不是像insert/update那樣保證檔案大小。
壓縮是一個 instant操作，它将一組檔案片作為輸入，将每個檔案切片中的所有日志檔案與其basefile檔案（parquet檔案）合并，以生成新的壓縮檔案片，并寫為時間軸上的一個commit。壓縮僅适用于讀時合并（MOR）表類型，并且由壓縮政策（預設選擇具有最大未壓縮日志的檔案片）決定選擇要進行壓縮的檔案片。這個壓縮政策會在每個寫操作之後評估。
從高層次上講，壓縮有兩種方式：
1）同步壓縮：這裡的壓縮由寫程式程序本身在每次寫入之後同步執行的，即直到壓縮完成後才能開始下一個寫操作。就操作而言，這個是最簡單的，因為無需安排單獨的壓縮過程，但保證的資料新鮮度最低。不過，如果可以在每次寫操作中壓縮最新的表分區，同時又能延遲遲到/較舊分區的壓縮，這種方式仍然非常有用。
2）異步壓縮：使用這種方式，壓縮過程可以與表的寫操作同時異步運作。這樣具有明顯的好處，即壓縮不會阻塞下一批資料寫入，進而産生近實時的資料新鮮度。Hudi DeltaStreamer之類的工具支援邊界的連續模式，其中的壓縮和寫入操作就是以這種方式在單個Spark運作時叢集中進行的。
清理是一項基本的即時操作，其執行的目的時删除舊的檔案片，并限制表占用的存儲空間。清理會在每次寫操作之後自動執行，并利用時間軸伺服器上緩存的時間軸中繼資料來避免掃描整個表來評估清理時機。
Hudi支援兩種清理方式：
1）按commits / deltacommits清理：這是增量查詢中最常見且必須使用的模式。以這種方式，Cleaner會保留最近N次commit/delta commit送出中寫入的所有檔案切片，進而有效提供在任何即時範圍内進行增量查詢的能力。盡管這對于增量查詢很有幫助，但由于保留了配置範圍内所有版本的檔案片，是以，在某些高寫入負載的場景下可能需要更大的存儲空間。
2）按保留的檔案片清理：這是一種更為簡單的清理方式，這裡我們僅儲存每個檔案組中的最後N個檔案片。諸如Apache Hive之類的某些查詢引擎會處理非常大的查詢，這些查詢可能需要幾個小時才能完成，在這種情況下，将N設定為足夠大以至于不會删除查詢仍然可以通路的檔案片是很有用的。
此外，清理操作會保證每個檔案組下面會一直隻有一個檔案片（最新的一片）。
Hudi還對表中存儲的資料執行了幾種秘鑰存儲管理功能。在DFS上存儲資料的關鍵是管理檔案大小和技術以及回收存儲空間。例如，HDFS在處理小檔案問題上臭名昭著--在NameNode上施加記憶體/RPC壓力，可能破壞整個叢集的穩定性。通常，查詢引擎可在适當大小的列檔案上提供更好的性能，因為它們可以有效地攤銷擷取列統計資訊等的成本。即使在某些雲資料存儲上，列出包含大量小檔案的目錄也會産生成本。
下面是一些Hudi高效寫，管理資料存儲的方法：
1）小檔案處理特性會剖析輸入的工作負載，并将内容配置設定到現有的檔案組，而不是建立新檔案組（這會導緻生成小檔案）。
2）在writer中使用一個時間軸緩存，這樣隻要Spark叢集不每次都重新開機，後續的寫操作就不需要列出DFS目錄來擷取指定分區路徑下的檔案片清單。
3）使用者還可以調整基本檔案和日志檔案大小之間的比值系數以及期望的壓縮率，以便将足夠數量的insert分到統一檔案組，進而生成大小合适的基本檔案。
4）智能調整bulk insert并行度，可以再次調整大小合适的初始檔案組。實際上，正确執行此操作非常關鍵，因為檔案組一旦建立就不能被删除，而智能如前面所述對其進行擴充。
鑒于這種靈活而全面的資料布局和豐富的時間線，Hudi能夠支援三種不同的查詢表方式，具體取決于表的類型。
可檢視給定delta commit或者commit即時操作後表的最新快照。在讀時合并（MOR）表的情況下，它通過即時合并最新檔案片的基本檔案和增量檔案來提供近實時表（幾分鐘）。對于寫時複制（COW），它可以替代現有的parquet表（或相同基本檔案類型的表），同時提供upsert/delete和其他寫入方面的功能。
可檢視自給定commit/delta commit即時操作以來新寫入的資料。有效的提供變更流來啟用增量資料管道。
可檢視給定的commit/compact即時操作的表的最新快照。僅将最新檔案片的基本/列檔案暴露給查詢，并保證與非Hudi表相同的列查詢性能。
版權聲明：
本文為大資料技術與架構整理，原作者獨家授權。未經原作者允許轉載追究侵權責任。
編輯｜冷眼丶
微信公衆号｜import_bigdata