如何从有效负载筛选架构元数据

du7egjpx  于 2021-06-06  发布在  Kafka
关注(0)|答案(1)|浏览(241)

我试图从有效负载中删除模式,下面是配置
连接器.属性

name=test-source-mysql-jdbc-autoincrement
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
value.converter=org.apache.kafka.connect.json.JsonConverter
tasks.max=1
connection.url=jdbc:mysql://127.0.0.1:3306/employee_db?user=root&password=root
table.whitelist=testemp
mode=incrementing
incrementing.column.name=employee_id
topic.prefix=test-mysql-jdbc-

下面是我的worker.properties

bootstrap.servers=localhost:9092

key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter

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

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

offset.storage.file.filename=/tmp/connect.offsets

offset.flush.interval.ms=10000
plugin.path=C:\Users\name\Desktop\kafka\libs

输出:

{"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"employee_id"},{"type":"string","optional":false,"field":"first_name"}],"optional":false,"name":"testemp"},"payload":{"employee_id":2,"first_name":"test"}}

例外输出:

{"payload":{"employee_id":2,"first_name":"test"}}

我试着在worker中禁用value.converter.schemas.enable=false,但仍然没有效果
我错过什么了吗?

6yoyoihd

6yoyoihd1#

有两种解决方法:
删除 value.converter 属性(使用相同的 value.converter )
value.converter.schemas.enable=false 在连接器配置中。
架构被添加到消息中,因为您已经覆盖了值转换器,并且没有禁用架构(默认情况下为 JsonConverter 架构已启用)。从kafka connect的Angular 来看,您使用了全新的转换器(它不会使用全局配置中的属性)
如果要禁用架构,则消息如下:

{
    "employee_id": 2,
    "first_name":"test"
}

相关问题