Коннектор Debezium mongodb kafka не выдает некоторые записи в теме, как это происходит в mongodb - PullRequest
Новая
       50

Коннектор Debezium mongodb kafka не выдает некоторые записи в теме, как это происходит в mongodb

0 голосов
/ 16 октября 2019

В моем mongodb есть эти данные 

mongo01:PRIMARY> db.col.find({"_id" : ObjectId("5d8777f188fef5555b")})
{ "_id" : ObjectId("5d8777f188fef5555b"), "attachments" : [ { "name" : "Je", "src" : "https://google.co", "type" : "image/png" } ], "tags" : [ 51, 52 ], "last_comment" : [ ], "hashtags" : [ "Je" ], "badges" : [ ], "feed_id" : "1", "company_id" : 1, "message" : "aJsm9LtK", "group_id" : "106", "feed_type" : "post", "thumbnail" : "", "group_tag" : false, "like_count" : 0, "clap_count" : 0, "comment_count" : 0, "created_by" : 520, "created_at" : "1469577278628", "updated_at" : "1469577278628", "status" : 1, "__v" : 0 }

mongo01:PRIMARY> db.col.find({"_id" : ObjectId("5d285b4554e3b584bf97759")})
{ "_id" : ObjectId("5d285b4554e3b584bf97759"), "attachments" : [ ], "tags" : [ ], "last_comment" : [ ], "company_id" : 1, "group_id" : "00e35289", "feed_type" : "post", "group_tag" : false, "status" : 1, "feed_id" : "3dc44", "thumbnail" : "{}", "message" : "s2np1HYrPuFF", "created_by" : 1, "html_content" : "", "created_at" : "144687057949", "updated_at" : "144687057949", "like_count" : 0, "clap_count" : 0, "comment_count" : 0, "__v" : 0, "badges" : [ ], "hashtags" : [ ] }

Я использую этот разъем debezium mongodb для получения данных mongodb в теме kafka. 

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" 
   http://localhost:8083/connectors/ -d '{
   "name": "mongo_connector-4",
    "config": {
    "connector.class": "io.debezium.connector.mongodb.MongoDbConnector",
    "mongodb.hosts": "mongo01/localhost:27017",
    "mongodb.name": "mongo_1",
    "collection.whitelist": "data.col",
    "key.converter.schemas.enable": false,
    "value.converter.schemas.enable": false,
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "transforms" : "unwrap",
    "transforms.unwrap.type" : "io.debezium.connector.mongodb.transforms.UnwrapFromMongoDbEnvelope",
    "transforms.unwrap.drop.tombstones" : "false",
    "transforms.unwrap.delete.handling.mode" : "drop",
    "transforms.unwrap.operation.header" : "true",
    "errors.tolerance" : "all",
    "snapshot.delay.ms":"120000",
    "poll.interval.ms":"3000",
    "heartbeat.interval.ms":"90000"
  }
}'

сейчас при печатитема в ksql я получаю, что для некоторых записей данные пришли со всеми столбцами (как это было в mongodb), в то время как для некоторых записей некоторые столбцы отсутствуют. 

ksql> print 'mongo_1.data.col' from beginning;
Format:JSON
{"ROWTIME":1571148520736,"ROWKEY":"{\"id\":\"5d8777f188fef5555b\"}","attachments":[{"name":"Je","src":"https://google.co","type":"image/png"}],"tags":[51,52],"last_comment":[],"hashtags":[],"badges":[],"feed_id":"1","company_id":1,"message":"aJsm9LtK","group_id":"106","feed_type":"post","thumbnail":"","group_tag":false,"like_count":0,"clap_count":0,"comment_count":0,"created_by":520,"created_at":"1469577278628","updated_at":"1469577278628","status":1,"__v":0,"id":"5d8777f188fef5555b"}
{"ROWTIME":1571148520736,"ROWKEY":"{\"id\":\"5d285b4554e3b584bf97759\"}","badges":[],"hashtags":[],"id":"5d285b4554e3b584bf97759"}

Почему это происходит и как решить эту проблему?

PS: единственное различие, которое я обнаружил, состоит в том, что обе записи имеют разный порядок столбцов.

При поискеоб этой проблеме только близко я нашел здесь https://github.com/hpgrahsl/kafka-connect-mongodb что-то, что они говорят о пост-обработке и редактировании полей, которые содержат конфиденциальные данные. Но, как вы видите, обе мои записи похожи и не содержат конфиденциальных данных (под конфиденциальными данными я имею в виду зашифрованные данные, возможно, они имели в виду что-то другое).

1 Ответ

0 голосов
/ 17 октября 2019

Не являются ли значения mssing после обновлений? НЕ забывайте, что соединитель MongoDB предоставляет patch для обновлений, а не after - https://debezium.io/documentation/reference/0.10/connectors/mongodb.html#change-events-value

Если вам нужно создать полный формат after, в случае MongoDB вам нужно ввести конвейер Kafka Streams, которыйсохранит событие после вставки в perisstent sotre и затем объединит патч с oribinal insart для создания финального события.

...