Kafka JDBC Connect：根据消息字段中的多个值插入键答案

【问题标题】：Kafka JDBC Connect: Insert key based on multiple values in message fieldsKafka JDBC Connect：根据消息字段中的多个值插入键
【发布时间】：2019-03-29 09:56:06
【问题描述】：

我在 JDBC 连接器发布到的主题上有以下 json

{"APP_SETTING_ID":9,"USER_ID":10,"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

我现在想通过将两个整数字段 - APP_SETTING_ID 和 USER_ID 相乘来插入此消息的键。所以这条消息的key变成了9*10 = 90

这种转变是否可以通过 Connect 实现，如果可以，请有人解释一下

【问题讨论】：

Kafka Connect 中没有内置的方法来乘以一个值。然而，这并不意味着你不能编写自己的转换器来做到这一点。否则，它将需要您的生产者代码或其他一些 Kafka Streams/KSQL 作业来执行该逻辑。或者您可以在 SQL 查询中编写乘法
您能否指出我编写自己的转换器的相关文档。谢谢

标签： jdbc apache-kafka apache-kafka-connect confluent-platform

【解决方案1】：

我会尝试看看你能走多远

query=SELECT APP_SETTING_ID, APP_SETTING_NAME, SETTING_KEY, (APP_SETTING_ID*USER_ID) as _key FROM MY_TABLE with (nolock)

然后添加一个 ExtractKey 变换

transforms=AddKeys,ExtractKey
# this make a map
transforms.AddKeys.type=org.apache.kafka.connect.transforms.ValueToKey
transforms.AddKeys.fields=_key  
# this gets one field from the map 
transforms.ExtractKey.type=org.apache.kafka.connect.transforms.ExtractField$Key
transforms.ExtractKey.field=_key

【讨论】：

嗯，没想到在SQL里加了。那对我有用。谢谢