Logstash不将csv数据加载到弹性搜索

Question

我无法用logstash将数据加载到localhost上的弹性搜索节点...我想让logstash读取csv文件并将这些数据加载到弹性搜索。但是什么都不起作用，我只能读取我手动添加到弹性搜索中的数据，看起来logstash什么都不做。

我的logstash配置是：

input {
    file {
        path => [ "C:\Users\Michele\Downloads\logstash-1.5.3\logstash-1.5.3\Users\*.csv" ]
        
        start_position => "beginning"
    }
}

filter {
    csv {
        columns => ["timestamp", "impianto", "tipo_misura", "valore","unita_misura"]
        separator => ","
    }
}

output {
    elasticsearch {
        action => "index"
        host => "localhost"
        cluster => "elasticsearch"
        node_name => "NCC-1701-A"
        index => "myindex"
        index_type => "pompe"
        workers => 1
    }
}

我的csv文件是：

2015-08-03T18:46:00,Abbiategrasso,Pressione gruppo 1,55.5,m
2015-08-03T18:46:10,Abbiategrasso,Pressione gruppo 1,44.4,m
2015-08-03T18:46:20,Abbiategrasso,Pressione gruppo 1,66.6,m
2015-08-03T18:46:30,Abbiategrasso,Pressione gruppo 1,33.3,m
2015-08-03T18:46:40,Abbiategrasso,Pressione gruppo 1,22.2,m
2015-08-03T18:46:50,Abbiategrasso,Pressione gruppo 1,77.7,m
2015-08-03T18:47:00,Abbiategrasso,Pressione gruppo 1,11.1,m
2015-08-03T18:47:10,Abbiategrasso,Pressione gruppo 1,44.4,m
2015-08-03T18:47:20,Abbiategrasso,Pressione gruppo 1,55.5,m
2015-08-03T18:47:30,Abbiategrasso,Pressione gruppo 1,33.3,m
2015-08-03T18:47:40,Abbiategrasso,Pressione gruppo 1,22.2,m
2015-08-03T18:47:50,Abbiategrasso,Pressione gruppo 1,66.6,m
2015-08-03T18:48:00,Abbiategrasso,Pressione gruppo 1,33.3,m
2015-08-03T18:48:10,Abbiategrasso,Pressione gruppo 1,77.7,m
2015-08-03T18:48:20,Abbiategrasso,Pressione gruppo 1,22.2,m
2015-08-03T18:48:30,Abbiategrasso,Pressione gruppo 1,88.8,m
2015-08-03T18:48:40,Abbiategrasso,Pressione gruppo 1,55.5,m
2015-08-03T18:48:50,Abbiategrasso,Pressione gruppo 1,33.3,m
2015-08-03T18:49:00,Abbiategrasso,Pressione gruppo 1,55.5,m

“MyIndex”索引上没有新的内容出现！但我不知道为什么...

将logstash配置文件从不获取文件的“*.csv”更改为“abbiategrasso.csv”

logstash“立即尝试”上的My--Debug消息是：

←[33mfailed action with response of 400, dropping action: ["index", {:_id=>nil,
:_index=>"abbiategrasso", :_type=>"pompe", :_routing=>nil}, #<LogStash::Event:0x
1cea7b7 @metadata_accessors=#<LogStash::Util::Accessors:0x1e577ee @store={"path"
=>"C:\\Users\\Michele\\Downloads\\logstash-1.5.3\\logstash-1.5.3\\Users\\abbiate
grasso.csv", "retry_count"=>0}, @lut={"[path]"=>[{"path"=>"C:\\Users\\Michele\\D
ownloads\\logstash-1.5.3\\logstash-1.5.3\\Users\\abbiategrasso.csv", "retry_coun
t"=>0}, "path"]}>, @cancelled=false, @data={"message"=>["2015-08-03T18:48:00,Abb
iategrasso,Pressione gruppo 1,66.6,m\r"], "@version"=>"1", "@timestamp"=>"2015-0
9-07T17:28:32.501Z", "host"=>"Michele-HP", "path"=>"C:\\Users\\Michele\\Download
s\\logstash-1.5.3\\logstash-1.5.3\\Users\\abbiategrasso.csv", "timestamp"=>"2015
-08-03T18:48:00", "impianto"=>"Abbiategrasso", "tipo_misura"=>"Pressione gruppo
1", "valore"=>"66.6", "unita_misura"=>"m"}, @metadata={"path"=>"C:\\Users\\Miche
le\\Downloads\\logstash-1.5.3\\logstash-1.5.3\\Users\\abbiategrasso.csv", "retry
_count"=>0}, @accessors=#<LogStash::Util::Accessors:0x2ff785 @store={"message"=>
["2015-08-03T18:48:00,Abbiategrasso,Pressione gruppo 1,66.6,m\r"], "@version"=>"
1", "@timestamp"=>"2015-09-07T17:28:32.501Z", "host"=>"Michele-HP", "path"=>"C:\
\Users\\Michele\\Downloads\\logstash-1.5.3\\logstash-1.5.3\\Users\\abbiategrasso
.csv", "timestamp"=>"2015-08-03T18:48:00", "impianto"=>"Abbiategrasso", "tipo_mi
sura"=>"Pressione gruppo 1", "valore"=>"66.6", "unita_misura"=>"m"}, @lut={"host
"=>[{"message"=>["2015-08-03T18:48:00,Abbiategrasso,Pressione gruppo 1,66.6,m\r"
], "@version"=>"1", "@timestamp"=>"2015-09-07T17:28:32.501Z", "host"=>"Michele-H
P", "path"=>"C:\\Users\\Michele\\Downloads\\logstash-1.5.3\\logstash-1.5.3\\User
s\\abbiategrasso.csv", "timestamp"=>"2015-08-03T18:48:00", "impianto"=>"Abbiateg
rasso", "tipo_misura"=>"Pressione gruppo 1", "valore"=>"66.6", "unita_misura"=>"
m"}, "host"], "path"=>[{"message"=>["2015-08-03T18:48:00,Abbiategrasso,Pressione
 gruppo 1,66.6,m\r"], "@version"=>"1", "@timestamp"=>"2015-09-07T17:28:32.501Z",
 "host"=>"Michele-HP", "path"=>"C:\\Users\\Michele\\Downloads\\logstash-1.5.3\\l
ogstash-1.5.3\\Users\\abbiategrasso.csv", "timestamp"=>"2015-08-03T18:48:00", "i
mpianto"=>"Abbiategrasso", "tipo_misura"=>"Pressione gruppo 1", "valore"=>"66.6"
, "unita_misura"=>"m"}, "path"], "message"=>[{"message"=>["2015-08-03T18:48:00,A
bbiategrasso,Pressione gruppo 1,66.6,m\r"], "@version"=>"1", "@timestamp"=>"2015
-09-07T17:28:32.501Z", "host"=>"Michele-HP", "path"=>"C:\\Users\\Michele\\Downlo
ads\\logstash-1.5.3\\logstash-1.5.3\\Users\\abbiategrasso.csv", "timestamp"=>"20
15-08-03T18:48:00", "impianto"=>"Abbiategrasso", "tipo_misura"=>"Pressione grupp
o 1", "valore"=>"66.6", "unita_misura"=>"m"}, "message"], "timestamp"=>[{"messag
e"=>["2015-08-03T18:48:00,Abbiategrasso,Pressione gruppo 1,66.6,m\r"], "@version
"=>"1", "@timestamp"=>"2015-09-07T17:28:32.501Z", "host"=>"Michele-HP", "path"=>
"C:\\Users\\Michele\\Downloads\\logstash-1.5.3\\logstash-1.5.3\\Users\\abbiategr
asso.csv", "timestamp"=>"2015-08-03T18:48:00", "impianto"=>"Abbiategrasso", "tip
o_misura"=>"Pressione gruppo 1", "valore"=>"66.6", "unita_misura"=>"m"}, "timest
amp"], "impianto"=>[{"message"=>["2015-08-03T18:48:00,Abbiategrasso,Pressione gr
uppo 1,66.6,m\r"], "@version"=>"1", "@timestamp"=>"2015-09-07T17:28:32.501Z", "h
ost"=>"Michele-HP", "path"=>"C:\\Users\\Michele\\Downloads\\logstash-1.5.3\\logs
tash-1.5.3\\Users\\abbiategrasso.csv", "timestamp"=>"2015-08-03T18:48:00", "impi
anto"=>"Abbiategrasso", "tipo_misura"=>"Pressione gruppo 1", "valore"=>"66.6", "
unita_misura"=>"m"}, "impianto"], "tipo_misura"=>[{"message"=>["2015-08-03T18:48
:00,Abbiategrasso,Pressione gruppo 1,66.6,m\r"], "@version"=>"1", "@timestamp"=>
"2015-09-07T17:28:32.501Z", "host"=>"Michele-HP", "path"=>"C:\\Users\\Michele\\D
ownloads\\logstash-1.5.3\\logstash-1.5.3\\Users\\abbiategrasso.csv", "timestamp"
=>"2015-08-03T18:48:00", "impianto"=>"Abbiategrasso", "tipo_misura"=>"Pressione
gruppo 1", "valore"=>"66.6", "unita_misura"=>"m"}, "tipo_misura"], "valore"=>[{"
message"=>["2015-08-03T18:48:00,Abbiategrasso,Pressione gruppo 1,66.6,m\r"], "@v
ersion"=>"1", "@timestamp"=>"2015-09-07T17:28:32.501Z", "host"=>"Michele-HP", "p
ath"=>"C:\\Users\\Michele\\Downloads\\logstash-1.5.3\\logstash-1.5.3\\Users\\abb
iategrasso.csv", "timestamp"=>"2015-08-03T18:48:00", "impianto"=>"Abbiategrasso"
, "tipo_misura"=>"Pressione gruppo 1", "valore"=>"66.6", "unita_misura"=>"m"}, "
valore"], "unita_misura"=>[{"message"=>["2015-08-03T18:48:00,Abbiategrasso,Press
ione gruppo 1,66.6,m\r"], "@version"=>"1", "@timestamp"=>"2015-09-07T17:28:32.50
1Z", "host"=>"Michele-HP", "path"=>"C:\\Users\\Michele\\Downloads\\logstash-1.5.
3\\logstash-1.5.3\\Users\\abbiategrasso.csv", "timestamp"=>"2015-08-03T18:48:00"
, "impianto"=>"Abbiategrasso", "tipo_misura"=>"Pressione gruppo 1", "valore"=>"6
6.6", "unita_misura"=>"m"}, "unita_misura"]}>>] {:level=>:warn, :file=>"/Users/M
ichele/Downloads/logstash-1.5.3/logstash-1.5.3/vendor/bundle/jruby/1.9/gems/logs
tash-output-elasticsearch-1.0.5-java/lib/logstash/outputs/elasticsearch.rb", :li
ne=>"531", :method=>"submit"}←[0m

Logstash无法上载数据...

如果使用Marvel.Sense手动记录了数据，请从索引中删除当前数据。为此，请使用索引的DELETE命令。下一次，logstash将能够创建一个符合其要求的新索引！

同样，对于logstash配置文件中的文件定义，不要使用“*.csv”，它不起作用...它不搜索文件夹中的所有文件，它搜索命名文件如下所示：（

我现在的配置是：

input {
    file {
        path => [ "C:\Users\Michele\Downloads\logstash-1.5.3\logstash-1.5.3\Users\abbiategrasso4.csv" ]
        
        start_position => "beginning"
    }
}

filter {
    csv {
        columns => ["timestamp", "impianto", "tipo_misura", "valore","unita_misura"]
        separator => ","
    }
    
    mutate {
        convert => { "valore" => "float" }
    }
}

output {
    elasticsearch {
        action => "index"
        host => "localhost"
        cluster => "elasticsearch"
        node_name => "NCC-1701-A"
        index => "abbiategrasso"
        document_type => "pompe"
        workers => 1
    }
    
    stdout { codec => rubydebug }
}

现在起作用了，我会尝试Kibana：）

Answer 1

当您进行故障排除时，我将从打印到标准输出开始。

output { stdout { codec => rubydebug } }

这样，您就可以看到问题是出在logstash还是出在ElasticSearch的接口上。

我还发现在执行文件输入时使用sincedb非常有用。可能是在排除故障时，读取了数据并修改了sincedb。在进行故障排除时，将以下行添加到文件{}中。但是，当您进入实际使用时，不要忘记删除它，否则您将重新摄取您不打算摄取的数据。

sincedb_path => "/dev/null"