Я пытаюсь использовать Logstash для сбора данных о продажах для нескольких клиентов от разных поставщиков.
Пока у меня есть корзина S3 (Входящие), в которую я могу поместить свои файлы (в настоящее время CSV), и в соответствии с префиксом клиентского кода в файле данные помещаются в эластичный индекс для каждого клиента. Все это прекрасно работает.
У меня проблема в том, что у меня есть данные от нескольких поставщиков, и мне нужен способ определить, какой файл принадлежит тому или иному поставщику. Добавление дополнительного столбца в CSV не вариант, поэтому я планировал добавить его к имени файла, чтобы в итоге я получил соглашение об именах файлов, например clientcode_vendorcode_reportdate.csv
.
Я могу что-то упустить, но кажется, что на S3 я не могу получить доступ к имени файла внутри моей конфигурации, что кажется сумасшедшим, учитывая, что в какой-то момент префикс явно читается. Я собирался использовать фильтр Grok или Ruby, чтобы просто разделить имя файла на символ подчеркивания, что дало бы мне три ключевые переменные, которые я могу использовать в своей конфигурации, но пока все попытки не увенчались успехом. Я даже не могу получить полный source_path или имя файла в качестве переменной.
Моя конфигурация пока выглядит примерно так: я удалил неудачные попытки использования фильтров Grok/Ruby.
input {
s3 {
access_key_id =>"MYACCESSKEYID"
bucket => "sales-inbox"
backup_to_bucket => "sales-archive"
delete => "true"
secret_access_key => "MYSECRETACCESSKEY"
region => "eu-west-1"
codec => plain
prefix => "ABC"
type => "ABC"
}
s3 {
access_key_id =>"MYACCESSKEYID"
bucket => "sales-inbox"
backup_to_bucket => "sales-archive"
delete => "true"
secret_access_key => "MYSECRETACCESSKEY"
region => "eu-west-1"
codec => plain
prefix => "DEF"
type => "DEF"
}
}
filter {
if [type] == "ABC" {
csv {
columns => ["Date","Units","ProductID","Country"]
separator => ","
}
mutate {
add_field => { "client_code" => "abc" }
}
}
else if [type] == "DEF" {
csv {
columns => ["Date","Units","ProductID","Country"]
separator => ","
}
mutate {
add_field => { "client_code" => "def" }
}
}
mutate
{
remove_field => [ "message" ]
}
}
output {
elasticsearch {
codec => json
hosts => "myelasticcluster.com:9200"
index => "sales_%{client_code}"
document_type => "sale"
}
stdout { codec => rubydebug }
}
Мы будем очень признательны за рекомендации тех, кто хорошо разбирается в конфигурациях Logstash!
bin/plugin install --no-verify
для берите разветвленную версию. Вы можете увидеть ключевой коммит, который я сделал здесь: github.com/EssentialMusic/logstash- ввод-s3/фиксация/ - person Raoot   schedule 03.03.2016