This is a logstash configuration file

station.conf

input{
    stdin{}
}

filter{
    csv{
    separator=>","
    columns=>["lon","lat","no_discount","discount","total_transactions"]
  }

    mutate{
    remove_field=>["message","host","@timestamp","@version"]
  }

    mutate{
    convert=>{"lon"=>"float"}
    convert=>{"lat"=>"float"}
    convert=>{"no_discount"=>"float"}
    convert=>{"discount"=>"float"}
    convert=>{"total_transactions"=>"float"}
  }

    mutate{
    rename=>{
    "lon"=>"[location][lon]"
    "lat"=>"[location][lat]"
   }
  }
}

output{
    stdout
        {
        codec=>rubydebug
    }
    elasticsearch
        {
        index=>"logstash-discount2"
    }
}