将NiFi连接到ElasticSearch

Question

我正在尝试解决一项任务,并会感谢任何帮助 - 文档链接,论坛链接或其他常见问题解答,除https://cwiki.apache.org/confluence/display/NIFI/FAQs,或任何有意义的答案这篇文章=).

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

目前,我可以通过"GetHTTP"处理器成功从InfluxDB获取信息.但是我不能配置相同类型的处理器来从Elastic DB获取所有必要选项的信息.我希望从"现在 - 减去 - <5-15分钟>"到"现在"的时间段内每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"参数的查询.这种"迷你语言"不支持所有字段/运营商(有关详细信息,请参阅此处).

对于您的用例,您可能需要InvokeHttp才能发出您描述的查询类型.本文介绍如何发出最近15分钟的查询.返回结果后,您可能需要使用EvaluateJsonPath和/或SplitJson的某种组合来处理各个文档,有关详细信息,请参阅Elasticsearch REST API文档(和NiFi处理器文档).