# 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:** 1
**Showing post:** 2

<div class="post-metadata">

### Author: ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@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")

</div>

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.

---

_[View the full topic](https://discuss.elastic.co/t/aggregate-filter-push-on-event-size-limit/180875)._