crimes.logstash.conf

# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.

input {
 file {
   path => ["/home/javier/Documents/logstash-7.12.0/bin/ejemplologstah.csv"]
   start_position => "beginning"
   sincedb_path => "/dev/null"
   mode => "read"
   exit_after_read => "true"
  }  
}

filter {
  csv {
                columns => ["CMPLNT_NUM","CMPLNT_FR_DT","CMPLNT_FR_TM","CMPLNT_TO_DT","CMPLNT_TO_TM","ADDR_PCT_CD","RPT_DT","KY_CD","OFNS_DESC","PD_CD","PD_DESC","CRM_ATPT_CPTD_CD","LAW_CAT_CD","BORO_NM","LOC_OF_OCCUR_DESC","PREM_TYP_DESC","JURIS_DESC","JURISDICTION_CODE","PARKS_NM","HADEVELOPT","HOUSING_PSA","X_COORD_CD","Y_COORD_CD","SUSP_AGE_GROUP","SUSP_RACE","SUSP_SEX","TRANSIT_DISTRICT","Latitude","Longitude","Lat_Lon","PATROL_BORO","STATION_NAME","VIC_AGE_GROUP","VIC_RACE","VIC_SEX"]
                separator => ","
                remove_field => ["message","@version","host","path","X_COORD_CD","Y_COORD_CD","Lat_Lon"]
  }

  mutate {
     add_field => {"[location][type]" => "Point"}
     add_field => {"[location][coordinates]" => ["%{[Latitude]}","%{[Longitude]}"] }
  }
  mutate {
     convert => { "[location][coordinates][0]" => "float" }
     convert => { "[location][coordinates][1]" => "float" }
  }
} 


output {
  mongodb {
    id => "mongodb_crimes_id"
    collection => "crimes"
    database => "crimesdb"
    uri => "mongodb://127.0.0.1:27017/crimesdb"
    codec => "json"
  }
}