logstash.conf

input {
  file {
    path => "/202401_PEP.csv"
    start_position => "beginning"
    sincedb_path => "/dev/null" 
    codec => plain {
      charset => "UTF-8"
    }    
  }
}

filter {
  csv {
    separator => ","
    skip_header => true
    columns => ["CPF", "Name", "FunctionAbbreviation", "FunctionDescription", "FunctionLevel", "AgencyName", "StartDateExercise", "EndDateExercise", "EndDateWaitingPeriod"]
  } 

  mutate {
    add_field => {
      "date_concatenated" => "%{StartDateExercise} %{EndDateExercise}"
    }
  }
    
  date {
    match => ["StartDateExercise", "dd/MM/yyyy"]
    target => "StartDateExercise"
  }

  if [EndDateExercise] == "Não informada" {
    mutate {
      remove_field => ["EndDateExercise"]
    }
  } else {
    date {
      match => ["EndDateExercise", "dd/MM/yyyy"]
      target => "EndDateExercise"
    }
  }
    
  fingerprint {
    source => ["CPF", "Name", "FunctionDescription", "AgencyName"]
    method => "SHA256"
    target => "[@metadata][document_id]"
  }

  mutate {
    remove_field => [ "message", "host", "path" ]
  } 
}

output {    
  # stdout { codec => "rubydebug" }

  elasticsearch {
    hosts => ["elasticsearch:9200"]
    index => "br_pep-%{+YYYY.MM.dd}"
    template => "/mapping.json"
    template_name => "doc"
    template_overwrite => true
    document_id => "%{[@metadata][document_id]}"
  }
}