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實現