我正在尝试从一些日志文件中加载/摄取数据，这些日志文件几乎是存储在某些第三供应商数据库中的数据的副本。数据是管道分隔的“键值”值，我可以使用 logstash 中的 kv 过滤器插件将其拆分。
样本数据 -
1.) 表="贸易"| TradeID="1234" |数量=100|价格=100.00|BuyOrSell="BUY"|Stock="ABCD Inc."
如果我们收到对上述记录的修改，
2.) 表="贸易"| 贸易ID="1234" | 数量=120|价格=101.74 |BuyOrSell="BUY"|Stock="ABCD Inc."
我们需要更新在第一个条目上创建的记录。所以，我需要将 TradeID 设为 id 字段并需要更新记录，这样就不会出现相同的 TradeID 记录重复。
logstash.conf 的代码有点像下面 -
input {
file {
path => "some path"
}
}
filter {
kv {
source => "message"
field_split => "\|"
value_split => "="
}
}
output {
elasticsearch {
hosts => ["https://localhost:9200"]
cacert => "path of .cert file"
ssl => true
ssl_certificate_verification => true
index => "trade-index"
user => "elastic"
password => ""
}
}
您需要更新您的elasticsearch输出，如下所示：
output {
elasticsearch {
hosts => ["https://localhost:9200"]
cacert => "path of .cert file"
ssl => true
ssl_certificate_verification => true
index => "trade-index"
user => "elastic"
password => ""
# add the following to make it work as an upsert
action => "update"
document_id => "%{TradeID}"
doc_as_upsert => true
}
}
因此当 Logstash 读取第一笔交易时，ID 为 1234 的文档将不存在，将被更新插入（即创建）。当第二笔交易被读取时，该文件存在并且将被简单地更新。
本文收集自互联网，转载请注明来源。
如有侵权，请联系 [email protected] 删除。
我来说两句