如何从Avro记录提取字段子集并使用另一Schema写入S3？

内置SMT组合实现 #

如果你的字段子集与Schema Registry中预定义Schema的字段完全匹配（字段名、类型、顺序一致），可以通过两个内置SMT的组合实现需求：

1. 提取字段子集：使用ReplaceField SMT的include参数过滤保留需要的字段，配置示例：

   transforms=filterFields
   transforms.filterFields.type=org.apache.kafka.connect.transforms.ReplaceField$Value
   transforms.filterFields.include=field_a,field_b,field_c
   

   该配置会保留指定字段，移除其他无关字段。

2. 绑定预定义Schema：使用SetSchemaMetadata SMT指定Schema Registry中预定义Schema的名称和版本，让Avro转换器获取对应Schema进行序列化，配置示例：

   transforms=filterFields,setTargetSchema
   transforms.filterFields.type=org.apache.kafka.connect.transforms.ReplaceField$Value
   transforms.filterFields.include=field_a,field_b,field_c
   transforms.setTargetSchema.type=org.apache.kafka.connect.transforms.SetSchemaMetadata$Value
   transforms.setTargetSchema.schema.name=your-predefined-schema-name
   transforms.setTargetSchema.schema.version=1
   

   需确保预定义Schema的字段结构与ReplaceField处理后的记录结构完全一致，否则会出现序列化失败。

自定义SMT方案 #

如果需求涉及更复杂的逻辑（比如字段重命名、类型转换、动态字段选择、Schema匹配校验等），内置SMT无法满足时，需要编写自定义SMT：

• 实现org.apache.kafka.connect.transforms.ValueTransformer接口，在transform方法中完成：
  1. 从原始记录中提取目标字段子集；
  2. 从Schema Registry中获取预定义的目标Schema；
  3. 将提取后的字段数据与目标Schema绑定，生成新的Connect记录。
• 自定义SMT可灵活处理各种复杂场景，完全匹配业务需求。

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