# Logstash Pipeline Filter (\*)

**URL:** <https://discuss.elastic.co/t/logstash-pipeline-filter/297266>\
**Category:** Logstash\
**Created:** [February 15, 2022, 3:42pm UTC](https://discuss.elastic.co/t/logstash-pipeline-filter/297266 "2022-02-15T15:42:36Z")\
**Posts on this page:** 5\
**Page:** 1

<div class="post-metadata">

**Author:** ![max\_cyril](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/max_cyril/32/100067_2.png) [@max\_cyril](https://discuss.elastic.co/u/max_cyril)\
**Post date:** [February 15, 2022, 3:42pm UTC](https://discuss.elastic.co/t/logstash-pipeline-filter/297266/1 "2022-02-15T15:42:36Z")

</div>

RE:  
Hi,I am struggling to write a logstash pipeline filter section

 ![image (2)](https://us1.discourse-cdn.com/elastic/original/3X/f/c/fc12410b77b2707e0f043ba01305a733ae29d129.png)

i have used "TRANSFORMS" from kibana interface, it did not solve the case.

\*\*\* i want to filter and only output event that correspond to the maximum of completion per users \*\*\*

for instance :

```auto

n° | user | completion

05 | u2 | 20
 ... | ... | ...
423 | u48 | 100
424 | u49 | -
425 | u50 | 0

```

here my trial:

```auto
input
{elasticsearch
{hosts => "...."
user => "..."
password => "..."
index => "index1"
codec =>"json"
docinfo => true
}}

filter {
aggregate {
task_id => "%{users}"
code => "map['completion'] = event.get('completion') ;
event.cancel if (map['completion']) != map['completion'].max()"
map_action => "create" }
}

output
{elasticsearch
{hosts => "..."
user => "..."
password => "..."
index => "index2"
document_type =>"%{[@metadata][_type]}"
document_id =>"%{[@metadata][_id]}"
}}

```

Thanks in advance.

---

<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:** [February 15, 2022, 5:12pm UTC](https://discuss.elastic.co/t/logstash-pipeline-filter/297266/2 "2022-02-15T17:12:07Z")

</div>

> [@max\_cyril](#):
>
> i want to filter and only output event that correspond to the maximum of completion per users

That requires logstash to look into the future and predict what events will occur after the current event. Tricky.

I think the best you can do is to use `push_map_as_event_on_timeout`.

```
aggregate {
    task_id => "%{users}"
    code => '
        map["completion"] ||= 0
        c = event.get("completion")
        if c > map["completion"]
            map["completion"] = c
        end
        event.cancel
    '
    push_map_as_event_on_timeout => true
    timeout_task_id_field => "users"
    timeout => 600 # 10 minutes timeout
}

```

Note that the event that is created will only contain the fields that you add to the map, so in this case it will have [completion] and [users] (because `timeout_task_id_field` is set). If you have other fields you want to preserve then add them to the map.

---

<div class="post-metadata">

**Author:** ![max\_cyril](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/max_cyril/32/100067_2.png) [@max\_cyril](https://discuss.elastic.co/u/max_cyril)\
**Post date:** [February 16, 2022, 11:00am UTC](https://discuss.elastic.co/t/logstash-pipeline-filter/297266/3 "2022-02-16T11:00:07Z")

</div>

thank you @Badger ,  
the output is actualy something like this:

```auto
users | completion
   - | number
   - | number
   u49 | -
   - | number

```

users field is no longer outputed ,therefore we can not know for which users the field completion correspond.

---

<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:** [February 16, 2022, 6:40pm UTC](https://discuss.elastic.co/t/logstash-pipeline-filter/297266/4 "2022-02-16T18:40:46Z")

</div>

If the [users] field is only output when it changes then you could use a ruby filter to add it back in. You need order preserved, so make sure that pipeline.workers is 1 and pipeline.ordered evaluates to true.

```
ruby {
    init => '@user = nil'
    code => '
        user = event.get("users")
        if user
            @user = user
        else
            event.set("users", @user)
        end
    '
}
```

---

<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:** [March 16, 2022, 6:41pm UTC](https://discuss.elastic.co/t/logstash-pipeline-filter/297266/5 "2022-03-16T18:41:28Z")

</div>

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