Elasticsearch / Storm 集成方法答案

【问题标题】：Elasticsearch / Storm integration methodsElasticsearch / Storm 集成方法
【发布时间】：2014-11-05 06:14:44
【问题描述】：

寻找 Elasticsearch 和 Apache Storm 之间的简单集成路径。 elasticsearch-hadoop 库中包含对此的支持，但这会给 Hadoop 堆栈带来大量依赖：从 Hive 到 Cascading，我根本不需要。有没有人在没有引入 elasticsearch-hadoop 的情况下成功实现了这种集成？谢谢。

【问题讨论】：

在这种情况下，集成是什么意思？你想达到什么目标？ ES 作为 ES 中 Storm 或 Storm 存储的输入？
两种方式。我想从一个 spout 迭代一个滚动，将元组发送到各种流中。反过来，一些螺栓会在处理后写入 ES。我正在实现 ES 既是源又是目标的管道。在将 spout 的源 ES 连接添加到设置函数中以确保在启动拓扑之前对其进行序列化后，我取得了一些进展。
您不会直接使用elasticsearch Java API 还是我不理解您的问题？
正确。我正在使用 Java API。问题是在 Storm 中高吞吐量处理 ES 数据的最佳实践是什么。我目前在多个 Storm spout 上打开一个传输客户端，每个都从一个滚动读取并发出 JSON 元组。然后每个 Bolt 都有自己的传输客户端连接——它们对传入的元组进行批处理并定期发出批量写入。这可行，但吞吐量非常低。我正在研究使用 ES 节点客户端连接来避免两跳操作。还会在创建过程中禁用目标索引中的副本。

标签： hadoop elasticsearch apache-storm

【解决方案1】：

在我的项目中，我们使用rabbitmq river 来索引风暴输出。这是写入elasticsearch的非常有效和方便的方式。您基本上将消息放入队列中，其余的由河流完成。如果有东西卡住了，数据就会简单地缓冲在队列中。

所以我想说，使用这种河流方法来编写和使用 elasticsearch Java API 来读取，就像 Kit Menke 建议的那样（或 Jest client，我们发现这很酷，它提供了基于 ApacheHttpAsyncClient 的异步 API，尽管我们'不是从风暴拓扑中的弹性搜索中读取，而是在不同的服务中读取）。

【讨论】：

谢谢，非常有用的观点。听起来您有一个 RabbitMQ 集群——它的大小与您的 ES 集群相当吗？如果没有，你能提供一个节点比率的想法吗？
我玩过工人数量、批量大小（用于螺栓上的批量插入）和滚动大小，无论我做什么，吞吐量似乎都稳定在 1500 文档/秒（~4KB/文档）左右。 Storm 和 ES 中都有大量未使用的 CPU、内存和 I/O 容量。喷口（源滚动）显示“完全延迟（毫秒）”为零，我假设这意味着源文档在 1 毫秒内处理。螺栓端的延迟（写入 ES）约为 8 毫秒，这似乎相当高。