### Author:  [@dyl](https://discuss.elastic.co/u/dyl)
#### Post date: [May 31, 2018, 1:07pm UTC](https://discuss.elastic.co/t/remove-duplicates-in-logstash-logs/134057/1 "2018-05-31T13:07:02Z")
Hello, I use filebeat to send logs to logstash and then I use Logstash to parse them. For exemple this line :
31/5/2018 **01:06:24.073** (TACHE) **30/5/2018** /D T **UE\_TECXXX\_J\_TN\_SSH002\_DELCLONE\_LPSACAS1** (14033)/ **UE\_TECXXX\_X\_LL\_OPE003\_TECHNIQUE** TER STATUS : TN Terminaison normale de la tâche (TN EXIT CODE **0** )
Become after groke parse :
```
{
"DATE_TACHE" => "30/5/2018"
"HEURE_TACHE" => "01:06:24.073"
"NOM_TACHE" => "UE_TECXXX_J_TN_SSH002_DELCLONE_LPSACAS1"
"LOCALISATION_TACHE" => "UE_TECXXX_X_LL_OPE003_TECHNIQUE"
"CODE ERREUR TACHE" => "0"
}
```
And sometime, the same task (tache in french) is done several times, but not at the same moment. That's why I would like to know if I had possibility to **keep the first event of the couple NOM\_TACHE/LOCALISATION\_TACHE and drop every other same identic couple at this one**
For exemple, after the first exemple above, this task should be dropped :
```
{
"DATE_TACHE" => "30/5/2018"
"HEURE_TACHE" => "07:16:24.143"
"NOM_TACHE" => "UE_TECXXX_J_TN_SSH002_DELCLONE_LPSACAS1"
"LOCALISATION_TACHE" => "UE_TECXXX_X_LL_OPE003_TECHNIQUE"
"CODE ERREUR TACHE" => "252"
}
```
I'm completly stuck on this problem...
Can somebody help me ?
Thx for all.
Here my pipeline :
```
### INPUT SECTION ###
input
{
beats
{
port => 5044
}
}
### FILTER SECTION ###
filter
{
grok
{
match => { "message" => ["%{DATE_EU:DATE_LOG} %{TIME:HEURE_TACHE} \(%{WORD:TYPE_TACHE}\) %{DATE_EU:DATE_TACHE}/D . %{WORD:NOM_TACHE}\(%{NUMBER:ID_TACHE}\)/%{WORD:LOCALISATION} (?[A-Z]\w++\s+[A-Z]\w+) : %{WORD:CODE_TACHE} %{GREEDYDATA:DESCRIPTION}" ] }
}
if ([message] =~ "CODE")
{
grok
{
match => { "DESCRIPTION" => ["%{NUMBER:CODE_ERREUR_TACHE}"] }
}
}
mutate
{
remove_field => ["@version","CODE_TACHE","DATE_LOG","ID_TACHE","STATUS","TYPE_TACHE","_id","_index","_score","_type","beat.hostname","beat.name","beat.version","filetype","host","offset","prospector.type","tags"]
convert => { "CODE_ERREUR_TACHE" => "integer" }
}
if [message] !~ "réalisées" and [message] !~ "CODE"
{
drop { }
}
if [message] =~ "groupe"
{
drop { }
}
date
{
match => ["DATE_TACHE", "dd/MM/yyyy"]
}
}
### OUTPUT SECTION ###
output
{
elasticsearch
{
hosts => "http://localhost:9200"
index => "vegaaprodd"
}
stdout { codec => rubydebug }
}
```
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [May 31, 2018, 1:59pm UTC](https://discuss.elastic.co/t/remove-duplicates-in-logstash-logs/134057/2 "2018-05-31T13:59:33Z")
If you are OK with saving the last instead of the first then you can use the fingerprint filter to generate an id based on your choice of fields, then set the document\_id to that id in the elasticsearch output. This will cause the documents to be overwritten when a new event with the same fields arrives.
---
### Author:  [@dyl](https://discuss.elastic.co/u/dyl)
#### Post date: [May 31, 2018, 2:36pm UTC](https://discuss.elastic.co/t/remove-duplicates-in-logstash-logs/134057/3 "2018-05-31T14:36:48Z")
Hey @Badger,
Thx for your reply, i'll try it fastly and come back to you if i've got questions ! 🙂
---
### Author:  [@dyl](https://discuss.elastic.co/u/dyl)
#### Post date: [June 1, 2018, 7:56am UTC](https://discuss.elastic.co/t/remove-duplicates-in-logstash-logs/134057/4 "2018-06-01T07:56:03Z")
@Badger
I tried to do what you said, but it doesn't work.
I modified my pipeline and add "fingerprint" section into "filter" section which looks like : (I think config is correct, I want to hash with DATE,NOM & LOCALISATION each event)
```
fingerprint
{
source => "%{[DATE_TACHE][NOM_TACHE][LOCALISATION_TACHE]}"
target => "fingerprint_id"
concatenate_sources => "true"
method => "MURMUR3"
}
```
And add a "document\_id" field into output section
```
output
{
elasticsearch
{
hosts => "http://localhost:9200"
index => "testdoublon"
document_id => "%{[fingerprint_id]}"
}
```
I tried with 8 logs lines to test (On these 8, 5 are duplicates and 3 are unique. So I should have 4 events into elasticsearch... But I only have 1 "docs.count" and 7 "docs.deleted" :
```
health status index uuid pri rep docs.count docs.deleted store.size pri.store.size
yellow open testdoublon 87WQvbi1R1ml6ih1nM_QMA 5 1 1 7 18.2kb 18.2kb
```
I don't know why... Theses 8 lines are (better with copy/paste and notepad++ 🙂 )
**31/05/2018** 07:26:52.720 (TACHE) 31/05/2018/D T **UE\_VGAXXX\_J\_TN\_VGA002\_SAV\_LOGBDD\_VEGA(26)/PR\_LWXXX\_TECVGAXXX** TER STATUS : TN Terminaison normale de la tâche (TN EXIT CODE 0)
**31/05/2018** 07:26:52.766 (TACHE) 31/05/2018/D T **OR\_WKFXXX\_J\_TN\_WKF998MT\_CALC\_STAT\_JOUR(15805)/OR\_XXXXXX\_X\_LL\_POC001\_PROTOPROD** TER STATUS : TN Terminaison normale de la tâche (TN EXIT CODE 0)
**31/05/2018** 07:28:19.013 (TACHE) 31/05/2018/D T **XX\_TECXXX\_J\_TN\_TST001\_TEST\_AGENT(13423)/MA\_NETXXX\_X\_LL\_OPE001\_EFLUIDNET** TER STATUS : TN Terminaison normale de la tâche (TN EXIT CODE 0)
**31/05/2018** 07:30:35.950 (TACHE) 31/05/2018/D T **UE\_TECXXX\_J\_TN\_MOV001\_PREVENERCOM\_2\_ARCH(11968)/PR\_LWXXX\_TECVGAXXX** TER STATUS : TN Terminaison normale de la tâche (TN EXIT CODE 0)
**31/05/2018** 07:41:18.663 (TACHE) 31/05/2018/D T **UE\_TECXXX\_J\_TN\_MOV001\_PREVENERCOM\_2\_ARCH(11968)/PR\_LWXXX\_TECVGAXXX** TER STATUS : TN Terminaison normale de la tâche (TN EXIT CODE 0)
**31/05/2018** 07:51:18.663 (TACHE) 31/05/2018/D T **UE\_TECXXX\_J\_TN\_MOV001\_PREVENERCOM\_2\_ARCH(11968)/PR\_LWXXX\_TECVGAXXX** TER STATUS : TN Terminaison normale de la tâche (TN EXIT CODE 0)
**31/05/2018** 08:41:18.663 (TACHE) 31/05/2018/D T **UE\_TECXXX\_J\_TN\_MOV001\_PREVENERCOM\_2\_ARCH(11968)/PR\_LWXXX\_TECVGAXXX** TER STATUS : TN Terminaison normale de la tâche (TN EXIT CODE 0)
**31/05/2018** 09:12:18.663 (TACHE) 31/05/2018/D T **UE\_TECXXX\_J\_TN\_MOV001\_PREVENERCOM\_2\_ARCH(11968)/PR\_LWXXX\_TECVGAXXX** TER STATUS : TN Terminaison normale de la tâche (TN EXIT CODE 0)
My new pipeline :
```
### INPUT SECTION ###
input
{
beats
{
port => 5044
}
}
### FILTER SECTION ###
filter
{
grok
{
match => { "message" => ["%{DATE_EU:DATE_LOG} %{TIME:HEURE_TACHE} \(%{WORD:TYPE_TACHE}\) %{DATE_EU:DATE_TACHE}/D . %{WORD:NOM_TACHE}\(%{NUMBER:ID_TACHE}\)/%{WORD:LOCALISATION} (?[A-Z]\w++\s+[A-Z]\w+) : %{WORD:CODE_TACHE} %{GREEDYDATA:DESCRIPTION}" ] }
}
if ([message] =~ "CODE")
{
grok
{
match => { "DESCRIPTION" => ["%{NUMBER:CODE_ERREUR_TACHE}"] }
}
}
mutate
{
remove_field => ["@version","CODE_TACHE","DATE_LOG","ID_TACHE","STATUS","TYPE_TACHE","_id","_index","_score","_type","beat.hostname","beat.name","beat.version","filetype","host","offset","prospector.type","tags"]
convert => { "CODE_ERREUR_TACHE" => "integer" }
}
if [message] !~ "réalisées" and [message] !~ "CODE"
{
drop { }
}
if [message] =~ "groupe"
{
drop { }
}
date
{
match => ["DATE_TACHE", "dd/MM/yyyy"]
}
fingerprint
{
source => "%{[DATE_TACHE][NOM_TACHE][LOCALISATION_TACHE]}"
target => "[fingerprint_id]"
concatenate_sources => "true"
method => "MURMUR3"
}
}
### OUTPUT SECTION ###
output
{
elasticsearch
{
hosts => "http://localhost:9200"
index => "testdoublon"
document_id => "%{[fingerprint_id]}"
}
stdout { codec => rubydebug }
}
```
Thx for all
---
### Author:  [@system](https://discuss.elastic.co/u/system)
#### Post date: [June 29, 2018, 7:56am UTC](https://discuss.elastic.co/t/remove-duplicates-in-logstash-logs/134057/5 "2018-06-29T07:56:10Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.