将 NiFi 连接到 ElasticSearch

Question

我正在尝试解决一项任务，我将不胜感激 - 文档链接、论坛链接或除 https://cwiki.apache.org/confluence/display/NIFI/FAQs 之外的其他常见问题解答，或本文中的任何有意义的答案 =)。

所以，我有以下任务： 我系统的初始部分每 5-15 分钟从不同的数据库源收集数据。然后我删除重复项，删除垃圾，根据逻辑组合来自不同来源的数据，然后将其作为多个流重定向到系统的第二部分。 据我所知，“NiFi”可以以最好的方式完成这项任务=）。

目前我可以通过“GetHTTP”处理器成功地从 InfluxDB 获取信息。但是，我无法配置相同类型的处理器来从 Elastic DB 获取所有必要选项的信息。我想每 5-15 分钟接收一次数据，时间段从“现在-减-”到“现在”。 （取决于调度程序周期）和几个额外的过滤器。如果我理解正确，这可以通过订阅“_index”或通过定期向 DB 请求所需的时间间隔来实现。

我知道 NiFi 有几个为 Elasticsearch 设计的特定处理器（FetchElasticsearch5、FetchElasticsearchHttp、QueryElasticsearchHttp、ScrollElasticsearchHttp）以及 GetHTTP 和 PostHTTP 处理器。然而，不幸的是，我缺乏信息甚至更好的例子——如何为我的目的配置它们的“属性”=(。

FetchElasticsearchHttp、QueryElasticsearchHttp有什么区别？哪一个更适合我的任务？ GetHTTP 和 QueryElasticsearchHttp 除了几个特定字段外，还有什么区别？如果我根据需要对其进行调整，GetHTTP 会以相同的方式执行吗？

有什么建议吗？

如果有任何帮助，我将不胜感激。

Answer 1

ElasticsearchHttp 处理器尝试通过根据您设置的属性生成适当的 REST API 调用来更轻松地与 ES 交互。如果您知道所需的完整 URL，则可以使用 GetHttp 或 InvokeHttp。然而 ESHttp 处理器让你只输入你正在寻找的东西，它会生成 URL 并返回结果。

FetchElasticsearch（及其变体）用于在您知道标识符时获取特定文档。这有时在搜索/查询之后使用，以便在您知道需要哪些文档后一次返回一个文档。

QueryElasticsearchHttp 用于当您想对文档进行 Lucene 样式的查询时，当您不一定知道您想要哪些文档时。它只会返回该索引的 index.max_result_window 值。要获得更多记录，您可以在之后使用ScrollElasticsearchHttp。 注意：QueryElasticsearchHttp 需要一个查询作为 URL 的“q”参数。这种“迷你语言”不支持所有字段/运算符（有关详细信息，请参阅here）。

对于您的用例，您可能需要InvokeHttp 才能发出您描述的那种查询。 This article 描述了如何在过去 15 分钟内发出查询。返回结果后，您可能需要 EvaluateJsonPath 和/或 SplitJson 的某种组合来处理各个文档，请参阅 Elasticsearch REST API 文档（和 NiFi 处理器文档）了解更多详细信息。