Kafka使用JSONConverter连接MongoDB Kafka连接器不起作用答案

【问题标题】：Kafka connect MongoDB Kafka connector using JSONConverter not workingKafka使用JSONConverter连接MongoDB Kafka连接器不起作用
【发布时间】：2020-06-02 13:01:03
【问题描述】：

我正在尝试将 Kafka 连接器配置为使用 mongoDB 作为源并将记录发送到 Kakfa 主题。我已成功完成此操作，但我正在尝试使用 JSONConverter 执行此操作，以便也将架构与有效负载一起保存。

我的问题是连接器保存数据如下：

{ "schema": { "type": "string" } , "payload": "{....}" }

换句话说，它会自动假设实际的 JSON 是一个字符串，并将架构保存为字符串。

这就是我设置连接器的方式：

curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
 "name": "newtopic",
 "config": { 
"tasks.max":1,
 "connector.class":"com.mongodb.kafka.connect.MongoSourceConnector", 
 "key.converter":"org.apache.kafka.connect.json.JsonConverter", 
 "key.converter.schemas.enabled": "true",
 "value.converter":"org.apache.kafka.connect.json.JsonConverter", 
 "value.converter.schemas.enabled": "true",
 "connection.uri":"[MONGOURL]", 
 "database":"dbname", 
 "collection":"collname", 
 "pipeline":"[]", 
 "topic.prefix": "", 
 "publish.full.document.only": "true" 
 }}'

我是否缺少配置的内容？是不是根本无法猜测存储在MongoDB中的文档的schema，所以用String？

【问题讨论】：

注意：Kafka Connect 并不特定于 Confluent。没有像 Confluent Kafka 这样的东西
您是否能够使用给定的答案之一解决此问题？

标签： apache-kafka apache-kafka-connect

【解决方案1】：

没什么好猜的。 Mongo 是无模式的，我上次检查过，所以模式是字符串或字节。我建议使用 AvroConverter 或将模式设置为 false

您可能还想尝试使用 Debezium 看看是否会得到不同的结果

【讨论】：

感谢您的回复，几个 cmets： 1- 我知道 mongo 是无模式的，我的印象是连接器会从提交的文档中推断模式。 2- Debezium 不是一个选项，因为融合连接器具有我需要的一些功能。 3- 我需要架构，而 Avro 使用 pyspark API 无法访问的架构注册表。
1- 有什么可以推断的？我以前从未使用过带有模式的 Mongo... 2 - Confluent 没有开发 Mongo 源连接器。 Mongo 公司做了... 3 - Avro/schema 注册表确实可以从 pyspark 获得，我以前做过

【解决方案2】：

MongoDB 连接器的最新version 1.3 应该可以解决这个问题。
它甚至提供了推断源模式的选项。

【讨论】：