将数据从 CSV 文件输入到 logstash答案

【问题标题】：Input data from CSV file to logstash将数据从 CSV 文件输入到 logstash
【发布时间】：2015-09-13 15:32:37
【问题描述】：

我有一个 csv 文件，带有以下标题：

"PacketId","MACAddress","Date","PacketLength","SourceIP","SourcePort","DestIP","DestPort"

我想使用 LogStash 将数据索引到 ElasticSearch，但无法为其编写过滤器。

filter {
    grok {
        match => message => "%{IP:SourceIP}"
    }
    }

上面的过滤器很好地提取了 SourceIP 字段，但是我如何编写 grok 模式来为所有字段提取它。

【问题讨论】：

标签： csv elasticsearch logstash-grok logstash-configuration

【解决方案1】：

让下面的CSV文件：

1,00-14-22-01-23-45,13/09/2015,32,128.248.1.43,9980,128.248.23.13,9880
1,01-74-02-84-13-98,14/09/2015,64,128.248.1.94,9280,128.248.13.84,9380

这里必须设置Logstash配置：

input {
    file {
        path => "/path/of/your/csv/test.csv"
        sincedb_path => "/path/of/your/csv/test.idx"
        start_position => "beginning"
    }
}

filter {
    csv {
        separator => ","
        columns => ["PacketId","MACAddress","Date","PacketLength","SourceIP","SourcePort","DestIP","DestPort"]
    }
}

output {
    stdout {
        codec => rubydebug      
    }
}

你会得到输出结果：

{
         "message" => [
        [0] "1,00-14-22-01-23-45,13/09/2015,32,128.248.1.43,9980,128.248.23.13,9880"
    ],
        "@version" => "1",
      "@timestamp" => "2015-09-14T20:11:28.976Z",
            "host" => "MyHost.local",
            "path" => "/path/of/your/csv/test.csv",
        "PacketId" => "1",
      "MACAddress" => "00-14-22-01-23-45",
            "Date" => "13/09/2015",
    "PacketLength" => "32",
        "SourceIP" => "128.248.1.43",
      "SourcePort" => "9980",
          "DestIP" => "128.248.23.13",
        "DestPort" => "9880"
}
{
         "message" => [
        [0] "1,01-74-02-84-13-98,14/09/2015,64,128.248.1.94,9280,128.248.13.84,9380"
    ],
        "@version" => "1",
      "@timestamp" => "2015-09-14T20:11:28.978Z",
            "host" => "MyHost.local",
            "path" => "/path/of/your/csv/test.csv",
        "PacketId" => "1",
      "MACAddress" => "01-74-02-84-13-98",
            "Date" => "14/09/2015",
    "PacketLength" => "64",
        "SourceIP" => "128.248.1.94",
      "SourcePort" => "9280",
          "DestIP" => "128.248.13.84",
        "DestPort" => "9380"
}

问候，阿兰

【讨论】：

【解决方案2】：

你需要先使用CSV filter，而不是grok。

【讨论】：