# Is there a way to use dynamic indexing through “es.index.write = {index}/type” in writeStream method in spark sturtured streaming?

**URL:** <https://discuss.elastic.co/t/is-there-a-way-to-use-dynamic-indexing-through-es-index-write-index-type-in-writestream-method-in-spark-sturtured-streaming/174334>\
**Category:** Elasticsearch\
**Created:** [March 28, 2019, 12:44pm UTC](https://discuss.elastic.co/t/is-there-a-way-to-use-dynamic-indexing-through-es-index-write-index-type-in-writestream-method-in-spark-sturtured-streaming/174334 "2019-03-28T12:44:55Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![Fahad\_Khan](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/fahad_khan/32/42682_2.png) [@Fahad\_Khan](https://discuss.elastic.co/u/Fahad_Khan)\
**Post date:** [March 28, 2019, 12:44pm UTC](https://discuss.elastic.co/t/is-there-a-way-to-use-dynamic-indexing-through-es-index-write-index-type-in-writestream-method-in-spark-sturtured-streaming/174334/1 "2019-03-28T12:44:56Z")

</div>

I'm using spark structured streaming to read so many files recursively and writing them into elasticsearch, I want to know is there a way to create index at runtime according to the file name spark reading from folder. suppose spark read file from client1 dir so it should index it in elasticsearch as client1.

I've read this [https://www.elastic.co/guide/en/elasticsearch/hadoop/master/configuration.html#cfg-multi-writes](https://www.elastic.co/guide/en/elasticsearch/hadoop/master/configuration.html#cfg-multi-writes), they did quite easily but the problem is I'm using spark structured streaming, in which there is no SavetoEs method instead there is WriteStream method, What i want to do is that i want to read files recursively from different directories and when i'll write them into elasticsearch i want to index them in elasticsearch as i-e file:/home/usr/client/message/single so it should be indexed like "client-message-single" in elasticsearh based on file path.

```
file_path= file://home/usr/client/email/message/singlemesage

file_split_path= file_path.split('/')

index = str(file_split_path[-4] + '-' + file_split_path[-3] + '-' + file_split_path[-2])

myjson['doc-index'] =index //adding key value in my json

df=myjson.writeStream.option("es.resource.write","{doc-index}/default").outputMode(""append").format("org.elasticsearch.spark.sql").start() df.awaitTermination()

```

while using above technique I'm getting following error

org.elasticsearch.hadoop.EsHadoopIllegalArgumentException: Cannot find match for {doc-index}/default at org.elasticsearch.hadoop.serialization.bulk.BulkEntryWriter.writeBulkEntry(BulkEntryWriter.java:136) at org.elasticsearch.hadoop.rest.RestRepository.writeToIndex(RestRepository.java:170) at org.elasticsearch.spark.rdd.EsRDDWriter.write(EsRDDWriter.scala:74) at org.elasticsearch.spark.sql.streaming.EsStreamQueryWriter.run(EsStreamQueryWriter.scala:41) at org.elasticsearch.spark.sql.streaming.EsSparkSqlStreamingSink$$anonfun$addBatch$2$$anonfun$2.apply(EsSparkSqlStreamingSink.scala:52) at org.elasticsearch.spark.sql.streaming.EsSparkSqlStreamingSink$$anonfun$addBatch$2$$anonfun$2.apply(EsSparkSqlStreamingSink.scala:51)

there is "doc-index" : "client-message-singlemessage" in json which read by spark stuctured streaming.

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [April 25, 2019, 12:44pm UTC](https://discuss.elastic.co/t/is-there-a-way-to-use-dynamic-indexing-through-es-index-write-index-type-in-writestream-method-in-spark-sturtured-streaming/174334/2 "2019-04-25T12:44:56Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.