如何通过Kafka Connect配置构造指定格式的Kafka消息？

完全不用修改表结构，仅通过配置Kafka Connect的内置转换插件就能构造出你需要的消息格式。核心思路是利用InsertField转换添加静态字段（包括嵌套对象），结合ExtractField处理JDBC源默认的输出结构。

具体配置示例（以JDBC源连接器为例） #

以下是关键配置片段，你可以根据数据库类型和连接器调整基础配置：

# JDBC源基础配置
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
connection.url=jdbc:mysql://你的数据库地址:3306/你的库名
connection.user=数据库用户名
connection.password=数据库密码
table.whitelist=user
mode=incrementing
incrementing.column.name=uuid

# 转换规则配置
transforms=extractPayload,addStaticProp,addStaticNestedFields
# 第一步：提取JDBC源返回的实际数据字段（默认结构包含schema和payload）
transforms.extractPayload.type=org.apache.kafka.connect.transforms.ExtractField$Value
transforms.extractPayload.field=payload

# 第二步：添加顶层静态字段staticProperty
transforms.addStaticProp.type=org.apache.kafka.connect.transforms.InsertField$Value
transforms.addStaticProp.static.field=staticProperty
transforms.addStaticProp.static.value=CREATED

# 第三步：添加嵌套的staticObject及属性
transforms.addStaticNestedFields.type=org.apache.kafka.connect.transforms.InsertField$Value
transforms.addStaticNestedFields.static.field=staticObject.staticProperty1
transforms.addStaticNestedFields.static.value=WEB
transforms.addStaticNestedFields.static.field=staticObject.staticProperty2
transforms.addStaticNestedFields.static.value=random.com

配置说明 #

1. ExtractField转换：JDBC源连接器默认输出的消息结构包含schema和payload两层，ExtractField会把payload里的uuid和email字段提取出来，得到干净的基础数据结构。
2. InsertField转换（顶层字段）：直接在消息顶层插入staticProperty字段，值固定为CREATED。
3. InsertField转换（嵌套字段）：通过点分隔的路径（如staticObject.staticProperty1），Kafka Connect会自动创建嵌套的staticObject对象，并填充对应的静态属性值。

注意事项 #

• 转换顺序很重要：必须先提取实际数据，再添加静态字段，否则会操作到包含schema的原始结构上。
• 如果使用非JDBC源连接器，只需调整基础配置部分，转换规则逻辑完全通用。

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