Kafka 连接转换：从 json 字段中提取一个 Long 值并作为键插入

Question

我有以下 json 关于 JDBC 连接器发布到

的主题

{"APP_SETTING_ID":9,"APP_SETTING_NAME":"my_name","SETTING_KEY":"my_setting_key"}

这是我的连接器文件

name=data.app_setting
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
poll.interval.ms=500
tasks.max=4
mode=timestamp
query=SELECT APP_SETTING_ID, APP_SETTING_NAME, SETTING_KEY,FROM MY_TABLE with (nolock)
timestamp.column.name=LAST_MOD_DATE
topic.prefix=data.app_setting

key.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false

transforms=InsertKey
transforms.InsertKey.type=org.apache.kafka.connect.transforms.ValueToKey
transforms.InsertKey.fields=APP_SETTING_ID

这确实添加了一个密钥，但它也是一种 json 格式，如

{"APP_SETTING_ID":9}

虽然我只想将 9 作为键而不是地图。在数据库中，它存储为 Long 值。

Answer 1

ValueToKey 采用值中的字段列表，returns 将这些字段映射到它们的值。

我认为您需要第二次转换以仅提取其中一个字段。

transforms=ReplaceKey,ExtractKey

# Replaces the key with fields in the value. Creates a map for all listed fields
transforms.ReplaceKey.type=org.apache.kafka.connect.transforms.ValueToKey
transforms.ReplaceKey.fields=APP_SETTING_ID

# Extracts a specfic field from the key, assuming it's a map/struct type
transforms.ExtractKey.type=org.apache.kafka.connect.transforms.ExtractField$Key
transforms.ExtractKey.field=APP_SETTING_ID

Kafka 连接转换：从 json 字段中提取一个 Long 值并作为键插入

Kafka Connect Transformation: Extract a Long value from json field and insert as key

transform

jdbc

apache-kafka

apache-kafka-connect