postgresql - Debezium Postgres和ElasticSearch-在ElasticSearch中存储复杂对象

我在Postgres中有一个数据库，其中的表“product”与“sales_Channel”连接为1到n。因此，一个产品可以有多个SalesChannel。现在，我想将其传输到ES并保持最新状态，因此我正在使用debezium和kafka。将单个表转移到ES没问题。我可以查询SalesChannels和Products。但是我需要带有所有SalesChannel的产品作为结果。我如何获得debezium的转让？
产品映射

    {
  "settings": {
    "number_of_shards": 1
  },
  "mappings": {
    "_doc": {
      "properties": {
        "id": {
          "type": "integer"
        }
      }
    }
  }
}

产品水槽

   {
  "name": "es-sink-product",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": "1",
    "topics": "product",
    "connection.url": "http://elasticsearch:9200",
    "transforms": "unwrap,key",
    "transforms.unwrap.type": "io.debezium.transforms.UnwrapFromEnvelope",
    "transforms.unwrap.drop.tombstones": "false",
    "transforms.unwrap.drop.deletes": "false",
    "transforms.key.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
    "transforms.key.field": "id",
    "key.ignore": "false",
    "type.name": "_doc",
    "behavior.on.null.values": "delete"
  }
}

最佳答案

您要么需要使用发件箱模式，请参见https://debezium.io/documentation/reference/1.2/configuration/outbox-event-router.html
或者您可以使用聚合对象，请参见
https://github.com/debezium/debezium-examples/tree/master/jpa-aggregations
https://github.com/debezium/debezium-examples/tree/master/kstreams-fk-join