Как написать с Spark на удаленный узел Elastic Search?

Question

У меня есть код:

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.elasticsearch.spark._
val sc = new SparkContext(conf)
conf.set("es.index.auto.create", "true")
conf.set("es.nodes", "1.2.3.4")
val numbers = Map("one" -> 1, "two" -> 2, "three" -> 3)
val airports = Map("arrival" -> "Otopeni", "SFO" -> "San Fran")
sc.makeRDD(Seq(numbers, airports)).saveToEs("spark/docs")

, но когда я его запускаю - он пытается перейти на Localhost:

sc.makeRDD(Seq(numbers, airports)).saveToEs("spark/docs")
19/06/11 11:56:16 ERROR rest.NetworkClient: Node [127.0.0.1:9200] failed (Connection refused (Connection refused)); no other nodes left - aborting...
19/06/11 11:56:16 ERROR rest.NetworkClient: Node [127.0.0.1:9200] failed (Connection refused (Connection refused)); no other nodes left - aborting...
19/06/11 11:56:16 ERROR executor.Executor: Exception in task 2.0 in stage 2.0 (TID 18)
org.elasticsearch.hadoop.EsHadoopIllegalArgumentException: Cannot detect ES version - typically this happens if the network/Elasticsearch cluster is not accessible or when targeting a WAN/Cloud instance without the proper setting 'es.nodes.wan.only'

Как настроить запись на удаленный сервер ES?

Answer 1

Pls. см конфигурация

es.nodes.discovery (по умолчанию true) Обнаруживать ли узлы в кластере es или использовать только те, которые указаны в es.nodes для запросов метаданных. Обратите внимание, что этот параметр применяется только при запуске; впоследствии при чтении и записи es использует целевые сегменты индекса (и их хост-узлы), если не включен es.nodes.client.only.

make es.nodes.discovery false
пример:

EsSpark.saveToEs(userTweetRDD, "twitter/test", Map("es.nodes" -> "xx.xx.xx.xxx", "es.cluster.name" -> xxxx-xxxxx"))

добавить

"es.nodes.discovery" -> "false"

в вашем случае

ваш пример:

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.elasticsearch.spark._


val conf:SparkConf = new SparkConf().setAppName("MYESAPP")
.setMaster("local")// for "local" for local testing if you are using yarn then "yarn"

conf.set("es.index.auto.create", "true")
conf.set("es.nodes", "1.2.3.4")
conf.set("es.nodes.discovery", "false")



val sc = new SparkContext(conf)

val numbers = Map("one" -> 1, "two" -> 2, "three" -> 3)
val airports = Map("arrival" -> "Otopeni", "SFO" -> "San Fran")
sc.makeRDD(Seq(numbers, airports)).saveToEs("spark/docs")