# Upsert ELasticSearch documents with Spark

**URL:** <https://discuss.elastic.co/t/upsert-elasticsearch-documents-with-spark/107716>\
**Category:** Elasticsearch\
**Tags:** es-hadoop\
**Created:** [November 15, 2017, 10:10am UTC](https://discuss.elastic.co/t/upsert-elasticsearch-documents-with-spark/107716 "2017-11-15T10:10:31Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![salvob41](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/salvob41/32/24222_2.png) [@salvob41](https://discuss.elastic.co/u/salvob41)\
**Post date:** [November 15, 2017, 10:10am UTC](https://discuss.elastic.co/t/upsert-elasticsearch-documents-with-spark/107716/1 "2017-11-15T10:10:31Z")

</div>

Is there an API with Spark, Spark Streaming or Structured Streaming Spark that I can use to upsert documents in ElasticSearch with a script?

Elastic-Hadoop has the `.saveToES` function but I cannot understand which parameter use in order to use the script field.

The structured streaming APIs work with dataframes. I found this API that write dataframe in ES:

```auto
queryES = df\
.writeStream \
.format("org.elasticsearch.spark.sql")\
.queryName("ESquery")\
.option("es.resource","index/doc_type") \
.option("checkpointLocation", "checkpoint").start()

```

but I cannot wrap my head around on where to specify the parameters for the upsert.

I would like to upsert document as (here a python code that insert documents if aren't in the index already or increment a counter if they are already in it):

```auto
for message in consumer:

    msg = json.loads(message.value)
    print(msg)
    index = INDEX_NAME
    es_id = msg["id"]
    script = {"script":"ctx._source.view+=1","upsert" : msg}
    es.update(index=index, doc_type="test", id=es_id, body=script)

```

---

<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:** [December 13, 2017, 10:21am UTC](https://discuss.elastic.co/t/upsert-elasticsearch-documents-with-spark/107716/2 "2017-12-13T10:21:11Z")

</div>

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