我在postgres中有一个数据库，它有一个表“product”，这个表是用“sales\u channel”连接1到n的。所以一个产品可以有多个销售渠道。现在我想把它转移到es并保持最新，所以我使用debezium和kafka。将单个表转移到es是没有问题的。我可以查询销售渠道和产品。但是我需要所有销售渠道的产品。我怎么才能把这个转移出去？
产品Map

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

产品Flume

{
  "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"
  }
}