docker - 推送自己的id。 Confluent kafka 连接 elasticsearch docker

Question

我正在使用 confluentinc/cp-kafka-connect docker 镜像。
我正在尝试使用 elasticsearch id 将 JSON 文件发送到 kafka。

{"_id":10000725, "_source": {"createdByIdentity":"tu_adminn","createdBy":"Admin Testuser"}}

这是我的连接器

{
  "name": "test-connector",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": "1",
    "topics": "andrii",
    "key.ignore": "false",
    "schema.ignore": "true",
    "connection.url": "http://elasticsearch:9200",
    "type.name": "test-type",
    "name": "elasticsearch-sink"
  }
}

当我使用 key.ignore = true 时，它​​会生成一些奇怪的 id。
我怎样才能准确地传递我的 id 和来源？

Answer 1

Per the docs :

如果您指定 key.ignore=true然后 Kafka Connect 将使用消息的 kafka 主题、分区和偏移量的组合键——这就是你看到的“奇怪的 id”。

如果要使用自己的 ID 创建 Elasticsearch 文档，可以设置 key.ignore=false Kafka Connect 将使用 Kafka 消息的 key 作为 ID。

如果您的 Kafka 消息没有适合您想要执行的操作的 key ，则需要设置它。一种选择是使用 KSQL 之类的东西。 :

CREATE STREAM target AS SELECT * FROM source PARTITION BY _id

免责声明:我在开源 KSQL 项目背后的公司 Confluent 工作