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

【问题标题】：Push own id. Confluent kafka connect elasticsearch docker推送自己的id。 Confluent kafka 连接 elasticsearch docker
【发布时间】：2019-01-14 10:00:36
【问题描述】：

我正在使用 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 和来源？

【问题讨论】：

标签： docker elasticsearch apache-kafka confluent-platform

【解决方案1】：

Per the docs：

如果您指定 key.ignore=true，那么 Kafka Connect 将使用消息的 kafka 主题、分区和偏移量的组合键——这就是您所看到的“奇怪的 id”。
如果您想使用自己的 ID 创建 Elasticsearch 文档，可以设置key.ignore=false，Kafka Connect 将使用 Kafka 消息的键作为 ID。

如果您的 Kafka 消息没有您想要执行的相应密钥，则需要设置它。一种选择是使用KSQL：

CREATE STREAM target AS SELECT * FROM source PARTITION BY _id

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

【讨论】：