logstash.conf

input {
  file {
    path => [ "/tmp/data/*.EXT", "/tmp/data/*.txt", "/tmp/data/**/*.EXT", "/tmp/data/**/*.txt"  ]
    start_position => "beginning"
    sincedb_path => "/dev/null"
  }
}

filter {
  csv {
    separator => ","
    columns => ["TRANSACTION_TYPE", "FILE_TYPE", "FILE_DATE", "FILE_NUM", "MPO_REFERENCE", "SERIAL_NUMBER",
    "SUB_BUILDING", "BUILDING_NAME", "DELIVER_POINT_ALIAS", "BUILDING_NUMBER", "DEPENDENT_STREET",
    "PRINCIPAL_STREET", "DBL_DPNDT_LOCLTY", "DEPENDENT_LOCALITY", "POST_TOWN", "COUNTY", "OUTCODE",
    "INCODE", "FILE_NUM", "LARGE_SITE_INDICATOR", "IGT"]
  }
  if [message] =~ /^"M12"/ or [message] =~ /^"M14"/ {
  drop {}
  }

  fingerprint {
    source => "message"
    target => "[@metadata][fingerprint]"
    method => "MURMUR3"
  }

}

output {
  elasticsearch {
    hosts => "${ES_HOST}"
    ssl => true
    index => "elecandgas"
    document_type => "mpo"
    document_id => "%{[@metadata][fingerprint]}"
  }
  stdout {}
}