NiFi在HTTP和特定Elasticsearch处理器上都无法写入elasticsearch。

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了NiFi在HTTP和特定Elasticsearch处理器上都无法写入elasticsearch。相关的知识,希望对你有一定的参考价值。

我正试图按照NiFi写到Elastic的入门教程。

https:/linkbynet.github.ioelasticsearchtuning20170207Bitcoin-ELK-NiFi.html。

我的NiFi版本是1.11.4(当前)和Elastic 7.7.1(当前),分别作为dockefile。https:/gist.github.comgeoHeil7e7426a4d2b2fd363c036d9b12069598。.

nifi flow xml的完整版本也可在以下网站获得。https:/github.comgeoHeilstreaming-referenceblobmasterfor_nificonfflow.xml.gz。

我遇到了以下问题。

  • 在... PutElasticsearch5 处理器。
    1. bitstamp-${timestamp:multiply(1000):format(“yyyy-MM-dd”)} 抱怨一个无效的模式
Query: ${timestamp:multiply(1000):format(“yyyy-MM-dd”)}, transferring to failure: org.apache.nifi.attribute.expression.language.exception.AttributeExpressionLanguageException: Invalid Expression: bitstamp-${timestamp:multiply(1000):format(“yyyy-MM-dd”)} due to Unexpected token '“yyyy-MM-dd”' at line 1, column 34.

但弹性的日志显示。["java.lang.IllegalStateException: Received message from unsupported version: [5.0.0] minimal compatible version is: [6.8.0]",

2. when simplifying the indexing to: `bitstamp-${timestamp}`, fails with:

```
PutElasticsearch5[id=97ed752a-0172-1000-0000-000024edfeac] Failed to insert into Elasticsearch due to None of the configured nodes are available: [{#transport#-1}{9Tgt3qGwSAm5hNsRIEiS-Q}{172.18.0.6}{172.18.0.6:9300}]. More detailed information may be available in the NiFi logs.: NoNodeAvailableException[None of the configured nodes are available: [{#transport#-1}{9Tgt3qGwSAm5hNsRIEiS-Q}{172.18.0.6}{172.18.0.6:9300}]]
```
And the logs keep complaining about an incompatible API. Still, I do not understand where 1) is a parsing failure of the expression langugage initially.

当试图使用HTTP API来解决二进制不兼容的问题时(并坚持使用(2)来解决),

PutElasticsearchHttp[id=9801caaa-0172-1000-ffff-ffffe063c238] Routing to failure due to exception: unexpected end of stream on http://elasticsearch:9300/...: java.io.IOException: unexpected end of stream on http://elasticsearch:9300/...

nifi_1            | 2020-06-10 19:59:34,717 ERROR [Timer-Driven Process Thread-9] o.a.n.p.e.PutElasticsearchHttp PutElasticsearchHttp[id=9801caaa-0172-1000-ffff-ffffe063c238] Routing to failure due to exception: java.io.IOException: unexpected end of stream on http://elasticsearch:9300/...: java.io.IOException: unexpected end of stream on http://elasticsearch:9300/...
nifi_1            | java.io.IOException: unexpected end of stream on http://elasticsearch:9300/...
nifi_1            |     at okhttp3.internal.http1.Http1ExchangeCodec.readResponseHeaders(Http1ExchangeCodec.java:236)
nifi_1            |     at okhttp3.internal.connection.Exchange.readResponseHeaders(Exchange.java:115)
nifi_1            |     at okhttp3.internal.http.CallServerInterceptor.intercept(CallServerInterceptor.java:94)
nifi_1            |     at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:142)
nifi_1            |     at okhttp3.internal.connection.ConnectInterceptor.intercept(ConnectInterceptor.java:43)
nifi_1            |     at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:142)
nifi_1            |     at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:117)
nifi_1            |     at okhttp3.internal.cache.CacheInterceptor.intercept(CacheInterceptor.java:94)
nifi_1            |     at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:142)
nifi_1            |     at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:117)
nifi_1            |     at okhttp3.internal.http.BridgeInterceptor.intercept(BridgeInterceptor.java:93)
nifi_1            |     at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:142)
nifi_1            |     at okhttp3.internal.http.RetryAndFollowUpInterceptor.intercept(RetryAndFollowUpInterceptor.java:88)
nifi_1            |     at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:142)
nifi_1            |     at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:117)
nifi_1            |     at okhttp3.RealCall.getResponseWithInterceptorChain(RealCall.java:229)
nifi_1            |     at okhttp3.RealCall.execute(RealCall.java:81)
nifi_1            |     at org.apache.nifi.processors.elasticsearch.AbstractElasticsearchHttpProcessor.sendRequestToElasticsearch(AbstractElasticsearchHttpProcessor.java:296)
nifi_1            |     at org.apache.nifi.processors.elasticsearch.PutElasticsearchHttp.onTrigger(PutElasticsearchHttp.java:315)
nifi_1            |     at org.apache.nifi.processor.AbstractProcessor.onTrigger(AbstractProcessor.java:27)
nifi_1            |     at org.apache.nifi.controller.StandardProcessorNode.onTrigger(StandardProcessorNode.java:1176)
nifi_1            |     at org.apache.nifi.controller.tasks.ConnectableTask.invoke(ConnectableTask.java:213)
nifi_1            |     at org.apache.nifi.controller.scheduling.TimerDrivenSchedulingAgent$1.run(TimerDrivenSchedulingAgent.java:117)
nifi_1            |     at org.apache.nifi.engine.FlowEngine$2.run(FlowEngine.java:110)
nifi_1            |     at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
nifi_1            |     at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
nifi_1            |     at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
nifi_1            |     at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
nifi_1            |     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
nifi_1            |     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
nifi_1            |     at java.lang.Thread.run(Thread.java:748)
nifi_1            | Caused by: java.io.EOFException: 
 not found: limit=24 content=54686973206973206e6f7420616e204854545020706f7274…
nifi_1            |     at okio.RealBufferedSource.readUtf8LineStrict(RealBufferedSource.java:240)
nifi_1            |     at okhttp3.internal.http1.Http1ExchangeCodec.readHeaderLine(Http1ExchangeCodec.java:242)
nifi_1            |     at okhttp3.internal.http1.Http1ExchangeCodec.readResponseHeaders(Http1ExchangeCodec.java:213)
nifi_1            |     ... 30 common frames omitted

URI被设置为:"http:/elasticsearch: http:/elasticsearch:9300

所以我有两个问题。

NiFi应该如何和elastic对话?最新支持的版本是什么?一般来说,HTTP(较慢)的处理方式是否更值得推荐呢? 如何修复表达式语言字符串的断裂。bitstamp-${timestamp:multiply(1000):format(“yyyy-MM-dd”)}

答案

1)这是一个复制粘贴的bug。当重新输入引号时,它能很好地解析表达式。

2)确实在二进制协议中存在不兼容的地方,人们正在努力解决这个问题。https:/issues.apache.orgjirabrowseNIFI-6403?jql=project%20%3D%20NIFI%20AND%20text%20~%20%22elasticsearch%207%22。 ;

事实上,在HTTP NiFi插件的支持上,甚至似乎还存在着一些小毛病。所以可能在一些边缘情况下,需要手动的HTTP处理器。

3)9200端口是二进制端口.切换到9300是需要的。

以上是关于NiFi在HTTP和特定Elasticsearch处理器上都无法写入elasticsearch。的主要内容,如果未能解决你的问题,请参考以下文章

如何在 Apache NiFi 的 ExecuteProcess 中运行 sed

对于没有类型的索引,您如何允许在 elasticsearch6 上使用特定的 http 方法?

Apache NiFi 实例挂在“计算流文件沿袭...”窗口

NIFI Apache NiFI 授权配置

带有物联网传感器的 Apache Nifi

Apache NiFi 入门指南