Logstash JSON conversion can we control

Elastic Stack Logstash
4

Hi Krishna!

I follow the filter topic...but some where I stuck....I given full error details in the below link...

Kindly see Logstash Aggregate plugin

And same here also..
This example from elastic documentation:
Given that you have this SQL query : SELECT country_name, town_name FROM town

{ "country_name": "France", "town_name": "Paris" }
{ "country_name": "France", "town_name": "Marseille" }
{ "country_name": "USA", "town_name": "New-York" }

And you would like these 2 result events to push them into elasticsearch :
{ "country_name": "France", "towns": [ {"town_name": "Paris"}, {"town_name": "Marseille"} ] }
{ "country_name": "USA", "towns": [ {"town_name": "New-York"} ] }

They mention we have to follow this filter
You can do that using push_previous_map_as_event aggregate plugin option :
filter {
aggregate {
task_id => "%{country_name}"
code => "
map['country_name'] = event.get('country_name')
map['towns'] ||= []
map['towns'] << {'town_name' => event.get('town_name')}
event.cancel()
"
push_previous_map_as_event => true
timeout => 3
}
}

And with respect to above filter...I have sql query like this..
Select pid, fsid, title, brand from table1;
But I don't want to aggregate it...but I want to put all the above fields inside of one another name like Shekhar...

So ..I used the following method...
filter {
aggregate {
task_id => "%{id}"
code => "

 map['shekhar'] ||= []
 map['shekhar'] << {'brand' => event.get('brand')}
map['shekhar'] << {'pid' => event.get('pid')}
map['shekhar'] << {'title' => event.get('title')}

map['shekhar'] << {'brand_id' => event.get('brand_id')}
event.cancel()
"
push_previous_map_as_event => true
timeout => 3
}
}

I got output as I want all fields inside of the shekhar[].....but only one output showing because of aggregation....it showing last value....so how I can edit the above filter to get all the documents.....

Thank you.