# Aggregate filter - push on event size limit
**URL:** https://discuss.elastic.co/t/aggregate-filter-push-on-event-size-limit/180875
**Category:** Logstash
**Created:** [May 13, 2019, 7:53pm UTC](https://discuss.elastic.co/t/aggregate-filter-push-on-event-size-limit/180875 "2019-05-13T19:53:04Z")
**Posts on this page:** 4
**Page:** 1
### Author:  [@bosand](https://discuss.elastic.co/u/bosand)
#### Post date: [May 13, 2019, 7:53pm UTC](https://discuss.elastic.co/t/aggregate-filter-push-on-event-size-limit/180875/1 "2019-05-13T19:53:04Z")
Hi there,
I would like to aggregate events until the aggregated event reaches a certain size limit (e.g. 256kb) at which point I would like to push the event to the queue. There are no start/end events.
Push on timeout does not solve my problem as hundreds of events may unpredictably arrive in a very short time interval. Is there a way to do that?
Thanks,
bosand
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [May 13, 2019, 8:58pm UTC](https://discuss.elastic.co/t/aggregate-filter-push-on-event-size-limit/180875/2 "2019-05-13T20:58:08Z")
At first I thought not, but there is a distinctly non-scaleable way to do it. In addition to the usual requirement of "--pipeline.workers 1" you need "--pipeline.batch.size 1" so that every event goes through the second aggregate filter before the first aggregate filter processes another event.
This is just a proof-of-concept that demonstrates how it could be done.
```
filter { json { source => "message" } }
input { stdin {} }
input { generator { count => 10 lines => ['{ "id": "a", "data": "123456789012345678901234567890123456789012345678901234567890"}'] } }
filter {
aggregate {
task_id => "%{id}"
code => '
map["task"] ||= ""
map["task"] += event.get("data")
if map["task"].bytesize > 200
event.set("[@metadata][timeToFlush]", true)
end
'
push_map_as_event_on_timeout => true
timeout => 10
timeout_task_id_field => "id"
timeout_code => '
event.set("[@metadata][timeToFlush]", true)
'
}
if [@metadata][timeToFlush] {
aggregate {
task_id => "%{id}"
code => '
event.set("task", map["task"])
map["task"] = ""
'
map_action => "update"
end_of_task => true
}
} else {
drop {}
}
}
```
The stdin generator is just there to prevent logstash shutting down the pipeline when the generator input finishes. If you remove that when using a generator input you would not get the timeout. For almost any other input it is not needed or useful.
---
### Author:  [@bosand](https://discuss.elastic.co/u/bosand)
#### Post date: [May 13, 2019, 9:23pm UTC](https://discuss.elastic.co/t/aggregate-filter-push-on-event-size-limit/180875/3 "2019-05-13T21:23:09Z")
This is a nice trick.
Thanks Badger!
---
### Author:  [@system](https://discuss.elastic.co/u/system)
#### Post date: [June 10, 2019, 9:23pm UTC](https://discuss.elastic.co/t/aggregate-filter-push-on-event-size-limit/180875/4 "2019-06-10T21:23:12Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.