0
試圖在Flink培訓/ exersizes上取得進展。有如下issue。
重新安裝Elasticsearch，現在版本爲：2.4.6。
Elasticsearch似乎工作。我可以通過http://localhost:9200訪問它。
創建index和type mapping根據練習指令使用cURL命令。
安裝了Kibana：4.6.5，Kibana能夠連接到Elasticsearch並解析我創建的index。Flink 1.3.1，Elasticsearch 2.4.6，RuntimeException：Elasticsearch客戶端未連接到任何Elasticsearch節點
現在，我嘗試執行Flink工作，其內容從卡夫卡主題數據，並試圖填充數據到Elasticsearch，並且，它無法創建以下錯誤Sink：
Caused by: java.lang.RuntimeException: Elasticsearch client is not connected to any Elasticsearch nodes!
at org.apache.flink.streaming.connectors.elasticsearch2.Elasticsearch2ApiCallBridge.createClient(Elasticsearch2ApiCallBridge.java:72)
at org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase.open(ElasticsearchSinkBase.java:272)
at org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:36)
at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:111)
at org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:376)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:253)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:702)
at java.lang.Thread.run(Thread.java:748)
同樣，這裏是我的built.sbt設置：
libraryDependencies ++= Seq(
"org.apache.flink" %% "flink-scala" % "1.3.1" % "provided",
"org.apache.flink" %% "flink-streaming-scala" % "1.3.1" % "provided",
"org.apache.flink" %% "flink-clients" % "1.3.1" % "provided",
"joda-time" % "joda-time" % "2.9.9",
"com.google.guava" % "guava" % "22.0",
"com.typesafe" % "config" % "1.3.0",
"org.apache.flink" % "flink-connector-kafka-0.10_2.10" % "1.3.1",
"org.apache.flink" % "flink-connector-elasticsearch2_2.10" % "1.3.1"
)
進口：
import org.apache.flink.streaming.connectors.elasticsearch.{ElasticsearchSinkFunction, RequestIndexer}
import org.apache.flink.streaming.connectors.elasticsearch2.ElasticsearchSink
和函數構建，並返回Sink。
通過參數("localhost", 9300, "elasticsearch")。嘗試通過"127.0.0.1"爲主機，有同樣的錯誤。
private def getSink(host: String, port: Int, cluster: String) = {
val config = Map(
// This instructs the sink to emit after every element, otherwise they would be buffered
"bulk.flush.max.actions" -> "1",
// default cluster name
"cluster.name" -> cluster
)
val jConfig: java.util.Map[String, String] = new java.util.HashMap()
jConfig.putAll(config.asJava)
println(s"jConfig: ${jConfig.toString()}")
val transports = List(new InetSocketAddress(InetAddress.getByName(host), port))
val jTransports = new util.ArrayList(transports.asJava)
println(s"jTransports: ${jTransports.toString()}")
val esSink = new ElasticsearchSink(jConfig, jTransports,
new MyElasticsearchInserter("nyc-idx", "popular-locations"))
esSink
}
這裏是Elasticsearch日誌信息，以表明它的狀態：
...
...
[2017-08-08 20:12:24,203][INFO ][node ] [Beautiful Dreamer] stopping ...
[2017-08-08 20:12:27,063][INFO ][node ] [Beautiful Dreamer] stopped
[2017-08-08 20:12:27,066][INFO ][node ] [Beautiful Dreamer] closing ...
[2017-08-08 20:12:28,104][INFO ][node ] [Beautiful Dreamer] closed
[2017-08-08 20:21:58,212][INFO ][node ] [Aries] version[2.4.6], pid[1502], build[5376dca/2017-07-18T12:17:44Z]
[2017-08-08 20:21:58,264][INFO ][node ] [Aries] initializing ...
[2017-08-08 20:22:00,569][INFO ][plugins ] [Aries] modules [reindex, lang-expression, lang-groovy], plugins [], sites []
[2017-08-08 20:22:00,857][INFO ][env ] [Aries] using [1] data paths, mounts [[/ (rootfs)]], net usable_space [1.6gb], net total_space [45gb], spins? [unknown], types [rootfs]
[2017-08-08 20:22:00,857][INFO ][env ] [Aries] heap size [1007.3mb], compressed ordinary object pointers [true]
[2017-08-08 20:22:07,759][INFO ][node ] [Aries] initialized
[2017-08-08 20:22:07,759][INFO ][node ] [Aries] starting ...
[2017-08-08 20:22:08,179][INFO ][transport ] [Aries] publish_address {127.0.0.1:9300}, bound_addresses {[::1]:9300}, {127.0.0.1:9300}
[2017-08-08 20:22:08,187][INFO ][discovery ] [Aries] elasticsearch/_BFBmx-XR5aHFhbN9nvX2g
[2017-08-08 20:22:11,796][INFO ][cluster.service ] [Aries] new_master {Aries}{_BFBmx-XR5aHFhbN9nvX2g}{127.0.0.1}{127.0.0.1:9300}, reason: zen-disco-join(elected_as_master, [0] joins received)
[2017-08-08 20:22:12,415][INFO ][http ] [Aries] publish_address {127.0.0.1:9200}, bound_addresses {[::1]:9200}, {127.0.0.1:9200}
[2017-08-08 20:22:12,415][INFO ][node ] [Aries] started
[2017-08-08 20:22:12,654][INFO ][gateway ] [Aries] recovered [2] indices into cluster_state
[2017-08-08 20:22:14,526][INFO ][cluster.routing.allocation] [Aries] Cluster health status changed from [RED] to [YELLOW] (reason: [shards started [[nyc-places][1], [nyc-places][0]] ...]). ..
...
...
讓我們回到使用的軟件版本，它可能不完全匹配由培訓站點使用的版本。
礦是Flink：1.3.1和Elasticsearch：2.4.6。