代码之家 › 专栏 › 技术社区 › Fizi

Kafka JDBC Connect:基于消息字段中的多个值插入键

confluent apache-kafka-connect apache-kafka jdbc

Fizi · 技术社区 · 6 年前

{"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 . 所以这个信息的关键是 9*10 = 90

这种转变是否可能通过连接,如果是的话,请有人阐明它

1 回复 | 直到 6 年前

OneCricketeer Gabriele Mariotti 6 年前

我想看看你能走多远

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

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

推荐文章

Vaibhav Sharma · 使用Servlet连接KafkaProducer

7 年前

Evaldas Buinauskas · Kafka Connect将JSON字符串转换为实际JSON

7 年前

Nandish Kotadia · 我可以编写自定义kafka connect转换来将JSON转换为AVRO吗?

7 年前

Cody Ferguson · PubSub-Kafka连接节点连接文件结束异常

7 年前

Jaime Caffarel Daniel Koverman · kafka SourceTask类的poll()方法是否线程安全?

7 年前

Warsame · confluent-3.2.0 S3连接器默认分区设置

7 年前

Prajakta Kulkarni · 如何从flink访问/读取卡夫卡主题数据?

8 年前

Renukaradhya · Kafka Connect:在分布式模式下创建新连接器就是创建新组

8 年前

LaserJesus · 为什么Kafka分布式连接器在我创建它的节点被杀死时死亡?

8 年前

Rushabh Shah · 如何在Windows 7上使用Kafka Connect,因为我看不到独立连接或分布式连接。kafka 2.10.0.1中/bin/Windows文件夹下的bat?

8 年前