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

Question

我正试图按照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”)}