добавить значение поля во встроенный объект массива json в монго через коннектор приемника kafka mongo

Question

Опубликовать первую вставку:

{"Customer_id": 2, "transaction_id": "1", "idd": [999, 1111], "id": 1}

а затем второй:

{"Customer_id": 2, "transaction_id": "2", "idd": [9, 10], "id": 1}

требуемый результат:

{"Customer_id": 2, "transaction_id": "2", "idd": [[9, 10] , [999, 1111]], "id": 1}

что я получаю:

{"Customer_id": 2, "transaction_id": "2", "idd": [9, 10] , "id": 1}

стратегия обновления этот массив обновлений не добавляется, но мой ожидаемый результат для добавления массива

конфигурация:

document.id.strategy=com.mongodb.kafka.connect.sink.processor.id.strategy.PartialValueStrategy
value.projection.list=customer_id,transaction_id
value.projection.type=whitelist
writemodel.strategy=com.mongodb.kafka.connect.sink.writemodel.strategy.ReplaceOneBusinessKeyStrategy

как мы можем это сделать через коннектор раковины mongo-kafka

Answer 1

Похоже, вы пытаетесь агрегировать потоковые данные

Если MongoDB каким-то образом не сможет предоставить эту функциональность, вставив тот же идентификатор документа (в чем я сомневаюсь, потому что откуда он знает, что вы хотите собирать только поле idd?), Вам придется использовать Kafka Streams / KSQL или другой уровень обработки с отслеживанием состояния, чтобы агрегировать ваши ценности. Коннектор приемника просто пересылает все, что видит, в базу данных; соединитель не знает предыдущих записей