elasticsearch-spark connector size limit parameter is ignored in query

Question

I'm trying to query elasticsearch with the elasticsearch-spark connector and I want to return only few results:

For example:

val conf = new SparkConf().set("es.nodes","localhost").set("es.index.auto.create", "true").setMaster("local")
val sparkContext = new SparkContext(conf)
val query = "{\"size\":1}"
println(sparkContext.esRDD("index_name/type", query).count())

However this will return all the documents in the index.

Answer 1

If I understand this correctly, you are executing a count operation, which does not return any documents. Do you expect it to return 1 because you specified size: 1? That's not happening, which is by design.

Edited to add: This is the definition of count() in elasticsearch-hadoop:

override def count(): Long = {
    val repo = new RestRepository(esCfg)
    try {
      return repo.count(true)
    } finally {
      repo.close()
    }
  }

It does not take the query into account at all, but considers the entire ES index as the RDD input.

Answer 2

This is actually on purpose. Since the connector does a parallel query, it also looks at the number of documents being returned so if the user specifies a parameter, it will overwrite it according to the es.scroll.limit setting (see the configuration option).

In other words, if you want to control the size, do so through that setting as it will always take precedence.

Beware that this parameter applies per shard. So, if you have 5 shards you might bet up to fice hits if this parameter is set to 1.

See https://www.elastic.co/guide/en/elasticsearch/hadoop/master/configuration.html

Answer 3

Some parameters are actually ignored from the query by design, such as : from, size, fields, etc.

They are used internally inside the elasticsearch-spark connector.

Unfortunately this list of unsupported parameters isn't documented. But if you wish to use the size parameter you can always rely on the pushdown predicate and use the DataFrame/Dataset limit method.

So you ought using the Spark SQL DSL instead e.g :

val df = sqlContext.read.format("org.elasticsearch.spark.sql")
                        .option("pushdown","true")
                        .load("index_name/doc_type")
                        .limit(10) // instead of size : 10

This query will return the first 10 documents returned by the match_all query that is used by default by the connector.

Note: The following isn't correct on any level.

This is actually on purpose. Since the connector does a parallel query, it also looks at the number of documents being returned so if the user specifies a parameter, it will overwrite it according to the es.scroll.limit setting (see the configuration option).

When you query elasticsearch it also run the query in parallel on all the index shards without overwriting them.