# Read ES Index from Spark Executors **URL:** \ **Category:** Elasticsearch\ **Tags:** es-hadoop\ **Created:** [May 16, 2016, 4:09am UTC](https://discuss.elastic.co/t/read-es-index-from-spark-executors/50090 "2016-05-16T04:09:29Z")\ **Posts on this page:** 5\ **Page:** 1
**Author:** ![NiravVira](https://avatars.discourse-cdn.com/v4/letter/n/ea666f/32.png) [@NiravVira](https://discuss.elastic.co/u/NiravVira)\ **Post date:** [May 16, 2016, 4:09am UTC](https://discuss.elastic.co/t/read-es-index-from-spark-executors/50090/1 "2016-05-16T04:09:29Z")
**Infrastructure** ES: 2.2.0 Spark: 1.6 Scala: 2.10 ElasticHadoop: elasticsearch-hadoop 2.2.0 ( i can use elasticsearch-spark-2.10 if needed) Kafka: 0.9.0 Host: CDH5.7 VM Spark Streaming Job Desc: Read data from Kafka topic. for unique Id, read ES index/type data from this correlation, aggregate & store in another index/type for search. SparkContext is created on driver along with ES settings. able to read ES fine as long as I do _.collect_ & send the execution to Spark Driver. **Question:** How to make the read from Spark Executor so I can leverage the parallelism. Googling suggests use of ConnectionPool at rdd partition level for connection to likes of Cassandra, etc. How to get a connection pool going for ES, any concrete impl / documentation for connection pool that can be used on Executor. _ **code snippet** _ > ``` > val stream: InputDStream[(String, String)] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, String)](ssc, kafkaParams, offsets, messageHandler) > > ``` > ``` > stream > .foreachRDD { rdd => > if (!rdd.isEmpty) { > ... > /* //this works fine so connectivity from driver to ES is good > val myrdd = sc.esRDD("somindex/sometype ", "?q=EntityId:993b0000-e516-6c3b-79a9-08d349e3fd34") > myrdd.foreach(x => println("item details" + x._2)) > */ > rdd.collect().foreach(item => { > ... some operation > val entityId = kafkaMessage.entityId //kafkaMessage resolved thru code not shown here > val esDataRDD = sc.esRDD("someindex/sometype", "?q=EntityId:"+entityId) > > ``` ---
**Author:** ![Nicolas\_Phung](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/nicolas_phung/32/694_2.png) [@Nicolas\_Phung](https://discuss.elastic.co/u/Nicolas_Phung)\ **Post date:** [May 20, 2016, 4:45pm UTC](https://discuss.elastic.co/t/read-es-index-from-spark-executors/50090/2 "2016-05-20T16:45:05Z")
Hello, I'm interested too if this best practices can be applied with elasticsearch-spark. It looks like we are running into some troubles with several Spark Streaming to write in Elasticsearch with elasticsearch-spark 2.3.1 : _ **code snippet** _ > [@](#): > > 16/05/19 14:59:19 WARN TaskSetManager: Lost task 2.0 in stage 1122.0 (TID 11389, slave04.local): 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:190) > 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$saveToEs$1.apply(EsSpark.scala:67) > at org.elasticsearch.spark.rdd.EsSpark$$anonfun$saveToEs$1.apply(EsSpark.scala:67) > 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:1142) > at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617 > at java.lang.Thread.run(Thread.java:745) > Caused by: org.elasticsearch.hadoop.rest.EsHadoopTransportException: java.net.BindException: Adresse déjà utilisée > 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:178) > ... 10 more > Caused by: java.net.BindException: Adresse déjà utilisée > at java.net.PlainSocketImpl.socketBind(Native Method) > at java.net.AbstractPlainSocketImpl.bind(AbstractPlainSocketImpl.java:382) > at java.net.Socket.bind(Socket.java:644) > at java.net.Socket.(Socket.java:433) > at java.net.Socket.(Socket.java:286) > 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 This happens when we have many Spark Streaming jobs and it seems to impact existing jobs that uses elasticsearch-spark to write into Elasticsearch. Does someone has any idea ? Regards, ---
**Author:** ![costin](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/costin/32/44950_2.png) [@costin](https://discuss.elastic.co/u/costin)\ **Post date:** [May 24, 2016, 10:28am UTC](https://discuss.elastic.co/t/read-es-index-from-spark-executors/50090/3 "2016-05-24T10:28:26Z")
Streaming is a special case since conceptually it's a long running process but Spark is actually batching so the whole micro-batching strategy means a series of small batch process/tasks which keep on writing to ES. As there's no proper API for that, the Spark docs suggests to keep the _connections_ alive and pass them to the batch in order to keep creating them. Which works if there's a programmatic approach (open, connection, write some data, check, close) but fails with a declarative one (take this data and save it). Spark 2 looks to be introducing/changing some APIs and once it is finalized, we'll look into providing hooks for reusing connections. ---
**Author:** ![NiravVira](https://avatars.discourse-cdn.com/v4/letter/n/ea666f/32.png) [@NiravVira](https://discuss.elastic.co/u/NiravVira)\ **Post date:** [May 25, 2016, 5:04am UTC](https://discuss.elastic.co/t/read-es-index-from-spark-executors/50090/4 "2016-05-25T05:04:37Z")
Costin, Thanks for your reply. My use case is about "read" ES as well. I am wondering if there are current options (w or w/o connection pool) to call .esRDD without collect & there by running the code on spark executors. Any samples are greatly appreciated. ---
**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:24pm UTC](https://discuss.elastic.co/t/read-es-index-from-spark-executors/50090/6 "2017-07-06T13:24:22Z")