crimes.elastic.logstash.final_old.conf

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

input {
 file {
   path => ["/home/gabriel/cai/logstash/logstash-7.12.0-linux-x86_64/logstash-7.12.0/bin/nypd_mil.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"]
  }
#    date {
#     match => [ "CMPLNT_FR_DT", "MM/dd/yyyy" ]
#    }
#  mutate {
#                add_field => {"[location][type]" => "Point"}
#                add_field => {"[location][coordinates]" => ["%{[Longitude]}","%{[Latitude]}"] }
#  }  
#  mutate {
#                convert => { "[location][coordinates][0]" => "float" }
#                convert => { "[location][coordinates][1]" => "float" }
#  }  
 if [Latitude] {
    mutate {
      add_field => {
        "[location][lon]" => "%{[Longitude]}"
        "[location][lat]" => "%{[Latitude]}"
      }
    }
    mutate {
      convert => {
        "[location][lon]" => "float"
        "[location][lat]" => "float"
      }
    }
 }
}


output {
elasticsearch{
    hosts=> ["localhost:9200"]
    index=> ["crimesdb"]
#    template => "C:\logstash-7.12.0\bin\crimes_map_template.json"
  }
}