Как обновить или частично обновить документы сценария в ElasticSearch с помощью Spark?

У меня есть псевдокод на питоне, который читает из потока Kafka и обновляет документы в Elasticsearch (увеличивая счетчик view, если документ уже существует.

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)

Поскольку я хочу использовать его в распределенной среде, я использую Spark Structured Streaming.

df.writeStream \
.format("org.elasticsearch.spark.sql")\
.queryName("ESquery")\
.option("es.resource","credentials/url") \
.option("checkpointLocation", "checkpoint").start()

или SparkStreaming в scala, который читает из KafkaStream:

// Initializing Spark Streaming Context and kafka stream
sparkConf.setMaster("local[2]")
val ssc = new StreamingContext(sparkConf, Seconds(10))
[...] 
val messages = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topicsSet, kafkaParams)
    )

[...]
val urls = messages.map(record => JsonParser.parse(record.value()).values.asInstanceOf[Map[String, Any]])
urls.saveToEs("credentials/credential")

.saveToEs(...) — это API elastic-hadoop.jar, задокументированный здесь. К сожалению, этот репозиторий не очень хорошо документирован. Поэтому я не могу понять, где я могу разместить команду сценария.

Кто-нибудь может мне помочь? заранее спасибо