# Aggregate filter plugin

**URL:** <https://discuss.elastic.co/t/aggregate-filter-plugin/119632>\
**Category:** Logstash\
**Created:** [February 13, 2018, 12:30pm UTC](https://discuss.elastic.co/t/aggregate-filter-plugin/119632 "2018-02-13T12:30:14Z")\
**Posts on this page:** 12\
**Page:** 1

<div class="post-metadata">

**Author:** ![scch](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/scch/32/25504_2.png) [@scch](https://discuss.elastic.co/u/scch)\
**Post date:** [February 13, 2018, 12:30pm UTC](https://discuss.elastic.co/t/aggregate-filter-plugin/119632/1 "2018-02-13T12:30:14Z")

</div>

| occurred\_on | Bytes\_Out | user\_id |
| --- | --- | --- |
| 2017-12-31T10:25:49.000-0600 | 1718 | ts2341 |
| 2017-12-31T10:25:08.000-0600 | 1070 | ts2341 |
| 2017-12-31T10:25:08.000-0600 | 355 | ts2341 |
| 2017-12-29T16:55:01.000-0600 | 255 | gs2121 |
| 2017-12-29T16:55:01.000-0600 | 255 | gs2121 |
| 2017-12-29T16:55:01.000-0600 | 255 | gs2121 |
| 2017-12-29T16:01:53.000-0600 | 2618803 | gs2121 |
| 2017-12-29T16:01:53.000-0600 | 1684 | gs2121 |
| 2017-12-29T16:01:53.000-0600 | 1351 | gs2121 |
| 2017-12-29T16:01:53.000-0600 | 95323 | gs2121 |
| 2017-12-29T16:01:52.000-0600 | 3500 | gs2121 |
| 2017-12-29T16:01:52.000-0600 | 255 | gs2121 |

Want to calculate and ingest, how much data is sent outside on daily basis. So i am trying to aggregate the logs on the basis of "userid" , sum the "Bytes\_out" column on daily basis.  
Request code/help on how to achieve this.

---

<div class="post-metadata">

**Author:** ![paz](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/paz/32/28003_2.png) [@paz](https://discuss.elastic.co/u/paz)\
**Post date:** [February 13, 2018, 12:40pm UTC](https://discuss.elastic.co/t/aggregate-filter-plugin/119632/2 "2018-02-13T12:40:36Z")

</div>

I assume you've had a look at the [aggregate filter](https://www.elastic.co/guide/en/logstash/current/plugins-filters-aggregate.html)? If so, do you have any configuration in place so far?

---

<div class="post-metadata">

**Author:** ![scch](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/scch/32/25504_2.png) [@scch](https://discuss.elastic.co/u/scch)\
**Post date:** [February 13, 2018, 1:47pm UTC](https://discuss.elastic.co/t/aggregate-filter-plugin/119632/3 "2018-02-13T13:47:16Z")

</div>

Hi Paz,  
Tried below...with no success, also i do not have idea about ruby ..  
filter {  
csv {  
separator =\> ","  
columns =\> ["occuredon","bytes\_out","userid"]  
}  
mutate {  
lowercase =\> ["userid"]  
convert =\> ["bytes\_out","integer"]  
}  
date {  
match =\> ["occuredon", "yyyy/mm/dd HH:mm:ss", "ISO8601"]  
target =\> "@timestamp"  
}  
aggregate {  
task\_id =\> "%{userid}"  
code =\> "map['sum'] += event.get(‘bytes\_out’);"  
push\_map\_as\_event\_on\_timeout =\> true  
timeout\_task\_id\_field =\> "userid"  
timeout =\> 3600 # 1hr timeout  
timeout\_tags =\> ['\_aggregatetimeout']  
timeout\_code =\> "event.set('bytesoutSum', event.get('sum'))"  
}

```
             }

             output {
                       stdout {codec => rubydebug}
                        }

```

###codec output####

"tags" =\> [  
[0] "\_aggregateexception"  
],

---

<div class="post-metadata">

**Author:** ![paz](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/paz/32/28003_2.png) [@paz](https://discuss.elastic.co/u/paz)\
**Post date:** [February 13, 2018, 2:01pm UTC](https://discuss.elastic.co/t/aggregate-filter-plugin/119632/4 "2018-02-13T14:01:30Z")

</div>

Hmm, that does sound like a code error indeed. Btw, it looks like that _bytes\_out_ is enclosed in some weird quotes. Can you try this one?

```auto
aggregate {
    task_id => "%{userid}"
    code => "map['bytesoutSum'] ||= 0 ; map['bytesoutSum'] += event.get('bytes_out').to_i"
    push_map_as_event_on_timeout => true
    timeout_task_id_field => "userid"
    timeout => 3600 # 1hr timeout
    timeout_tags => ['_aggregatetimeout']
}
```

---

<div class="post-metadata">

**Author:** ![scch](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/scch/32/25504_2.png) [@scch](https://discuss.elastic.co/u/scch)\
**Post date:** [February 13, 2018, 2:24pm UTC](https://discuss.elastic.co/t/aggregate-filter-plugin/119632/5 "2018-02-13T14:24:01Z")

</div>

I am not getting error "tags" =\> [[0] "\_aggregateexception"],

however i am not getting consolidated logs for an hour. Below is the output of ruby codec.

{  
"@timestamp" =\> 2017-12-31T12:47:53.000Z,  
"host" =\> "[redes6.Delta.com](http://redes6.Delta.com)",  
"bytes\_out" =\> 1925,  
"path" =\> "/tmp/itd/msfs/MSFS2.csv",  
"@version" =\> "1",  
"userid" =\> "cd41763",  
"occured\_on" =\> "2017-12-31T06:47:53.000-0600",  
"message" =\> "2017-12-31T06:47:53.000-0600,1925,cd41763\r"  
}  
{  
"@timestamp" =\> 2017-12-31T12:48:23.000Z,  
"host" =\> "[redes6.Delta.com](http://redes6.Delta.com)",  
"bytes\_out" =\> 2669,  
"path" =\> "/tmp/itd/msfs/MSFS2.csv",  
"@version" =\> "1",  
"userid" =\> "cd41763",  
"occured\_on" =\> "2017-12-31T06:48:23.000-0600",  
"message" =\> "2017-12-31T06:48:23.000-0600,2669,cd41763\r"  
}  
{  
"@timestamp" =\> 2017-12-31T12:48:24.000Z,  
"host" =\> "[redes6.Delta.com](http://redes6.Delta.com)",  
"bytes\_out" =\> 1888,  
"path" =\> "/tmp/itd/msfs/MSFS2.csv",  
"@version" =\> "1",  
"userid" =\> "cd41763",  
"occured\_on" =\> "2017-12-31T06:48:24.000-0600",  
"message" =\> "2017-12-31T06:48:24.000-0600,1888,cd41763\r"

---

<div class="post-metadata">

**Author:** ![paz](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/paz/32/28003_2.png) [@paz](https://discuss.elastic.co/u/paz)\
**Post date:** [February 13, 2018, 2:34pm UTC](https://discuss.elastic.co/t/aggregate-filter-plugin/119632/6 "2018-02-13T14:34:55Z")

</div>

That's expected behavior. What you see is the actual logs themselves. They still pass through to the output since you're not dropping them specifically.

The aggregation filter does not alter the original messages themselves, it just creates new ones. You should be seeing aggregated events 1 hour after Logstash started.  
More specifically, each new _user\_id's_ relevant aggregated event should spawn 1 hour after the first time you receive that _user\_id_.

---

<div class="post-metadata">

**Author:** ![scch](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/scch/32/25504_2.png) [@scch](https://discuss.elastic.co/u/scch)\
**Post date:** [February 13, 2018, 2:39pm UTC](https://discuss.elastic.co/t/aggregate-filter-plugin/119632/7 "2018-02-13T14:39:37Z")

</div>

ok... will push them to ES and check...and get back...

---

<div class="post-metadata">

**Author:** ![scch](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/scch/32/25504_2.png) [@scch](https://discuss.elastic.co/u/scch)\
**Post date:** [February 13, 2018, 3:05pm UTC](https://discuss.elastic.co/t/aggregate-filter-plugin/119632/8 "2018-02-13T15:05:28Z")

</div>

Hi Paz,  
Thanks,it is working 🙂 ....one more query , i am using .csv file as input, can i use "time stamp" present in .csv file, rather than system time, for time out...

thanks..

---

<div class="post-metadata">

**Author:** ![paz](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/paz/32/28003_2.png) [@paz](https://discuss.elastic.co/u/paz)\
**Post date:** [February 15, 2018, 1:27pm UTC](https://discuss.elastic.co/t/aggregate-filter-plugin/119632/9 "2018-02-15T13:27:37Z")

</div>

> [@scch](#):
>
> i am using .csv file as input, can i use "time stamp" present in .csv file, rather than system time, for time out...

You mean like using that timestamp to control when the aggregation will expire, or something else?

---

<div class="post-metadata">

**Author:** ![scch](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/scch/32/25504_2.png) [@scch](https://discuss.elastic.co/u/scch)\
**Post date:** [February 15, 2018, 3:04pm UTC](https://discuss.elastic.co/t/aggregate-filter-plugin/119632/10 "2018-02-15T15:04:17Z")

</div>

Yes Paz, i want to use date and time present in .csv file for time out rather system time.  
example hourly, daily, or weekly aggregation of data.

---

<div class="post-metadata">

**Author:** ![scch](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/scch/32/25504_2.png) [@scch](https://discuss.elastic.co/u/scch)\
**Post date:** [February 20, 2018, 3:49pm UTC](https://discuss.elastic.co/t/aggregate-filter-plugin/119632/11 "2018-02-20T15:49:10Z")

</div>

this one is resolved...  
for time stamp i am parsing date one more time in aggregation filter.  
and for aggregation i am using %{+d} along with user id to aggregate days data. similarly %{+ww} can be used to aggregate weakly data .

task\_id =\> "%{userid}\_%{+d}"

---

<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:** [March 20, 2018, 3:49pm UTC](https://discuss.elastic.co/t/aggregate-filter-plugin/119632/12 "2018-03-20T15:49:19Z")

</div>

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