# Aggregate documents

**URL:** <https://discuss.elastic.co/t/aggregate-documents/360599>\
**Category:** Logstash\
**Created:** [May 31, 2024, 7:08am UTC](https://discuss.elastic.co/t/aggregate-documents/360599 "2024-05-31T07:08:48Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![reed](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/reed/32/122863_2.png) [@reed](https://discuss.elastic.co/u/reed)\
**Post date:** [May 31, 2024, 7:08am UTC](https://discuss.elastic.co/t/aggregate-documents/360599/1 "2024-05-31T07:08:48Z")

</div>

Hi all,

I've created a pipeline that aggregate all input documents and generate a count for each group.

in this moment I have two problems :

1. at the first execution the aggregation works fine and the count is correct
2. after the first execution each count double, I think why the map remain in memory
3. all input documents are sent to the output

what I want is:

1. find a way to release the map after each execution
2. send to the output only the aggregate documents

please help me

thank you in advance  
Diego

here my pipeline:

```auto
input {
  elasticsearch{
    read all documents
  }
}

filter {

  ruby {
    code => "
      require 'time'
      if event.get('reg512Created')
        date = Time.parse(event.get('reg512Created'))
        event.set('reg512Created_day', date.strftime('%d-%m-%Y'))
      end
    "
  }

  aggregate {
    task_id => "%{reg512UPNRequestor}-%{reg512Created_day}"
    code => "
      map['regCount'] ||= 0; map['regCount'] += 1
      event.set('regCount', map['regCount'])
    "
    push_previous_map_as_event => true
    timeout => 45
  }

  aggregate {
    task_id => "%{reg512UPNRequestor}-%{reg512Created_day}"
    end_of_task => true
  }

}

output {
  elasticsearch{
    receive all documents
  }
}

```

---

<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 31, 2024, 10:48am UTC](https://discuss.elastic.co/t/aggregate-documents/360599/2 "2024-05-31T10:48:56Z")

</div>

I wouldn't expect this to work.

Getting rid of the source documents and just keeping the aggregations is simple. Just add `event.cancel` to the code option of the first aggregate.

The second aggregate will not get triggered by the aggregations since those events do not have the [reg512UPNRequestor] and [reg512Created\_day] fields. The filter will be [a no-op](https://github.com/logstash-plugins/logstash-filter-aggregate/blob/e216a60fb96cfe679c28210747838ea36ef067d8/lib/logstash/filters/aggregate.rb#L175). You need to add them to the map if you want them to be in those aggregated events. You may want to add the task\_id as well.

I would expect the second aggregate to be triggered by every source document. It is clearly not, but I do not understand why.

---

<div class="post-metadata">

**Author:** ![reed](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/reed/32/122863_2.png) [@reed](https://discuss.elastic.co/u/reed)\
**Post date:** [May 31, 2024, 11:08am UTC](https://discuss.elastic.co/t/aggregate-documents/360599/3 "2024-05-31T11:08:18Z")

</div>

Hi Badger, thanks for your reply, I've solved so

```auto
  aggregate {
    task_id => "%{reg512UPNRequestor}-%{reg512Created_day}"
    code => "
      map['field-1'] = event.get('reg512UPNRequestor')
      map['field-2'] = event.get('reg512Created_day')
      map['myCount'] ||= 0; map['myCount'] += 1
      event.cancel()
    "
    push_previous_map_as_event => true
    timeout => 3
  }

```

this works very fine