1.摘要

Apache Hudi 的Payload是一種可擴(kuò)展的數(shù)據(jù)處理機(jī)制，通過不同的Payload我們可以實(shí)現(xiàn)復(fù)雜場景的定制化數(shù)據(jù)寫入方式，大大增加了數(shù)據(jù)處理的靈活性。Hudi Payload在寫入和讀取Hudi表時(shí)對(duì)數(shù)據(jù)進(jìn)行去重、過濾、合并等操作的工具類，通過使用參數(shù) "hoodie.datasource.write.payload.class"指定我們需要使用的Payload class。本文我們會(huì)深入探討Hudi Payload的機(jī)制和不同Payload的區(qū)別及使用場景。

2. 為何需要Payload

在數(shù)據(jù)寫入的時(shí)候，現(xiàn)有整行插入、整行覆蓋的方式無法滿足所有場景要求，寫入的數(shù)據(jù)也會(huì)有一些定制化處理需求，因此需要有更加靈活的寫入方式以及對(duì)寫入數(shù)據(jù)進(jìn)行一定的處理，Hudi提供的playload方式可以很好的解決該問題，例如可以解決寫入時(shí)數(shù)據(jù)去重問題，針對(duì)部分字段進(jìn)行更新等等。

3. Payload的作用機(jī)制

寫入Hudi表時(shí)需要指定一個(gè)參數(shù)hoodie.datasource.write.precombine.field，這個(gè)字段也稱為Precombine Key，Hudi Payload就是根據(jù)這個(gè)指定的字段來處理數(shù)據(jù)，它將每條數(shù)據(jù)都構(gòu)建成一個(gè)Payload，因此數(shù)據(jù)間的比較就變成了Payload之間的比較。只需要根據(jù)業(yè)務(wù)需求實(shí)現(xiàn)Payload的比較方法，即可實(shí)現(xiàn)對(duì)數(shù)據(jù)的處理。

Hudi所有Payload都實(shí)現(xiàn)HoodieRecordPayload接口，下面列出了所有實(shí)現(xiàn)該接口的預(yù)置Payload類。

下圖列舉了HoodieRecordPayload接口需要實(shí)現(xiàn)的方法，這里有兩個(gè)重要的方法preCombine和combineAndGetUpdateValue，下面我們對(duì)這兩個(gè)方法進(jìn)行分析。

3.1 preCombine分析

從下圖可以看出，該方法比較當(dāng)前數(shù)據(jù)和oldValue，然后返回一條記錄。

從preCombine方法的注釋描述也可以知道首先它在多條相同主鍵的數(shù)據(jù)同時(shí)寫入Hudi時(shí)，用來進(jìn)行數(shù)據(jù)去重。
調(diào)用位置

其實(shí)該方法還有另一個(gè)調(diào)用的地方，即在MOR表讀取時(shí)會(huì)對(duì)Log file中的相同主鍵的數(shù)據(jù)進(jìn)行處理。
如果同一條數(shù)據(jù)多次修改并寫入了MOR表的Log文件，在讀取時(shí)也會(huì)進(jìn)行preCombine。

3.2 combineAndGetUpdateValue分析

該方法將currentValue（即現(xiàn)有parquet文件中的數(shù)據(jù)）與新數(shù)據(jù)進(jìn)行對(duì)比，判斷是否需要持久化新數(shù)據(jù)。

由于COW表和MOR表的讀寫原理差異，因此combineAndGetUpdateValue的調(diào)用在COW和MOR中也有所不同：

在COW寫入時(shí)會(huì)將新寫入的數(shù)據(jù)與Hudi表中存的currentValue進(jìn)行比較，返回需要持久化的數(shù)據(jù)

在MOR讀取時(shí)會(huì)將經(jīng)過preCombine處理的Log中的數(shù)據(jù)與Parquet文件中的數(shù)據(jù)進(jìn)行比較，返回需要持久化的數(shù)據(jù)

4.常用Payload處理邏輯的對(duì)比

了解了Payload的內(nèi)核原理，下面我們對(duì)比分析下集中常用的Payload實(shí)現(xiàn)的方式。

4.1 OverwriteWithLatestAvroPayload

OverwriteWithLatestAvroPayload 的相關(guān)方法實(shí)現(xiàn)如下

可以看出使用OverwriteWithLatestAvroPayload 會(huì)根據(jù)orderingVal進(jìn)行選擇（這里的orderingVal即precombine key的值），而combineAndGetUpdateValue永遠(yuǎn)返回新數(shù)據(jù)。

4.2 OverwriteNonDefaultsWithLatestAvroPayload

OverwriteNonDefaultsWithLatestAvroPayload繼承OverwriteWithLatestAvroPayload，preCombine方法相同，重寫了combineAndGetUpdateValue方法，新數(shù)據(jù)會(huì)按字段跟schema中的default value進(jìn)行比較，如果default value非null且與新數(shù)據(jù)中的值不同時(shí)，則在新數(shù)據(jù)中更新該字段。由于通常schema定義的default value都是null，在此場景下可以實(shí)現(xiàn)更新非null字段的功能，即如果一條數(shù)據(jù)有五個(gè)字段，使用此Payload更新三個(gè)字段時(shí)不會(huì)影響另外兩個(gè)字段原來的值。

4.3 DefaultHoodieRecordPayload

DefaultHoodieRecordPayload同樣繼承OverwriteWithLatestAvroPayload重寫了combineAndGetUpdateValue方法，通過下面代碼可以看出該P(yáng)ayload使用precombine key對(duì)現(xiàn)有數(shù)據(jù)和新數(shù)據(jù)進(jìn)行比較，判斷是否要更新該條數(shù)據(jù)。

下面我們以COW表為例展示不同Payload讀寫結(jié)果測試

5. 測試

我們使用如下幾條源數(shù)據(jù)，以key為主鍵，col3為preCombine key寫Hudi表。

首先我們一次寫入col0是'aa'、'bb'的兩條數(shù)據(jù)，由于他們的主鍵相同，所以在precombine時(shí)會(huì)根據(jù)col3比較去重，最終寫入Hudi表的只有一條數(shù)據(jù)。（注意如果寫入方式是insert或bulk_insert則不會(huì)去重）

查詢結(jié)果

下面我們使用col0是'cc'的數(shù)據(jù)進(jìn)行更新，這是由于三種Payload的處理邏輯不同，最終寫入的數(shù)據(jù)結(jié)果也不同。

OverwriteWithLatestAvroPayload完全用新數(shù)據(jù)覆蓋了舊數(shù)據(jù)。

OverwriteNonDefaultsWithLatestAvroPayload由于更新數(shù)據(jù)中col1 col2為null，因此該字段未被更新。

DefaultHoodieRecordPayload由于cc的col3小于bb的，因此該數(shù)據(jù)未被更新。

6. 總結(jié)

通過上面分析我們清楚了Hudi常用的幾種Payload機(jī)制，總結(jié)對(duì)比如下

雖然Hudi提供了多個(gè)預(yù)置Payload，但是仍不能滿足一些特殊場景的數(shù)據(jù)處理工作：例如用戶在使用Kafka-Hudi實(shí)時(shí)入湖，但是用戶的一條數(shù)據(jù)的修改不在一條Kafka消息中，而是多條相同主鍵的數(shù)據(jù)消息到，第一條里面有col0，col1的數(shù)據(jù)，第二條有col2，col3的數(shù)據(jù)，第三條有col4的數(shù)據(jù)，這時(shí)使用Hudi自帶的Payload就無法完成將這三條數(shù)據(jù)合并之后寫入Hudi表的工作，要實(shí)現(xiàn)這個(gè)邏輯就要通過自定義Payload，重寫Payload中的preCombine和combineAndGetUpdateValue方法來實(shí)現(xiàn)相應(yīng)的業(yè)務(wù)邏輯，并在寫入時(shí)通過hoodie.datasource.write.payload.class指定我們自定義的Payload實(shí)現(xiàn)

以上就是Apache Hudi靈活的Payload機(jī)制硬核解析的詳細(xì)內(nèi)容，更多關(guān)于Apache Hudi Payload機(jī)制解析的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論