Apache Hudi 的Payload是一種可擴展的數據處理機制，通過不同的Payload我們可以實現複雜場景的定製化數據寫入方式，大大增加了數據處理的靈活性。Hudi Payload在寫入和讀取Hudi表時對數據進行去重、過濾、合併等操作的工具類，通過使用參數 "hoodie.datasource.write.payload.class"指定我們需要使用的Payload class。
1.摘要
Apache Hudi 的Payload是一種可擴展的數據處理機制，通過不同的Payload我們可以實現複雜場景的定製化數據寫入方式，大大增加了數據處理的靈活性。Hudi Payload在寫入和讀取Hudi表時對數據進行去重、過濾、合併等操作的工具類，通過使用參數 "hoodie.datasource.write.payload.class"指定我們需要使用的Payload class。本文我們會深入探討Hudi Payload的機制和不同Payload的區別及使用場景。
2. 為何需要Payload
在數據寫入的時候，現有整行插入、整行覆蓋的方式無法滿足所有場景要求，寫入的數據也會有一些定製化處理需求，因此需要有更加靈活的寫入方式以及對寫入數據進行一定的處理，Hudi提供的playload方式可以很好的解決該問題，例如可以解決寫入時數據去重問題，針對部分字段進行更新等等。
3. Payload的作用機制
寫入Hudi表時需要指定一個參數hoodie.datasource.write.precombine.field，這個字段也稱為Precombine Key，Hudi Payload就是根據這個指定的字段來處理數據，它將每條數據都構建成一個Payload，因此數據間的比較就變成了Payload之間的比較。只需要根據業務需求實現Payload的比較方法，即可實現對數據的處理。
Hudi所有Payload都實現HoodieRecordPayload接口，下面列出了所有實現該接口的預置Payload類。
下圖列舉了HoodieRecordPayload接口需要實現的方法，這裡有兩個重要的方法preCombine和combineAndGetUpdateValue，下面我們對這兩個方法進行分析。
3.1 preCombine分析
從下圖可以看出，該方法比較當前數據和oldValue，然後返回一條記錄。
從preCombine方法的注釋描述也可以知道首先它在多條相同主鍵的數據同時寫入Hudi時，用來進行數據去重。調用位置
其實該方法還有另一個調用的地方，即在MOR表讀取時會對Log file中的相同主鍵的數據進行處理。如果同一條數據多次修改並寫入了MOR表的Log文件，在讀取時也會進行preCombine。
3.2 combineAndGetUpdateValue分析
該方法將currentValue（即現有parquet文件中的數據）與新數據進行對比，判斷是否需要持久化新數據。
由於COW表和MOR表的讀寫原理差異，因此combineAndGetUpdateValue的調用在COW和MOR中也有所不同：
•在COW寫入時會將新寫入的數據與Hudi表中存的currentValue進行比較，返回需要持久化的數據•在MOR讀取時會將經過preCombine處理的Log中的數據與Parquet文件中的數據進行比較，返回需要持久化的數據
4.常用Payload處理邏輯的對比
了解了Payload的內核原理，下面我們對比分析下集中常用的Payload實現的方式。
4.1 OverwriteWithLatestAvroPayload
OverwriteWithLatestAvroPayload 的相關方法實現如下
可以看出使用OverwriteWithLatestAvroPayload 會根據orderingVal進行選擇（這裡的orderingVal即precombine key的值），而combineAndGetUpdateValue永遠返回新數據。
4.2 OverwriteNonDefaultsWithLatestAvroPayload
OverwriteNonDefaultsWithLatestAvroPayload繼承OverwriteWithLatestAvroPayload，preCombine方法相同，重寫了combineAndGetUpdateValue方法，新數據會按字段跟schema中的default value進行比較，如果default value非null且與新數據中的值不同時，則在新數據中更新該字段。由於通常schema定義的default value都是null，在此場景下可以實現更新非null字段的功能，即如果一條數據有五個字段，使用此Payload更新三個字段時不會影響另外兩個字段原來的值。
4.3 DefaultHoodieRecordPayload
DefaultHoodieRecordPayload同樣繼承OverwriteWithLatestAvroPayload重寫了combineAndGetUpdateValue方法，通過下面代碼可以看出該Payload使用precombine key對現有數據和新數據進行比較，判斷是否要更新該條數據。
下面我們以COW表為例展示不同Payload讀寫結果測試
5. 測試
我們使用如下幾條源數據，以key為主鍵，col3為preCombine key寫Hudi表。
首先我們一次寫入col0是'aa'、'bb'的兩條數據，由於他們的主鍵相同，所以在precombine時會根據col3比較去重，最終寫入Hudi表的只有一條數據。（注意如果寫入方式是insert或bulk_insert則不會去重）
查詢結果
下面我們使用col0是'cc'的數據進行更新，這是由於三種Payload的處理邏輯不同，最終寫入的數據結果也不同。
OverwriteWithLatestAvroPayload完全用新數據覆蓋了舊數據。
OverwriteNonDefaultsWithLatestAvroPayload由於更新數據中col1 col2為null，因此該字段未被更新。
DefaultHoodieRecordPayload由於cc的col3小於bb的，因此該數據未被更新。
6. 總結
通過上面分析我們清楚了Hudi常用的幾種Payload機制，總結對比如下
雖然Hudi提供了多個預置Payload，但是仍不能滿足一些特殊場景的數據處理工作：例如用戶在使用Kafka-Hudi實時入湖，但是用戶的一條數據的修改不在一條Kafka消息中，而是多條相同主鍵的數據消息到，第一條裡面有col0，col1的數據，第二條有col2，col3的數據，第三條有col4的數據，這時使用Hudi自帶的Payload就無法完成將這三條數據合併之後寫入Hudi表的工作，要實現這個邏輯就要通過自定義Payload，重寫Payload中的preCombine和combineAndGetUpdateValue方法來實現相應的業務邏輯，並在寫入時通過hoodie.datasource.write.payload.class指定我們自定義的Payload實現。
推薦閱讀
一文帶你了解Lakehouse的並發控制：我們是否過於樂觀？
查詢性能提升3倍！Apache Hudi 查詢優化了解下？
來自Apache Hudi PMC Chair的新年大禮包，請注意查收！（附帶2021年精選文章集合）
基於Apache Hudi + Flink的億級數據入湖實踐
OnZoom基於Apache Hudi的流批一體架構實踐
留言列表