Avro与Debezium-Hudi集成：嵌套Value字段处理及Schema配置问询

纯Avro问题前置 #

是否存在兼容如下消息的Avro Schema，其中before和after字段必须为相同记录类型：

{
  "before": null,
  "after": {
    "id": 1,
    "name": "Bob"
  },
  "op": "c"
}

当定义如下Schema时：

{
  "type": "record",
  "name": "Envelope",
  "fields": [
    {
      "name": "before",
      "default": null,
      "type": [
        "null",
        {
          "type": "record",
          "name": "Value",
          "fields": [
            {
              "name": "id",
              "type": "int"
            },
            {
              "name": "name",
              "type": "string"
            }
          ]
        }
      ]
    },
    {
      "name": "after",
      "default": null,
      "type": [
        "null",
        "Value"
      ]
    },
    {
      "name": "op",
      "type": "string"
    }
  ]
}

该Schema要求额外嵌套层，仅接受如下格式的消息：

{
  "after": {
    "Value": {
      "id": 10,
      "name": "Bob"
    }
  },
  "before": null,
  "op": "c"
}

查阅Avro文档及示例后得知这是预期/正常行为，为何需要额外的Value对象，而非将字段直接置于after或before字段的根层级？

额外上下文 #

使用Debezium-Postgres Kafka连接器及Confluent's Schema Registry完成以下操作：

1. 捕获上游Postgres数据库的变更日志事件
2. 将事件Avro序列化后发布至Kafka主题
3. 通过Hudi's DeltaStreamer摄取该Kafka主题，将其存储/索引至S3等Blob存储作为数据湖

第3步中，DeltaStreamer期望字段位于after或before字段的根层级，但实际接收到的是after.Value.<...>或before.Value.<...>格式的数据。

核心问题 #

1. 在纯Avro中，是否可实现支持before.id而非before.Value.id的Schema？
2. 若可以，如何配置Confluent的AvroConverter以自动推断Schema并按此序列化变更日志事件？

更新1 #

基于回复得知，此行为源于Avro处理联合类型的方式：当为字段指定联合类型时，Avro会用Schema名称作为键的JSON对象包裹实际值，以此确定每个数据项使用的联合类型。

疑惑的是，处理PostgreSQL => Debezium => Kafka => Apache Hudi流程的用户似乎未遇到此问题，怀疑Hudi或许能处理嵌套Value的消息。

由于这些是数据库变更日志事件，存在after = NULL（DELETE操作）和before = NULL（INSERT操作）的情况，是否存在单一SMT配置/组合可将如下消息：

// 消息1
{
  "after": {
    "Value": {
      "id": 10,
      "name": "Bob"
    }
  },
  "before": null,
  "op": "c"
}

// 消息2
{
  "after": null,
  "before": {
    "Value": {
      "id": 10,
      "name": "Bob"
    }
  },
  "op": "d"
}

转换为：

// 消息1
{
  "after": {
    "id": 10,
    "name": "Bob"
  },
  "before": null,
  "op": "c"
}

// 消息2
{
  "after": null,
  "before": {
    "id": 10,
    "name": "Bob"
  },
  "op": "d"
}

若简单SMT配置无法处理此解嵌套操作，或许在Hudi阶段实现转换更合理，但理想状态是无需额外转换器仅用于解包Value对象。

问题1：纯Avro中能否实现支持before.id的Schema？ #

可以实现。Avro的JSON编码分为严格模式和非严格模式：当联合类型的第一个元素是null时，非严格模式下可以省略类型包装，直接将记录值放在before/after字段的根层级。

你当前定义的Schema本身是正确的，关键在于使用支持非严格JSON编码的Avro工具链，就能生成符合预期的消息格式。

问题2：配置Confluent AvroConverter实现预期序列化 #

要让Confluent的AvroConverter输出无Value包装的格式，只需在连接器配置中添加非严格编码开关：

key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://your-schema-registry:8081
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://your-schema-registry:8081
# 启用非严格JSON编码，自动省略联合类型的包装层
value.converter.avro.strict=false

该配置会让AvroConverter在序列化时，当before/after字段非空且联合类型首元素是null时，直接输出原始记录结构，而非包装在Value键下。

SMT解嵌套方案 #

如果无法修改AvroConverter配置，可以使用Kafka Connect的ExtractField$Value SMT组合处理，配置示例如下：

transforms=unwrapAfter,unwrapBefore
transforms.unwrapAfter.type=org.apache.kafka.connect.transforms.ExtractField$Value
transforms.unwrapAfter.field=after.Value
transforms.unwrapAfter.ignore.null=true
transforms.unwrapBefore.type=org.apache.kafka.connect.transforms.ExtractField$Value
transforms.unwrapBefore.field=before.Value
transforms.unwrapBefore.ignore.null=true

此配置会自动提取after.Value和before.Value的内容，同时忽略null值避免报错，完全匹配你需要的消息格式。

Hudi侧处理方案 #

Hudi对Debezium变更日志有原生支持，部分版本会自动处理Value嵌套层。如果你的版本不支持，可以通过两种方式解决：

1. 自定义转换器：配置hoodie.deltastreamer.transformer.class，编写简单的Java/Scala类解嵌套Value层；
2. Spark SQL预处理：通过创建视图提前解嵌套，再将视图作为DeltaStreamer的数据源：

CREATE VIEW cleaned_events AS
SELECT 
  CASE WHEN after.Value IS NOT NULL THEN after.Value ELSE NULL END AS after,
  CASE WHEN before.Value IS NOT NULL THEN before.Value ELSE NULL END AS before,
  op
FROM kafka_topic

为何其他用户未遇到此问题？ #

多数用户会直接使用Debezium官方推荐的Avro配置，其中默认可能已启用非严格编码；或者Hudi的内置Debezium适配逻辑会自动处理Value嵌套，无需额外配置。

内容的提问来源于stack exchange，提问作者samser