MongoDB作为接收器连接器没有按预期捕获数据-kafka

Question

我目前使用MySQL数据库作为源连接器，使用下面的配置，我想监控数据库的更改并将其发送到mongoDB，

这是我的源连接器配置，

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '''{
"name": "source_mysql_connector",  
"config": {  
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
"tasks.max": "1",  
"database.hostname": "host.docker.internal",  
"database.port": "3306",
"database.user": "test",
"database.password": "$apr1$o7RbW.GvrPIY1",
"database.server.id": "8111999",  
"database.server.name": "db_source",  
"database.include.list": "example",  
"database.history.kafka.bootstrap.servers": "broker:29092",  
"database.history.kafka.topic": "schema-changes.example",
"database.allowPublicKeyRetrieval":"true",
"include.schema.changes": "true"
}
}'''

这是我的接收器连接器(mongodb(配置，

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '''{
"name": "sink_mongodb_connector",  
"config": {  
"connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
"tasks.max":"1",
"topics":"db_source.example.employees",
"connection.uri":"mongodb://172.17.0.1:27017/example?w=1&journal=true",
"database":"example",
"collection":"employees",
"value.converter": "io.confluent.connect.avro.AvroConverter",
"value.converter.schema.registry.url": "http://schema-registry:8081"
}
}'''

使用它，我能够建立连接，捕捉数据更改，并将它们存储到名为employees的表的mongodb集合中，

但这里的问题是，当我检查mongodb中的集合时，文档是这样保存的，

{ "_id" : ObjectId("60d0e6939e00e22f274ccac1"), "before" : null, "after" : { "id" : NumberLong(11), "name" : "Steve Shining", "team" : "DevOps", "birthday" : 11477 }, "source" : { "version" : "1.5.0.Final", "connector" : "mysql", "name" : "db_source", "ts_ms" : NumberLong("1624303251000"), "snapshot" : "false", "db" : "example", "sequence" : null, "table" : "employees", "server_id" : NumberLong(6030811), "gtid" : null, "file" : "mysql-bin.000003", "pos" : NumberLong(5445), "row" : 2, "thread" : null, "query" : null }, "op" : "c", "ts_ms" : NumberLong("1624303251190"), "transaction" : null }

{ "_id" : ObjectId("60d0e6939e00e22f274ccac2"), "before" : null, "after" : { "id" : NumberLong(12), "name" : "John", "team" : "Support", "birthday" : 6270 }, "source" : { "version" : "1.5.0.Final", "connector" : "mysql", "name" : "db_source", "ts_ms" : NumberLong("1624303251000"), "snapshot" : "false", "db" : "example", "sequence" : null, "table" : "employees", "server_id" : NumberLong(6030811), "gtid" : null, "file" : "mysql-bin.000003", "pos" : NumberLong(5445), "row" : 3, "thread" : null, "query" : null }, "op" : "c", "ts_ms" : NumberLong("1624303251190"), "transaction" : null }

但是我的mysql数据库是这样的，

mysql> select * from employees;
+----+---------------+-----------+------------+------------+
| id   | name                | team          |  birthday   |
+----+---------------+-----------+------------+------------+
|  1    | Peter Smith     | DevOps     | 2003-07-21  |
| 11    | Steve Shining | DevOps     | 2001-06-04 |
| 12   | John                  | Support    | 1987-03-03  |
+----+---------------+-----------+------------+------------+

，我希望我的收藏品看起来像这样

{ "_id" : ObjectId("60d0e6939e00e22f274ccac2"), "name" : "John", "team" : "Support", "birthday" : "1987-03-03 "}

我在这里做错了什么？即使删除消息像这样存储在集合中，它也无法识别消息和所有消息。我该如何修复它？连日期都没有正确存储？

更新：

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '''{
"name": "sink_mongodb_connector",  
"config": {  
"connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
"tasks.max":"1",
"topics":"db_source.example.employees",
"connection.uri":"mongodb://172.17.0.1:27017/example?w=1&journal=true",
"database":"example",
"collection":"employees",
"value.converter": "io.confluent.connect.avro.AvroConverter",
"value.converter.schema.registry.url": "http://schema-registry:8081",
"transforms": "unwrap",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.unwrap.drop.tombstones": "false",
"transforms.unwrap.delete.handling.mode": "rewrite"
}
}'''

Answer 1

问题与Mongo无关，而是默认的Debezium格式。

您看到的是之前、之后和其他CDC事件元数据。

无法识别消息

它是，不过。。。"after" : { "id" : NumberLong(12), "name" : "John", "team" : "Support", "birthday" : 6270 }

您需要提取/展平事件，以便仅获得"；在"；现场

https://debezium.io/documentation/reference/configuration/event-flattening.html

---

关于生日/日期值，似乎是一个单独的问题