使用spark递增聚合hudi表值

Question

我有一个火花流作业，每10秒加载一次apache hudi表中的数据。如果该行已经存在，它将更新hudi表中的行。实际上，它正在执行重新插入操作。

但是，在hudi表中，有一个an列，该列也使用新值进行更新。

for example
 1 batch, id=1, amount value=10. --> in table, amount value = 10
 2 batch, id=1, amount value=20. --> in table, amount value = 20

但我需要的是金额值，如果是30而不是20。我需要递增地聚合“金额”列。

hudi是否支持增量聚合，不使用外部缓存/db？

Answer 1

Apache默认使用类org.apache.hudi.common.model.OverwriteWithLatestAvroPayload预组合数据记录并重新插入旧的存储记录，这只需检查您的数据same是否包含具有最大ordering字段的记录，然后用从插入的数据中选择的新记录替换旧的存储记录。

但是，您可以通过实现接口org.apache.hudi.common.model.HoodieRecordPayload并将配置hoodie.compaction.payload.class设置为您的类来创建自己的记录有效负载类。(这里更倾诉)