卡夫卡数据到弹性搜索答案

【问题标题】：kafka data to elaticsearch卡夫卡数据到弹性搜索
【发布时间】：2018-02-06 17:22:09
【问题描述】：

我正在尝试使用 kafka connect elasticsearch 将数据从 apache kafka 生产者发送到弹性搜索

在 elasticsearch 中出现以下错误：

[2017-12-21T11:00:54,979][DEBUG][o.e.a.b.TransportShardBulkAction] 
[pageviews7][0] failed to execute bulk item (index) BulkShardRequest 
[[pageviews7][0]] containing [index {[pageviews7][kafkaconnect]
[pageviews7+0+0], source[{"key1":"value1"}]}]
org.elasticsearch.index.mapper.MapperParsingException: failed to find 
type parsed [string] for [key1]

以下是我的 kafka-connect-elasticsearch 属性文件：

name=elasticsearch-sink
connector.class=
io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
tasks.max=1
topics=pageviews7
key.ignore=true
connection.url=http://localhost:9200
type.name=kafkaconnect
schemas.enable=false
schema.ignore=true

【问题讨论】：

标签： elasticsearch apache-kafka apache-kafka-connect

【解决方案1】：

尝试使用 stream-reactor Kafka 连接器进行 Elastic-Search。一种使用 TCP 协议，另一种使用 HTTP 协议，它们已经在生产环境中进行了实战测试，每天将 5+ 十亿个事件放入 ES 索引中，并且它们非常易于配置且有据可查

https://github.com/landoop/stream-reactor

【讨论】：

【解决方案2】：

这看起来像是将文档索引到 Elasticsearch 中的问题：

尝试手动将文档发布到 ES 中
您应该得到同样的错误。这意味着您的索引映射有问题。
如果我怀疑索引存在映射问题，您可以使用 topic.index.map 属性写入另一个索引或删除当前索引 (仅当您不关心数据时）。

如何手动卷曲到 ES 中的示例：

curl -XPOST <ES-url>/pageviews7/test_id -d '{"key1":"value1"}'

删除您当前的索引：

curl -XDELETE <ES-url>/pageviews7

【讨论】：

【解决方案3】：

您运行的是哪个版本的 Elasticsearch？您可以在 Kafka Connect 连接器配置中设置 schema.ignore=true 来解决此错误。

【讨论】：