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

uoifb46i 于 2021-06-06 发布在 Kafka

关注(0)|答案(1)|浏览(461)

我有以下关于jdbc连接器发布到的主题的json

{"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当作钥匙而不是Map。在数据库中，它作为长值存储。

apache-kafka jdbc transform apache-kafka-connect

来源：https://stackoverflow.com/questions/52748317/kafka-connect-transformation-extract-a-long-value-from-json-field-and-insert-as

1条答案

按热度按时间

aor9mmx11#

ValueToKey 获取值中的字段列表，并返回这些字段到其值的Map。
我认为你需要第二次变换来只提取这些字段中的一个。

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

赞(0）回复(0）举报 2021-06-06

我来回答

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

1条答案

相关问题

热门标签

最新问答