csv - 无法通过Logstash加载CSV文件

Question

我是ELK的新手，我正尝试通过Logstash加载本地存储的.csv文件，以便可以将其与Elasticsearch一起使用。

logstash配置文件如下所示:

   input {
    file {
       path => "C:\ELK-Stack\Cars Data Set\cars.csv"
       start_position => "beginning"
      sincedb_path => "/dev/null"
      }
  }

filter {
         csv {
                separator =>","
                columns => ["maker","model","mileage","manufacture-year","engine_displacement","engine_power","body_type", "color_slug","stk_year","transmission","door_count","seat_count","fuel_type","date_created","date_last_seen", "price_eur"]
             }
     mutate {convert => ["mileage", "integer"]}
     mutate {convert => ["price_eur", "float"]}
     mutate {convert => ["door_count", "integer"]}
     mutate {convert => ["engine_power", "integer"]}
     mutate {convert => ["seat_count", "integer"]}

     }

output {
 elasticsearch {
 hosts => ["localhost:9200"]}
 index => "cars"
document_type => "sold_cars"
  }
stdout {}
}

文件的路径是: C:\ELK-Stack\Cars Data Set\cars.csv
我得到的输出看起来像这样:



.csv文件具有超过一百万行。任何帮助，将不胜感激。

编辑:

现在，我正在处理另一个数据集，无法通过logstash加载它。

    input {
  file {
    path => "C:\ELK-Stack\311.csv"
    start_position => "beginning"
    sincedb_path => "NUL"
  }
}

filter {
         csv {
                separator =>","
                columns => ["Unique Key","Created Date","Closed Date","Agency","Agency Name","Complaint Type","Descriptor", "Location Type","Incident Zip","Incident Address","Street Name","Cross Street 1","Cross Street 2","Intersection Street 1","Intersection Street 2", "Address Type", "City", "Landmark", "Facility Type", "Status", "Due Date", "Resolution Description", "Resolution Action Updated Date", "Community Board", "BBL", "Borough", "X Coordinate (State Plane)", "Y Coordinate (State Plane)", "Open Data Channel Type", "Park Facility Name", "Park Borough", "Vehicle Type", "Taxi Company Borough", "Taxi Pick Up Location", "Bridge Highway Name", "Bridge Highway Segment", "Latitude", "Longitude", "Location"]
             }
     mutate {convert => ["Unique Key", "integer"]}
     mutate {convert => ["Created Date", "timestamp"]}
     mutate {convert => ["Closed Date", "timestamp"]}
     mutate {convert => ["Due Date", "timestamp"]}
     mutate {convert => ["Resolution Action Updated Date", "timestamp"]}
     mutate {convert => ["X Coordinate (State Plane)", "integer"]}
     mutate {convert => ["X Coordinate (State Plane)", "integer"]}
     mutate {convert => ["Latitude", "integer"]}
     mutate {convert => ["Longitude", "integer"]}
     mutate {convert => ["Location", "integer"]}
     }

output {
  elasticsearch {
      hosts => ["localhost:9200"]
      index => "311"
  }
  stdout {}
}

任何想法可能有什么问题吗？

Answer 1

您的配置中有两个错误，第一个是输出块中的错字，hosts行中的右花括号，这在错误日志中进行了描述。

exception => "LogStash:ConfigurationError"

这是错误的一行: hosts => ["localhost:9200"]}
这是固定配置

output {
  elasticsearch {
      hosts => ["localhost:9200"]
      index => "cars"
  }
  stdout {}
}

并且由于您正在运行Logstash 7.5，因此 document_type选项已从7.0版中删除。

第二个错误是在您的输入块中，即使在运行Windows时也应使用正斜杠，但直接指向 sincedb_path的 /dev/null/是Linux / macOS配置，在Windows上，您应使用 NUL。

这是正确的配置

input {
  file {
    path => "C:/ELK-Stack/Cars Data Set/cars.csv"
    start_position => "beginning"
    sincedb_path => "NUL"
  }
}