Kafka Connect自定义转换：仅单次添加静态字段而非全字段

首先需要修正你给出的目标数据结构，原始示例存在JSON语法错误，合法的结构应该是将原始表数据作为一个子字段（比如data），或者直接与静态字段平级。以下是两种场景的可行配置：

场景1：原始数据作为子字段，外层添加静态字段 #

目标输出：

{"id":"123", "type":"test" , "timestamp":"1234567", "data": {"author": "Philip K. Dick", "character": "Palmer Eldritch"}}

连接器配置片段：

# 定义转换顺序：先包裹原始数据，再添加静态字段
transforms=wrapData,insertId,insertType,insertTimestamp

# WrapField转换：将原始表数据包裹到"data"字段
transforms.wrapData.type=org.apache.kafka.connect.transforms.WrapField$Value
transforms.wrapData.field=data

# 添加静态id字段
transforms.insertId.type=org.apache.kafka.connect.transforms.InsertStatic$Value
transforms.insertId.field=id
transforms.insertId.value=123

# 添加静态type字段
transforms.insertType.type=org.apache.kafka.connect.transforms.InsertStatic$Value
transforms.insertType.field=type
transforms.insertType.value=test

# 添加静态timestamp字段
transforms.insertTimestamp.type=org.apache.kafka.connect.transforms.InsertStatic$Value
transforms.insertTimestamp.field=timestamp
transforms.insertTimestamp.value=1234567

场景2：静态字段与原始字段平级 #

目标输出：

{"id":"123", "type":"test", "timestamp":"1234567", "author": "Philip K. Dick", "character": "Palmer Eldritch"}

连接器配置片段：

# 定义转换链
transforms=insertId,insertType,insertTimestamp

# 添加静态id字段
transforms.insertId.type=org.apache.kafka.connect.transforms.InsertStatic$Value
transforms.insertId.field=id
transforms.insertId.value=123

# 添加静态type字段
transforms.insertType.type=org.apache.kafka.connect.transforms.InsertStatic$Value
transforms.insertType.field=type
transforms.insertType.value=test

# 添加静态timestamp字段
transforms.insertTimestamp.type=org.apache.kafka.connect.transforms.InsertStatic$Value
transforms.insertTimestamp.field=timestamp
transforms.insertTimestamp.value=1234567

关键说明 #

• 每个InsertStatic$Value转换负责添加一个独立的静态字段，通过field指定字段名，value指定静态值，只会在消息值的最外层添加一次，不会重复添加到所有列。
• WrapField$Value用于将原始表数据包裹到子字段，避免字段名冲突，同时让结构更清晰。

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