如何将日期字段转换为 logstash 配置文件中的时间戳？

Question

我编写了 Logstash 配置文件，它读取一个 csv 文件并在 elasticsearch 中对其进行索引。但是我在这样做时遇到了一些问题。

1) 在我的输入文件中，我无法将日期列转换为时间戳。 csv文件中日期列的值为：

      - 2016 年 1 月 22 日 20:38 [m/d/yyyy H:mm]

   所以任何人都可以告诉我如何在 kibana 中将日期列转换为时间戳？

2）此外，谁能告诉我如何将纬度和经度列转换为geoip。问题是我只有纬度和经度，我没有 geoip 过滤器中必填字段的来源。

    例如：投诉纬度：12.91518877 投诉经度：77.48066853

所以我不确定要在源字段中输入什么

geoip {
    source => "??"
    }

3）在处理我的文件中的一些行时，logstash 给了我以下错误： 解析 csv 文件时出错。并非所有行都向我显示此错误，但非常一行确实如此，因此我丢失了大量数据，因为我无法对其进行索引。

这是我的 logstash 配置文件：

input{

    file{
         path =>["D:\Project\Logstash Config\Icmc\complaints.csv"]
         start_position => "beginning"
         #sincedb_path => "/dev/null"
         sincedb_path => "/tmp/since.db"
        }

   }

  filter{

      csv{
            separator => ","
            columns =>["category_name", "complaint_sub_category_iid", 
                       "parent_cat_name", "category_parent_iid", 
                       "civic_agency_name", "complaint_title", 
                       "complaint_user_iid", "user_iid", "user_full_name", 
                       "complaint_mobile_number", "complaint_ward_iid", 
                       "ward_name", "complaint_location",  
                       "complaint_address_1", "complaint_latitude", 
                       "complaint_longitude", 
                       "complaint_created","latest_comp_satus_id", 
                       "latest_comp_status_name", "complaint_description"]

            remove_field => ["message"]
        }

     mutate{
         convert => { "complaint_latitude" => "float"}
         }
    mutate{
         convert => { "complaint_longitude" => "float"}
          }
     }

filter{
     geoip {
         source => "clientip"
          }
     }

filter{

        date{
            match => ["complaint_created", "M/d/yyyy H:mm"]
            target => "@timestamp"
           }

        mutate{
            add_field => ["[geoip][lnglat]" , "%{[complaint_longitude]}", 
                     "tmplat", "%{[complaint_latitude]}"]
             }

        mutate{
            merge => ["[geoip][lnglat]", "tmplat"]
             }

        mutate{
            convert => ["[geoip][lnglat]", "float"]
            remove_field => ["tmplat"]
            remove_field => ["complaint_created"]
             }
   }


  output{
        elasticsearch{
        hosts =>["localhost:9200"]
        index => "icmc"
        #document_type => "complaints_filed"
        user => "elastic"
        password => "elastic"
        }

        stdout {  }
      }

Answer 1

对于 #1，由于 Kibana 使用 @timestamp 字段，您可以在 csv 过滤器中使用 @timestamp 代替日期列 (complaint_created?) 或使用 date 过滤器来定位 @timestamp。