# Aggregate filter plugin - need help to make it work

**URL:** <https://discuss.elastic.co/t/aggregate-filter-plugin-need-help-to-make-it-work/140901>\
**Category:** Logstash\
**Created:** [July 20, 2018, 1:32pm UTC](https://discuss.elastic.co/t/aggregate-filter-plugin-need-help-to-make-it-work/140901 "2018-07-20T13:32:03Z")\
**Posts on this page:** 4\
**Page:** 1

<div class="post-metadata">

**Author:** ![Jevgenij](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/jevgenij/32/33543_2.png) [@Jevgenij](https://discuss.elastic.co/u/Jevgenij)\
**Post date:** [July 20, 2018, 1:32pm UTC](https://discuss.elastic.co/t/aggregate-filter-plugin-need-help-to-make-it-work/140901/1 "2018-07-20T13:32:03Z")

</div>

Hello!

I have some data in joined db tables.

To collect it I use the following query:

```
SELECT `item`.`id`, `item`.`name`, `colour`.`name` AS `colour` FROM `item`
LEFT JOIN `item_has_colour` AS `ic` ON `ic`.`item` = `item`.`id`
LEFT JOIN `colour` ON `colour`.`id` = `ic`.`colour`
ORDER BY `item`.`id`

```

So my final dataset looks like this:

```
mysql> SELECT `item`.`id`, `item`.`name`, `colour`.`name` AS `colour` FROM `item` LEFT JOIN `item_has_colour` AS `ic` ON `ic`.`item` = `item`.`id` LEFT JOIN `colour` ON `colour`.`id` = `ic`.`colour` ORDER BY `item`.`id`;
+----+-------+--------+
| id | name | colour |
+----+-------+--------+
| 1 | cat | red |
| 1 | cat | green |
| 1 | cat | yellow |
| 2 | dog | red |
| 2 | dog | yellow |
| 3 | mouse | red |
| 3 | mouse | blue |
+----+-------+--------+
7 rows in set (0,00 sec)

```

In order to aggregate information (`colour` here) belonging to a same record (cat, dog or mouse) I'm trying to use Aggregate filter plugin, according to its [docs](https://www.elastic.co/guide/en/logstash/current/plugins-filters-aggregate.html#plugins-filters-aggregate-example4).

So here is my `logstash.conf` with `filter` section:

```
input {
  jdbc { 
    jdbc_driver_library => "/home/user/mysql-connector-java-8.0.11.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://localhost:3306/testdb"
    jdbc_user => "root"
    jdbc_password => "root"
    statement => "SELECT `item`.`id`, `item`.`name`, `colour`.`name` AS `colour` FROM `item` LEFT JOIN `item_has_colour` AS `ic` ON `ic`.`item` = `item`.`id` LEFT JOIN `colour` ON `colour`.`id` = `ic`.`colour` ORDER BY `item`.`id`"
  }
}
filter {
  aggregate {
    task_id => "%{id}"
    code => "
      map['id'] = event.get('id')
      map['name'] = event.get('name')
      map['colours'] ||= []
      map['colours'] << {'colour' => event.get('colour')}
      event.cancel()
    "
    push_previous_map_as_event => true
    timeout => 3
  }
}
output {
  stdout { codec => json_lines }
  elasticsearch {
    "hosts" => "localhost:9200"
    "index" => "test-migrate"
    "document_type" => "data"
    "document_id" => "%{id}"
  }
}

```

The filter content is almost without changes copied from example in documentation. And as far as I can understand it, it should work correctly.

But when I run Logstash, something weird happens.

Here is its output:

```
[INFO] 2018-07-20 14:50:56.565 [[main]<jdbc] jdbc - (0.026601s) SELECT `item`.`id`, `item`.`name`, `colour`.`name` AS `colour` FROM `item` LEFT JOIN `item_has_colour` AS `ic` ON `ic`.`item` = `item`.`id` LEFT JOIN `colour` ON `colour`.`id` = `ic`.`colour` ORDER BY `item`.`id`
{"name":"cat","@timestamp":"2018-07-20T12:50:56.848Z","colours":[{"colour":"yellow"}],"@version":"1","id":1}
{"name":"mouse","@timestamp":"2018-07-20T12:50:56.848Z","colours":[{"colour":"red"}],"@version":"1","id":3}
{"name":"dog","@timestamp":"2018-07-20T12:50:56.855Z","colours":[{"colour":"yellow"}],"@version":"1","id":2}
{"name":"dog","@timestamp":"2018-07-20T12:50:56.841Z","colours":[{"colour":"red"}],"@version":"1","id":2}
{"name":"cat","@timestamp":"2018-07-20T12:50:56.847Z","colours":[{"colour":"red"},{"colour":"green"}],"@version":"1","id":1}
{"tags":["_aggregatefinalflush"],"colours":[{"colour":"blue"}],"@version":"1","id":3,"@timestamp":"2018-07-20T12:50:57.372Z","name":"mouse"}

```

Here you can see what there were more than 3 events, and only `cat` got 2 of its 3 `colour`s.

Every next run produces different results.

For example:

```
[INFO] 2018-07-20 15:27:00.243 [[main]<jdbc] jdbc - (0.017218s) SELECT `item`.`id`, `item`.`name`, `colour`.`name` AS `colour` FROM `item` LEFT JOIN `item_has_colour` AS `ic` ON `ic`.`item` = `item`.`id` LEFT JOIN `colour` ON `colour`.`id` = `ic`.`colour` ORDER BY `item`.`id`
{"id":2,"colours":[{"colour":"red"}],"@version":"1","@timestamp":"2018-07-20T13:27:00.461Z","name":"dog"}
{"id":1,"colours":[{"colour":"yellow"},{"colour":"red"},{"colour":"green"}],"@version":"1","@timestamp":"2018-07-20T13:27:00.507Z","name":"cat"}
{"id":2,"colours":[{"colour":"yellow"}],"@version":"1","@timestamp":"2018-07-20T13:27:00.510Z","name":"dog"}
{"tags":["_aggregatefinalflush"],"id":3,"@version":"1","@timestamp":"2018-07-20T13:27:00.602Z","name":"mouse","colours":[{"colour":"red"},{"colour":"blue"}]}

```

So here `cat` got all of her 3 `colour`s, and `mouse` got its 2.

But the `dog` was processed in two separate events, loosing its `colour`s.

It looks like input strings, fetched by db query, are processed in random order.

Any clue on how to fix it? Am I missing something?

---

<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:** [July 20, 2018, 2:11pm UTC](https://discuss.elastic.co/t/aggregate-filter-plugin-need-help-to-make-it-work/140901/2 "2018-07-20T14:11:17Z")

</div>

If you are using an aggregate filter you have to set --pipeline.workers 1. I think different threads are picking up different rows. So one thread is aggregating red and green for cat, but yellow for cat gets processed in a different thread.

Note the second paragraph of the Description section of the docs!

---

<div class="post-metadata">

**Author:** ![Jevgenij](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/jevgenij/32/33543_2.png) [@Jevgenij](https://discuss.elastic.co/u/Jevgenij)\
**Post date:** [July 20, 2018, 2:22pm UTC](https://discuss.elastic.co/t/aggregate-filter-plugin-need-help-to-make-it-work/140901/3 "2018-07-20T14:22:16Z")

</div>

How could I miss that? 🤦‍♂️

Thanks you a lot for pointing that out!

It works flawlessly now.

---

<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:** [August 17, 2018, 2:22pm UTC](https://discuss.elastic.co/t/aggregate-filter-plugin-need-help-to-make-it-work/140901/4 "2018-08-17T14:22:21Z")

</div>

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