Kafka 将 Debezium 与 ms sql 服务器连接起来。密钥提取配置问题

Question

我正在使用 Debezium 和 kafka connect。我需要从下面的有效负载中提取“listid”，并能够将其分配给 kafka 消息键。使用我的连接器文件中的配置，我无法提取该值。感谢您为解决此问题提供的任何帮助。

{
    "before": null,
    "after": {
        "listid": 19,
        "billingid": "0",
        "userid": "test",

连接器属性

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


value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8081

transforms=unwrap,extract
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
transforms.extractkey.type==org.apache.kafka.connect.transforms.ExtractField$Key
transforms.extractkey.field=listid


key.converter.schemas.enable=false
value.converter.schemas.enable=true

---

Sooooo 我正在继续实验，并希望分享对问题的更好解释以及我所实验的内容。

跟随 Robin 的博客https://www.confluent.fr/blog/simplest-useful-kafka-connect-data-pipeline-world-thereabouts-part-3/

1- 从终端运行 kafka-avro 消费者

键

{"LIST_ID":10058}

价值

{"before":null,"after":{"test.dbo.test.Value":{"LIST_ID":10058,"billingid 
etc...

2- 我正在尝试获取值为 10058 的键

3 - 这就是我的转换配置

key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://localhost:8081
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8081
transforms=createKey,extractInt,
transforms.createKey.type=org.apache.kafka.connect.transforms.ValueToKey
transforms.createKey.fields=LIST_ID
transforms.extractInt.type=org.apache.kafka.connect.transforms.ExtractField$Key
transforms.extractInt.field=LIST_ID

我在针对 mssql 服务器的管道中使用 debezium 连接器。

Answer 1

看起来像是一个错字（extract vs extractkey）：

transforms=unwrap,extractkey
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
transforms.extractkey.type=org.apache.kafka.connect.transforms.ExtractField$Key
transforms.extractkey.field=listid