logstash-aemet.conf

# This is the Logstash configuration file to get the AEMET data into Elasticserach

# An input, standard raw tcp, please specify a port of your liking

input {
  tcp {
    port => 11515
    type => "aemet-new"
  }
}

# Filter block to parse the incoming CSV
# Also takes care of setting some values to numbers and drops documents where the temperature is null, as these possibly have not been updated yet
filter {
    if [type] == "aemet-new" {
      csv {
        columns =>["computed_id", "province", "stationid", "location", "geolocation", "dat                                                                                                    etime", "temperature","wind_speed", "wind_direction", "wind_gust", "wind_gust_direction",                                                                                                     "rain_mm", "pressure", "pressure_tendency", "humidity"]
        separator => ","
        skip_empty_columns => "true"
      }
      mutate {
        convert => {
            "temperature" => "float"
            "wind_speed" => "float"
            "wind_gust" => "float"
            "humidity" => "float"
            "pressure" => "float"
            "rain_mm" => "float"
            "pressure_tendency" => "float"
        }
      }
      date {
        match => [ "datetime", "dd/MM/yyyy HH:mm" ]
        timezone => "Europe/Madrid"
        target => "@timestamp"
      }
      if ![temperature] {
       drop { }
      }
   }
}

# This output block will send all events of type "aemet-new" to Elasticsearch at the configured                                                                                                    red
# host and port into monthly indices of the pattern, "aemer-new-YYYY.MM"

output {
  if [type] == "aemet-new" {
    elasticsearch {
      hosts => [ "<your_elasticsearch_host:9200", "your_other_elasticserach_host:9200" ]
      index => "aemet-new-%{+YYYY.MM}"
      document_id => "%{computed_id}"
    }
  }
}