kafka connect - 如何从有效负载中筛选架构元数据

Question

>我正在尝试从有效负载中删除架构，以下是配置

连接器属性

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-

下面是我的工人属性

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:UsersnameDesktopkafkalibs

输出：

{"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"}}

我尝试按照此处的建议在工人中禁用 value.converter.schemas.enable= false，仍然没有效果

我错过了什么吗？

Answer 1

有两个选项可以修复它：

• 从连接器配置中删除value.converter属性(使用相同的value.converter(
• 在连接器配置中设置value.converter.schemas.enable=false。

架构

已添加到消息中，因为您已覆盖值转换器并且未禁用架构(默认情况下，JsonConverter架构已启用(。从 Kafka Connect 的角度来看，您使用了全新的转换器(它不会使用全局配置中的属性(

如果要禁用架构，则消息将如下所示：

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