# Logstash filter aggregate with 2 nested fields **URL:** \ **Category:** Logstash\ **Created:** [March 17, 2024, 4:05am UTC](https://discuss.elastic.co/t/logstash-filter-aggregate-with-2-nested-fields/355548 "2024-03-17T04:05:12Z")\ **Posts on this page:** 6\ **Page:** 1
**Author:** ![Lucas\_Giusti](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/lucas_giusti/32/132704_2.png) [@Lucas\_Giusti](https://discuss.elastic.co/u/Lucas_Giusti)\ **Post date:** [March 17, 2024, 4:05am UTC](https://discuss.elastic.co/t/logstash-filter-aggregate-with-2-nested-fields/355548/1 "2024-03-17T04:05:12Z")
Goodnight. Could you help me with the problem below? I have the mysql table below: ```auto productId productName categoryId categoryName locationId locationPrice ------------------------------------------------------------------------------------------ 1 Product1 1 Category1 1 100 1 Product1 1 Category1 2 200 2 Product2 1 Category1 1 100 2 Product2 1 Category1 2 200 2 Product2 2 Category2 1 100 2 Product2 2 Category2 2 200 2 Product2 3 Category3 3 300 ``` and I have the following logstash code: ```auto input { jdbc { jdbc_driver_library => "/usr/share/logstash/logstash-core/lib/jars/mysql-connector-java-8.0.16.jar" jdbc_driver_class => "com.mysql.jdbc.Driver" jdbc_connection_string => "" jdbc_user => "" jdbc_password => "" jdbc_paging_enabled => true jdbc_paging_mode => "explicit" jdbc_page_size => 1000 tracking_column => "updatedat" use_column_value => true tracking_column_type => "numeric" schedule => "*/1 * * * *" statement => "SELECT a.productId, a.productName, b.categoryId, b.categoryName, c.locationId, c.locationPrice FROM Product as a INNER JOIN Category as b on a.categoryId = b.id INNER JOIN Location as c on a.locationId = c.id" } } filter { aggregate { task_id => "%{productId}" code => " map['productId'] ||= event.get('productId') map['productName'] ||= event.get('productName') map['categories'] ||= [] unless map['categories'].any? { |cat| cat['categoryId'] == event.get('categoryId') } map['categories'] << { 'categoryId' => event.get('categoryId'), 'categoryName' => event.get('categoryName') } end map['locations'] ||= [] #unless map['locations'].any? { |loc| loc['locationId'] == event.get('locationId') } map['locations'] << { 'locationId' => event.get('locationId'), 'locationPrice' => event.get('locationPrice') } #end event.cancel() " push_previous_map_as_event => true timeout => 3 } } output { stdout { codec => "rubydebug"} } ``` and I would like to obtain the following result: ```auto [ { "productId": 1, "productName": "Product1", "categories": [ { "categoryId": 1, "categoryName": "Category1" } ], "locations": [ { "locationId": 1, "locationPrice": 100 }, { "locationId": 2, "locationPrice": 200 } ] }, { "productId": 2, "productName": "Product2", "categories": [ { "categoryId": 1, "categoryName": "Category1" }, { "categoryId": 2, "categoryName": "Category2" }, { "categoryId": 3, "categoryName": "Category3" } ], "locations": [ { "locationId": 1, "locationPrice": 100 }, { "locationId": 2, "locationPrice": 200 }, { "locationId": 3, "locationPrice": 300 } ] } ] ``` but I'm getting the following result: ```auto [ { "productId": 1, "productName": "Product1", "categories": [ { "categoryId": 1, "categoryName": "Category1" } ], "locations": [ { "locationId": 1, "locationPrice": 100 } ] }, { "productId": 2, "productName": "Product2", "categories": [ { "categoryId": 1, "categoryName": "Category1" }, { "categoryId": 2, "categoryName": "Category2" }, { "categoryId": 3, "categoryName": "Category3" } ], "locations": [ { "locationId": 1, "locationPrice": 100 }, { "locationId": 2, "locationPrice": 200 }, { "locationId": 3, "locationPrice": 300 } ] } ] ``` Note that for product 1, the number of locations is limited to the number of categories. Does anyone know what the problem is and how to solve it?`Preformatted text` ---
**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:** [March 17, 2024, 10:24pm UTC](https://discuss.elastic.co/t/logstash-filter-aggregate-with-2-nested-fields/355548/2 "2024-03-17T22:24:31Z")
If you use that jdbc input, remove the aggregate filter, and add an output with a json codec then what do the events look like? I'm not going to build a database instance to reproduce your issue, but can read json from a file to do so. ---
**Author:** ![Lucas\_Giusti](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/lucas_giusti/32/132704_2.png) [@Lucas\_Giusti](https://discuss.elastic.co/u/Lucas_Giusti)\ **Post date:** [March 18, 2024, 4:26pm UTC](https://discuss.elastic.co/t/logstash-filter-aggregate-with-2-nested-fields/355548/3 "2024-03-18T16:26:25Z")
unfortunately I couldn't reproduce it with this example I showed. In fact, my data is much larger, however, it is the same structure as this example. I just wanted to make it easier to understand. It seems to me that the processing is working in parallel for the same task\_id, and thus overwriting the previous processing. That's why sometimes the final record has a record in the categories array, after I run it again it has another, for example. I'm new to logstash and don't understand it very well. Could this be happening? Is there any way to guarantee that we will only have one execution for each task\_id until the end of processing? ---
**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:** [March 18, 2024, 4:53pm UTC](https://discuss.elastic.co/t/logstash-filter-aggregate-with-2-nested-fields/355548/4 "2024-03-18T16:53:37Z")
> [@Lucas\_Giusti](#): > > It seems to me that the processing is working in parallel for the same task\_id, and thus overwriting the previous processing. Are you setting pipeline.workers to 1, as the [documentation](https://www.elastic.co/guide/en/logstash/current/plugins-filters-aggregate.html#plugins-filters-aggregate-description) says you must? Otherwise different threads may process different subsets of the data on different runs. ---
**Author:** ![Lucas\_Giusti](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/lucas_giusti/32/132704_2.png) [@Lucas\_Giusti](https://discuss.elastic.co/u/Lucas_Giusti)\ **Post date:** [March 18, 2024, 9:23pm UTC](https://discuss.elastic.co/t/logstash-filter-aggregate-with-2-nested-fields/355548/5 "2024-03-18T21:23:22Z")
The problem was resolved after including the pipeline.workers to 1 configuration. Thank yo very much. ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [April 15, 2024, 9:23pm UTC](https://discuss.elastic.co/t/logstash-filter-aggregate-with-2-nested-fields/355548/6 "2024-04-15T21:23:39Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.