使用 NiFi 拉取 Elasticsearch 索引答案

【问题标题】：Using NiFi to pull Elasticsearch Indexes使用 NiFi 拉取 Elasticsearch 索引
【发布时间】：2017-11-20 19:12:07
【问题描述】：

我正在使用 NiFi（最近开始使用它，它似乎适合我的需求）。我们最近建立了一个 Spark/Hadoop 集群，并且已经使用 Elasticsearch 大约 2 年了。我的目标是从 Elasticsearch 获取特定索引到 HDFS（特别是系统日志）。我正在做一个异常检测的机器学习项目，但想处理来自 HDFS 的数据以加快速度。

所以，有一点背景 - 我们的系统日志索引每天都不同（logstash-syslog-2017-11-20，等等）。我只需要来自系统日志的消息，所以基本上我想要做的是：

ES -> NiFi -> Parse JSON to give me back text -> write each message to its own line in a text file.

最后，在我的 HDFS 中，我会有每个索引（天）的消息文本文件，例如：

syslog-2017-11-19
syslog-2017-11-20
syslog-2017-11-21

等等……

我被一些事情难住了：

构建它需要哪些组件？我看到有 GenerateFlowFile，我认为我需要使索引名称动态化。
由于我想拉整个索引，我想我需要使用“ScrollElasticSearchHttp”，但我不确定。还有其他选择，但我不知道什么是最好的。使用 PySpark 时，我使用 ES-Hadoop 连接器完成了简单的查询以获取整个索引，但必须将滚动大小增加到 10k，以使其运行得更快。只是对我应该使用什么处理器感到困惑。

如果有人可以告诉我这个结构的概念（什么处理器、连接器等），我需要从 syslog 获取消息的索引，从 ES 到我的 HDFS，这将是很棒的。还在学习这个，所以请原谅我对此的无知。非常感谢您的帮助！

【问题讨论】：

评论不用于扩展讨论；这个对话是moved to chat。

标签： elasticsearch apache-nifi

【解决方案1】：

还有ListenBeats 处理器。您可以将 Logstash 重定向到 NiFi，Nifi 可以同时写入 EL 和 HDF。确实，这会将 NiFi 置于您的关键路径上。

还可以编写自己的处理器，并且可以非常轻松地做到这一点。关注this article

我最近也发现了 Nifi，我觉得很棒。玩了一下，因此我不是专家。

【讨论】：

我最终使用了 ScrollElasticsearchHttp 处理器，根据我上面的评论，我似乎有一些选项格式不正确。一旦我得到了正确的格式，它就起作用了。我希望 NiFi 文档有更多示例/明确的示例来显示格式并将其与 ES-Hadoop 格式化选项的方式区分开来。无论如何，现在一切正常。我有兴趣研究编写自己的处理器 - 是否有指南或其他内容？
只需点击答案中的链接即可。您会在那里找到指南和示例。
杜...对不起，你完全给了我链接，然后我问了。我检查了那个链接，好东西。我认为 ScrollElasticSearchHttp 是我现在的答案，但我认为，基于我们庞大的不同数据源清单，我很可能会在某个时候需要它。很好的信息。

【解决方案2】：

我将发布我的第一条评论作为答案，因为这最终成为了我的解决方案。

我最终使用了 ScrollElasticsearchHttp 处理器，根据我上面的评论，我似乎有一些选项格式不正确。一旦我得到了正确的格式，它就起作用了。我希望 NiFi 文档有更多示例/明确的示例来显示格式并将其与 ES-Hadoop 格式化选项的方式区分开来。无论如何，现在一切正常。我有兴趣研究编写自己的处理器 - 是否有指南或其他内容？

【讨论】：

你能分享你提到的文件吗
azdatasci，您能否在此处发布您的 NiFi 数据流的屏幕截图。特别是您设置的选项。