Pipeline - Create a delta field

[Ely_96 (Elisa)]

Hi Guys,

I'm trying to study how to create a delta field: I mean a field that should contains a difference between 2 fields inside an existing index.
So, I have 2 examples of CSVs:

ID;Month;Date;Date_string;Name;Surname;Score
1;202112;2021-12-01;DEC-2021;Mario;Rossi;15
2;202112;2021-12-01;DEC-2021;Francesca;Bianchi;12

The second CSV updates my data:

ID;Month;Date;Date_string;Name;Surname;Score
3;202202;2022-02-01;FEB-2022;Mario;Rossi;18
4;202202;2022-02-01;FEB-2022;Francesca;Bianchi;10

As you can see, the score of Mario Bianchi changes (from 15 to 18) and also the score of Francesca Bianchi (from 12 to 10).

The pipeline that ingest the data:

input {
  file {
    path => "C:/elastic_d/logstash/bin/data/lookup/data_2.txt"
    start_position => "beginning"
    sincedb_path => "NULL"
    type => "csv"
  }
}

filter {
  csv {
    separator => ";"
    skip_header => "true"
    columns => ["ID","Month","Date","Date_string","Name","Surname","Score"]
  }

  mutate {convert => {"Score" => "integer"}}

}

output {
  elasticsearch {
    hosts     => ["localhost:9200"]
    index     => "lookup"
  }
}

Then I created a new pipeline in order to create a new index that should contains a new field "Stage_1" with the difference of score (the variation of them):

input {
  elasticsearch {
      hosts           => ["localhost:9200"]
      index           => "lookup"
  }
}

filter {
  elasticsearch {
    hosts                     => ["localhost:9200"]
    index                     => "lookup"
    query                     => "Name:%{Name} AND Surname:%{Surname} AND ID:>%{ID}"
    add_tag                   => "event_benchmark"
  }

  if "event_benchmark" not in [tags] {
    mutate {add_field => { "stage_1" => 0 }}
    mutate {convert => {"stage_1" => "integer"}}
  }
}

output {
    elasticsearch {
    hosts     => ["localhost:9200"]
    index     => "lookup_clean"
  }
}

But I really don't know how to:

1. Get the last event (date desc order) with "event_benchmark" tag
2. Calculate the delta, as difference between Score (without tag "event_benchmark" so the last one) and the last event (in date desc order) with the tag "event_becnhmark". For each key Name + Surname.
3. Insert the calculated value inside a field called "Stage_1"

Could you please help me? Any suggestions... pls

Thanks in advance
Ely