无法从Kafka Connect AVRO连接器解析主题

Question

我使用Neo4j sink连接器从Kafka主题读取数据并将其转储到Neo4j数据库。Kafka中可用的消息/数据是AVRO格式，因此我试图使用AVRO转换器通过提供模式注册表详细信息来解析数据。但是，在消费消息时，我看到一个DataError异常。

下面是创建连接器的配置。

{
"topics": "mytopic",
"connector.class": "streams.kafka.connect.sink.Neo4jSinkConnector",
"tasks.max":"1",
"key.converter.schemas.enable":"true",
"values.converter.schemas.enable":"true",
"errors.retry.timeout": "-1",
"errors.retry.delay.max.ms": "1000",
"errors.tolerance": "none",
"errors.deadletterqueue.topic.name": "deadletter-topic",
"errors.deadletterqueue.topic.replication.factor":1,
"errors.deadletterqueue.context.headers.enable":true,
"key.converter":"org.apache.kafka.connect.storage.StringConverter",
"key.converter.enhanced.avro.schema.support":true,
"value.converter.enhanced.avro.schema.support":true,
"value.converter":"io.confluent.connect.avro.AvroConverter",
"value.converter.schema.registry.url":"https://schema-url/",
"value.converter.basic.auth.credentials.source":"USER_INFO",
"value.converter.basic.auth.user.info":"user:pass",
"errors.log.enable": true,
"schema.ignore":"false",
"errors.log.include.messages": true,
"neo4j.server.uri": "neo4j://my-ip:7687/neo4j",
"neo4j.authentication.basic.username": "neo4j",
"neo4j.authentication.basic.password": "neo4j",
"neo4j.encryption.enabled": false,
"neo4j.topic.cypher.mytopic": "MERGE (p:Loc_Con{name: event.geography.name})"
}

这是我得到的期望。

ErrorData(originalTopic=mytopic, timestamp=1652188554497, partition=0, offset=2140111, exception=org.apache.kafka.connect.errors.DataException: Exception thrown while processing field 'geography', key=9662840       , value=Struct{geography=Struct{geoId=43333,geoType=Business Defined Area,name=Norarea,status=Active,validFrom=Sat Apr 09 00:00:00 GMT 2012,validTo=Fri Dec 31 00:00:00 GMT 9999, executingClass=class streams.kafka.connect.sink.Neo4jSinkTask)

我想知道这里出了什么问题，我也尝试过字符串和JSON转换器，但也有解析失败。那么是否有解析数据的选项呢?

Answer 1

我找到问题所在了。我需要一个转换来提取名为geography的数据字段，它基本上是从整个JSON中提取地理字段并将其分配回' event".

"transforms": "ExtractField", 
"transforms.ExtractField.type": "org.apache.kafka.connect.transforms.ExtractField$Value", 
"transforms.ExtractField.field": "geography"