# Error Address already in use when I use ElasticSearch with Spark **URL:** \ **Category:** Elasticsearch\ **Tags:** es-hadoop\ **Created:** [August 11, 2016, 12:30pm UTC](https://discuss.elastic.co/t/error-address-already-in-use-when-i-use-elasticsearch-with-spark/57812 "2016-08-11T12:30:07Z")\ **Posts on this page:** 7\ **Page:** 1
**Author:** ![Guillermo\_Ortiz](https://avatars.discourse-cdn.com/v4/letter/g/258eb7/32.png) [@Guillermo\_Ortiz](https://discuss.elastic.co/u/Guillermo_Ortiz)\ **Post date:** [August 11, 2016, 12:30pm UTC](https://discuss.elastic.co/t/error-address-already-in-use-when-i-use-elasticsearch-with-spark/57812/1 "2016-08-11T12:30:07Z")
I'm using Spark 1.6 and Elasticsearch 2.2.0. When there are too many batches I got this exception 2016-08-16 09:20:54,237 [Executor task launch worker-1] ERROR org.apache.spark.executor.Executor - Exception in task 2.0 in stage 28.0 (TID 307) 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' at org.elasticsearch.hadoop.rest.InitializationUtils.discoverEsVersion(InitializationUtils.java:196) at org.elasticsearch.hadoop.rest.RestService.createWriter(RestService.java:379) at org.elasticsearch.spark.rdd.EsRDDWriter.write(EsRDDWriter.scala:40) at org.elasticsearch.spark.rdd.EsSpark$$anonfun$doSaveToEs$1.apply(EsSpark.scala:84) at org.elasticsearch.spark.rdd.EsSpark$$anonfun$doSaveToEs$1.apply(EsSpark.scala:84) at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66) at org.apache.spark.scheduler.Task.run(Task.scala:89) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) at java.lang.Thread.run(Thread.java:745) Caused by: org.elasticsearch.hadoop.rest.EsHadoopTransportException: java.net.BindException: Address already in use at org.elasticsearch.hadoop.rest.NetworkClient.execute(NetworkClient.java:121) at org.elasticsearch.hadoop.rest.RestClient.execute(RestClient.java:434) at org.elasticsearch.hadoop.rest.RestClient.execute(RestClient.java:414) at org.elasticsearch.hadoop.rest.RestClient.execute(RestClient.java:418) at org.elasticsearch.hadoop.rest.RestClient.get(RestClient.java:122) at org.elasticsearch.hadoop.rest.RestClient.esVersion(RestClient.java:564) at org.elasticsearch.hadoop.rest.InitializationUtils.discoverEsVersion(InitializationUtils.java:184) ... 10 more Caused by: java.net.BindException: Address already in use at java.net.PlainSocketImpl.socketBind(Native Method) at java.net.AbstractPlainSocketImpl.bind(AbstractPlainSocketImpl.java:376) at java.net.Socket.bind(Socket.java:631) at java.net.Socket.(Socket.java:423) at java.net.Socket.(Socket.java:280) at org.apache.commons.httpclient.protocol.DefaultProtocolSocketFactory.createSocket(DefaultProtocolSocketFactory.java:80) at org.apache.commons.httpclient.protocol.DefaultProtocolSocketFactory.createSocket(DefaultProtocolSocketFactory.java:122) at org.apache.commons.httpclient.HttpConnection.open(HttpConnection.java:707) at org.apache.commons.httpclient.HttpMethodDirector.executeWithRetry(HttpMethodDirector.java:387) at org.apache.commons.httpclient.HttpMethodDirector.executeMethod(HttpMethodDirector.java:171) at org.apache.commons.httpclient.HttpClient.executeMethod(HttpClient.java:397) at org.apache.commons.httpclient.HttpClient.executeMethod(HttpClient.java:323) at org.elasticsearch.hadoop.rest.commonshttp.CommonsHttpTransport.execute(CommonsHttpTransport.java:468) at org.elasticsearch.hadoop.rest.NetworkClient.execute(NetworkClient.java:104) ... 16 more I have updated the driver elasticsearch-hadoop-2.3.6.jar. I have read this post as well [https://github.com/elastic/elasticsearch-hadoop/issues/570](https://github.com/elastic/elasticsearch-hadoop/issues/570) I'm working with the last version for elasticsearch-hadoop I seems that I got the error when there are a lot of batches from Spark. The code is very simple: ``` def createContext(checkpoint: String, kafkaBrokers: Map[String, String], topic: Set[String], esHost: String, indexName: String, batchTime : Int, environment: String): StreamingContext = { val sparkConf = new SparkConf() sparkConf.set("spark.es.index.auto.create", "true") sparkConf.set("es.batch.size.entries", "500") //I tried to tune this params to fix this situation, but it doesn't work. sparkConf.set("es.batch.write.retry.count", "10") sparkConf.set("es.batch.write.retry.wait", "60") val ssc = new StreamingContext(sparkConf, Seconds(batchTime)) ssc.checkpoint(checkpoint) val directKafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaBrokers, topic) directKafkaStream.foreachRDD { rdd => val message = parseLog(rdd, environment) //Just some random parsing. message.saveJsonToEs(indexName + "_{time:YYYY.MM.dd}/logs") } ssc ``` } I have four nodes in my ES cluster and 4 executors with 2 cores each one. ---
**Author:** ![feuyeux](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/feuyeux/32/6139_2.png) [@feuyeux](https://discuss.elastic.co/u/feuyeux)\ **Post date:** [January 22, 2017, 11:51am UTC](https://discuss.elastic.co/t/error-address-already-in-use-when-i-use-elasticsearch-with-spark/57812/2 "2017-01-22T11:51:54Z")
I just met the same issue, and my implementation is mostly same with Guillermo: ```auto messages.foreachRDD(rdd => save2Es(rdd, Map( ConfigurationOptions.ES_RESOURCE -> { logParam.esIndex + "-" + LocalDate.now + "/" + logParam.esType }), false)) ... } def save2Es(rdd: RDD[_], cfg: Map[String, String], hasMeta: Boolean) = EsSpark.doSaveToEs(rdd, cfg, hasMeta) ``` And, I'm using Spark 1.5.1 and Elasticsearch5.0.2, and the elastic-spark is: ```xml org.elasticsearch elasticsearch-spark-13_2.10 5.0.2 ``` ---
**Author:** ![james.baiera](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/james.baiera/32/10209_2.png) [@james.baiera](https://discuss.elastic.co/u/james.baiera)\ **Post date:** [January 25, 2017, 8:53pm UTC](https://discuss.elastic.co/t/error-address-already-in-use-when-i-use-elasticsearch-with-spark/57812/3 "2017-01-25T20:53:36Z")
I would avoid saving SparkStreaming data using `doSaveToEs` with the `foreachRDD` method. This causes the connector to run normally and the rate at which the connector opens connections to Elasticsearch when running in a streaming context causes the spark workers to run out of connection resources to use; The connector is constantly setting up and tearing down connections every few seconds for each RDD that comes in. To fix this, [I recommend using the SparkStreaming native support in the connector](https://www.elastic.co/guide/en/elasticsearch/hadoop/current/spark.html#spark-streaming). The SparkStreaming native support has facilities built into it to enable safe connection pooling for multiple jobs in parallel. This should mitigate connection resource exhaustion during streaming job executions. ---
**Author:** ![feuyeux](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/feuyeux/32/6139_2.png) [@feuyeux](https://discuss.elastic.co/u/feuyeux)\ **Post date:** [February 3, 2017, 7:31am UTC](https://discuss.elastic.co/t/error-address-already-in-use-when-i-use-elasticsearch-with-spark/57812/4 "2017-02-03T07:31:41Z")
I think the reason using doSaveToEs in foreachRDD is to change the resource which may be named daily, in streaming API, the resource just be execute once at the beginning on spark driver, and spark executor cannot change it every time when DStream is coming. ---
**Author:** ![james.baiera](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/james.baiera/32/10209_2.png) [@james.baiera](https://discuss.elastic.co/u/james.baiera)\ **Post date:** [February 3, 2017, 7:38pm UTC](https://discuss.elastic.co/t/error-address-already-in-use-when-i-use-elasticsearch-with-spark/57812/5 "2017-02-03T19:38:55Z")
> [@feuyeux](#): > > I think the reason using doSaveToEs in foreachRDD is to change the resource which may be named daily, in streaming API, the resource just be execute once at the beginning on spark driver, and spark executor cannot change it every time when DStream is coming. @feuyeux A possible fix for this is: 1. You could generate a field on your documents (lets call it "indexDate") 2. Populate it with the date of the index you want to ingest each document into (using a map call and a function that strips the time element off a date, let's say) 3. Configure the load function with a [dynamic index name](https://www.elastic.co/guide/en/elasticsearch/hadoop/current/configuration.html#cfg-multi-writes) for the resource property so that the new field determines the index name on each document. 4. Setting `es.mapping.exclude` [property](https://www.elastic.co/guide/en/elasticsearch/hadoop/current/configuration.html#cfg-mapping) to reference the new field so that it does not show up in your final document that is sent to Elasticsearch. ```auto // Assuming that you are using the regular Stream[Map]: stream .map(doc => doc + ("indexDate" -> determineIndexDate(doc("datetime")))) .saveToEs("index-{indexDate}/type", Map("es.mapping.exclude" -> "indexDate")) ``` ---
**Author:** ![feuyeux](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/feuyeux/32/6139_2.png) [@feuyeux](https://discuss.elastic.co/u/feuyeux)\ **Post date:** [February 21, 2017, 11:54am UTC](https://discuss.elastic.co/t/error-address-already-in-use-when-i-use-elasticsearch-with-spark/57812/6 "2017-02-21T11:54:07Z")
Hi James, Thanks a lot, and your solution works for me. I like to share my code: ```scala val conf = logEnv.getSparkConf(YOUR_APP_NAME) conf.set(ConfigurationOptions.ES_NODES, commonParam.esNodes) conf.set(ConfigurationOptions.ES_PORT, commonParam.esPort) conf.set(ConfigurationOptions.ES_INDEX_AUTO_CREATE, true.toString) /*The es.mapping.exclude feature is ignored when es.input.json is specified*/ conf.set(ConfigurationOptions.ES_INPUT_JSON, false.toString) conf.set(ConfigurationOptions.ES_MAPPING_EXCLUDE, "logIdx") conf.set(ConfigurationOptions.ES_MAPPING_ID, "logId") val ssc = new StreamingContext(conf, Durations.seconds(commonParam.duration)) for (logParam <- logParams) { val lines = YOUR_BIZ_Utils.createStream(...).cache() val messages = lines.map(line => { try { val message: String = new String(line, Charsets.UTF_8) val traceLog: YOUR_POJO = JSON.parseObject(message, classOf[YOUR_POJO]) val logId: String = traceLog.YOUR_BIZ_KEY + "_" + traceLog.ANOTHER_BIZ_KEY val log: Map[String, String] = Map( "logId" -> logId, "logIdx" -> LocalDate.now.toString, "tag" -> traceLog.tag, ...) log } catch { case e: Exception => { e.printStackTrace() null } } }).filter(_ != null).cache() EsSparkStreaming.saveToEs(messages, logParam.esIndex + "-{logIdx}/" + logParam.esType, /*per task instance, multiplied at runtime by the total number of Hadoop tasks running*/ Map(ConfigurationOptions.ES_BATCH_SIZE_BYTES -> "1mb", ConfigurationOptions.ES_BATCH_SIZE_ENTRIES -> logParam.batch)) } ``` If you wanna provide `ES_MAPPING_EXCLUDE` to exclude some fields, you should set `ES_INPUT_JSON` to `false`, as the comments said that: > The es.mapping.exclude feature is ignored when es.input.json is specified Accordingly, you should save data using key-value style, instead of json serialized from POJO object. Most important thing in indexing process is to set `ES_BATCH_SIZE_ENTRIES` parameter, which is 1000 as default. In my opinion, it's an art for tradeoff duration and task batch, the cost time for indexing the stream should just smaller than the duration time. So, it's also associated with the number of the executor instances. ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [July 6, 2017, 1:22pm UTC](https://discuss.elastic.co/t/error-address-already-in-use-when-i-use-elasticsearch-with-spark/57812/7 "2017-07-06T13:22:30Z")