# ElasticSearch spark esRDD not returing the aggregate values in aggregated query

**URL:** <https://discuss.elastic.co/t/elasticsearch-spark-esrdd-not-returing-the-aggregate-values-in-aggregated-query/19846>\
**Category:** Elasticsearch\
**Created:** [September 17, 2014, 1:13pm UTC](https://discuss.elastic.co/t/elasticsearch-spark-esrdd-not-returing-the-aggregate-values-in-aggregated-query/19846 "2014-09-17T13:13:37Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![siva\_pradeep](https://avatars.discourse-cdn.com/v4/letter/s/ecccb3/32.png) [@siva\_pradeep](https://discuss.elastic.co/u/siva_pradeep)\
**Post date:** [September 17, 2014, 1:13pm UTC](https://discuss.elastic.co/t/elasticsearch-spark-esrdd-not-returing-the-aggregate-values-in-aggregated-query/19846/1 "2014-09-17T13:13:37Z")

</div>

Hi,

I have a query which filters the rows and then applies the aggregation. I  
tried running the query in "Sense" it gave me the expected result. But when  
I try to run the same query using elasticsearch-spark\_2.10 I get the rows  
filtered by the query but not the aggregation result. I am sure I am  
missing some thing but unable to figure out that.

Here is the query

GET \_search  
{  
"query" : {  
"bool": {  
"must": [  
{  
"filtered": {  
"query": {  
"range": {  
"@timestamp": {  
"from": "2014-09-03T01:40:37.437Z",  
"to": "2014-09-03T01:45:11.437Z"  
}  
}  
}  
}  
}  
]  
}  
},

"size": 0,

"fields": ["cid","entity"],  
"aggs": {  
"cid": {  
"terms": {  
"field": "cid",  
"min\_doc\_count": 2,  
"size": 100  
},

```
  "aggs": {
    "tn": {
      "terms": {
        "field": "entity"
      }
    }
  }
}

```

}  
}

Query Result:

{  
"took": 10005,  
"timed\_out": false,  
"\_shards": {  
"total": 10,  
"successful": 10,  
"failed": 0  
},  
"hits": {  
"total": 2430,  
"max\_score": 0,  
"hits": []  
},  
"aggregations": {  
"cid": {  
"buckets": [  
{  
"key": " [01abcecc9a20cd3d6ae6be3509d014ba@76.96.107.168](mailto:01abcecc9a20cd3d6ae6be3509d014ba@76.96.107.168)",  
"doc\_count": 2,  
"tn": {  
"buckets": [  
{  
"key": "15052563268",  
"doc\_count": 2  
}  
]  
}  
}  
]  
}  
}  
}

Spark program :

object PresenceFilter extends App {

val query: String = "{\n\n "query" : {\n\n "bool": {\n\n  
"must": [\n\n {\n\n "filtered": {\n\n  
"query": {\n\n "range": {\n\n  
"@timestamp": {\n\n "from":  
"2014-09-03T01:40:37.437Z",\n\n "to":  
"2014-09-03T01:45:11.437Z"\n\n }\n\n  
}\n\n }\n\n }\n\n }\n\n ]\n\n }\n\n  
},\n \n "size": 0,\n \n "fields": ["cid","entity"],\n\n  
"aggs": {\n\n "cid": {\n\n "terms": {\n\n "field":  
"cid",\n\n "min\_doc\_count": 2,\n\n "size": 100\n\n  
},\n \n "aggs": {\n\n "tn": {\n\n "terms":  
{\n\n "field": "entity"\n\n }\n\n }\n\n  
}\n\n }\n }\n\n}"

val sparkConf = new SparkConf()  
.setAppName("PresenceAnalysis")  
.setMaster("local[4]")  
.set("es.nodes", "[prs-wch-10.sys.comcast.net](http://prs-wch-10.sys.comcast.net)")  
.set("es.port", "9200")  
.set("es.resource", s"presence-2014.09.03/presence")  
.set("es.endpoint", "\_search")  
// .set("es.query", query)  
val sc = new SparkContext(sparkConf)

sc.esRDD.count returns 2430 rows

How do I get the aggregation part (the following part) of the result into  
the program

"aggregations": {  
"cid": {  
"buckets": [  
{  
"key": " [01abcecc9a20cd3d6ae6be3509d014ba@76.96.107.168](mailto:01abcecc9a20cd3d6ae6be3509d014ba@76.96.107.168)",  
"doc\_count": 2,  
"tn": {  
"buckets": [  
{  
"key": "15052563268",  
"doc\_count": 2  
}  
]  
}  
}

Please advise.

Thanks,  
Siva P

--  
You received this message because you are subscribed to the Google Groups "elasticsearch" group.  
To unsubscribe from this group and stop receiving emails from it, send an email to [elasticsearch+unsubscribe@googlegroups.com](mailto:elasticsearch+unsubscribe@googlegroups.com).  
To view this discussion on the web visit [https://groups.google.com/d/msgid/elasticsearch/d75bf6db-58c5-4ca8-b8d2-f45a3dd0a339%40googlegroups.com](https://groups.google.com/d/msgid/elasticsearch/d75bf6db-58c5-4ca8-b8d2-f45a3dd0a339%40googlegroups.com).  
For more options, visit [https://groups.google.com/d/optout](https://groups.google.com/d/optout).

---

<div class="post-metadata">

**Author:** ![Jeff\_Steinmetz](https://avatars.discourse-cdn.com/v4/letter/j/ecd19e/32.png) [@Jeff\_Steinmetz](https://discuss.elastic.co/u/Jeff_Steinmetz)\
**Post date:** [October 17, 2014, 3:46pm UTC](https://discuss.elastic.co/t/elasticsearch-spark-esrdd-not-returing-the-aggregate-values-in-aggregated-query/19846/2 "2014-10-17T15:46:18Z")

</div>

Siva,

Try the latest build of elasticsearch-hadoop, ver 2.1.0 Beta 2

> **[Elasticsearch Platform — Find real-time answers at scale](https://www.elastic.co)**
>
> Power insights and outcomes with the Elasticsearch Platform and AI. See into your data and find answers that matter with enterprise solutions designed to help you build, observe, and protect. Try Elasticsearch free today.

The esRDD has been changed to sparks PairRDD  
[https://spark.apache.org/docs/1.1.0/api/scala/index.html#org.apache.spark.rdd.PairRDDFunctions](https://spark.apache.org/docs/1.1.0/api/scala/index.html#org.apache.spark.rdd.PairRDDFunctions)

The RDD will now be key/value (tuples) that look like (String, Map[String,  
ANY])

so you could start to walk the json key/value hierarchy with something like:

esRDD.flatMap { args =\> args.\_2.get("aggregations") }

(the syntax above is not exact, since your specific query result may have a  
different first key/value pair as the first object )

Best,  
Jeff Steinmetz  
Director of Data Science  
Ekho, Inc.

> **[Ekho](http://www.ekho.me)**
>
> Ekho

@jeffsteinmetz

On Wednesday, September 17, 2014 6:13:37 AM UTC-7, siva pradeep wrote:

> Hi,
> 
> I have a query which filters the rows and then applies the aggregation. I  
> tried running the query in "Sense" it gave me the expected result. But when  
> I try to run the same query using elasticsearch-spark\_2.10 I get the rows  
> filtered by the query but not the aggregation result. I am sure I am  
> missing some thing but unable to figure out that.
> 
> Here is the query
> 
> GET \_search  
> {  
> "query" : {  
> "bool": {  
> "must": [  
> {  
> "filtered": {  
> "query": {  
> "range": {  
> "@timestamp": {  
> "from": "2014-09-03T01:40:37.437Z",  
> "to": "2014-09-03T01:45:11.437Z"  
> }  
> }  
> }  
> }  
> }  
> ]  
> }  
> },
> 
> "size": 0,
> 
> "fields": ["cid","entity"],  
> "aggs": {  
> "cid": {  
> "terms": {  
> "field": "cid",  
> "min\_doc\_count": 2,  
> "size": 100  
> },
> 
> ```
> "aggs": {
> "tn": {
> "terms": {
> "field": "entity"
> }
> }
> }
> }
> 
> ```
> 
> }  
> }
> 
> Query Result:
> 
> {  
> "took": 10005,  
> "timed\_out": false,  
> "\_shards": {  
> "total": 10,  
> "successful": 10,  
> "failed": 0  
> },  
> "hits": {  
> "total": 2430,  
> "max\_score": 0,  
> "hits":   
> },  
> "aggregations": {  
> "cid": {  
> "buckets": [  
> {  
> "key": " [01abcecc9a20cd3d6ae6be3509d014ba@76.96.107.168](mailto:01abcecc9a20cd3d6ae6be3509d014ba@76.96.107.168)  
> \<javascript:\>",  
> "doc\_count": 2,  
> "tn": {  
> "buckets": [  
> {  
> "key": "15052563268",  
> "doc\_count": 2  
> }  
> ]  
> }  
> }  
> ]  
> }  
> }  
> }
> 
> Spark program :
> 
> object PresenceFilter extends App {
> 
> val query: String = "{\n\n "query" : {\n\n "bool": {\n\n  
> "must": [\n\n {\n\n "filtered": {\n\n  
> "query": {\n\n "range": {\n\n  
> "@timestamp": {\n\n "from":  
> "2014-09-03T01:40:37.437Z",\n\n "to":  
> "2014-09-03T01:45:11.437Z"\n\n }\n\n  
> }\n\n }\n\n }\n\n }\n\n ]\n\n }\n\n  
> },\n \n "size": 0,\n \n "fields": ["cid","entity"],\n\n  
> "aggs": {\n\n "cid": {\n\n "terms": {\n\n "field":  
> "cid",\n\n "min\_doc\_count": 2,\n\n "size": 100\n\n  
> },\n \n "aggs": {\n\n "tn": {\n\n "terms":  
> {\n\n "field": "entity"\n\n }\n\n }\n\n  
> }\n\n }\n }\n\n}"
> 
> val sparkConf = new SparkConf()  
> .setAppName("PresenceAnalysis")  
> .setMaster("local[4]")  
> .set("es.nodes", "[prs-wch-10.sys.comcast.net](http://prs-wch-10.sys.comcast.net)")  
> .set("es.port", "9200")  
> .set("es.resource", s"presence-2014.09.03/presence")  
> .set("es.endpoint", "\_search")  
> // .set("es.query", query)  
> val sc = new SparkContext(sparkConf)
> 
> sc.esRDD.count returns 2430 rows
> 
> How do I get the aggregation part (the following part) of the result into  
> the program
> 
> "aggregations": {  
> "cid": {  
> "buckets": [  
> {  
> "key": " [01abcecc9a20cd3d6ae6be3509d014ba@76.96.107.168](mailto:01abcecc9a20cd3d6ae6be3509d014ba@76.96.107.168)  
> \<javascript:\>",  
> "doc\_count": 2,  
> "tn": {  
> "buckets": [  
> {  
> "key": "15052563268",  
> "doc\_count": 2  
> }  
> ]  
> }  
> }
> 
> Please advise.
> 
> Thanks,  
> Siva P

--  
You received this message because you are subscribed to the Google Groups "elasticsearch" group.  
To unsubscribe from this group and stop receiving emails from it, send an email to [elasticsearch+unsubscribe@googlegroups.com](mailto:elasticsearch+unsubscribe@googlegroups.com).  
To view this discussion on the web visit [https://groups.google.com/d/msgid/elasticsearch/01d05c62-3095-4d9e-9407-4357add26896%40googlegroups.com](https://groups.google.com/d/msgid/elasticsearch/01d05c62-3095-4d9e-9407-4357add26896%40googlegroups.com).  
For more options, visit [https://groups.google.com/d/optout](https://groups.google.com/d/optout).

---

<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:** [July 6, 2017, 12:55am UTC](https://discuss.elastic.co/t/elasticsearch-spark-esrdd-not-returing-the-aggregate-values-in-aggregated-query/19846/3 "2017-07-06T00:55:28Z")

</div>