Logstash 在 elasticsearch 中更新文档

Question

尝试通过 logstash 更新 elasticsearch 中的特定字段。是否可以通过 logstash 仅更新一组字段？

请找到下面的代码，

    input {
  file {
        path => "/**/**/logstash/bin/*.log"
        start_position => "beginning"
        sincedb_path => "/dev/null"
        type => "multi"
  }
}

filter {
        csv {
                separator => "|"
                columns => ["GEOREFID","COUNTRYNAME", "G_COUNTRY", "G_UPDATE", "G_DELETE",  "D_COUNTRY", "D_UPDATE", "D_DELETE"]

}
elasticsearch {
         hosts => ["localhost:9200"]
         index => "logstash-data-monitor"
         query => "GEOREFID:%{GEOREFID}"
         fields => [["JSON_COUNTRY","G_COUNTRY"],
                    ["XML_COUNTRY","D_COUNTRY"]]
           }

if [G_COUNTRY] {
mutate {
    update => { "D_COUNTRY" => "%{D_COUNTRY}"
 }
  }
}
}
output {
elasticsearch {
        hosts => ["localhost:9200"]
        index => "logstash-data-monitor"
        document_id => "%{GEOREFID}"
  }
}

我们使用上述配置时，空值字段将被删除，而不是跳过空值更新。

数据来自 2 个不同的来源。一个来自 XML 文件，另一个来自 JSON 文件。

XML 日志格式：GEO-1|CD|23|John|892|加拿大|31-01-2017|QC|-|-|-|-|- JSON 日志格式：GEO-1|AS|33|-|- |-|-|-|迈克|123|美国|31-01-2017|QC

添加一个日志时，将在索引中创建新文档。读取第二个日志文件时，现有文档应该得到更新。如果日志文件是 XML，更新应该只发生在前 5 个字段中，如果日志文件是 JSON，则更新应该发生在最后 5 个字段中。请建议我们如何在 logstash 中执行此操作。

用上面的代码试过了。请检查，任何人都可以帮助解决这个问题吗？

Answer 1

要让 Elasticsearch 输出执行任何操作，index 您需要告诉它执行其他操作。

elasticsearch {
        hosts => ["localhost:9200"]
        index => "logstash-data-monitor"
        action => "update"
        document_id => "%{GEOREFID}"
}

这可能应该包含在条件中，以确保您只更新需要更新的记录。不过，还有另一种选择，doc_as_upsert

elasticsearch {
        hosts => ["localhost:9200"]
        index => "logstash-data-monitor"
        action => "update"
        doc_as_upsert => true
        document_id => "%{GEOREFID}"
}

这告诉插件如果它是新的就插入，如果不是则更新。

但是，您尝试使用两个输入来定义文档。这让事情变得复杂。此外，您没有提供两种输入，所以我会即兴发挥。要提供不同的输出行为，您需要定义两个输出。

input {
  file {
    path => "/var/log/xmlhome.log"
    [other details]
  }

  file {
    path => "/var/log/jsonhome.log"
    [other details]
  }
}

filter { [some stuff ] }

output {
  if [path] == '/var/log/xmlhome.log' {
    elasticsearch {
      [XML file case]
    }
  } else if [path] == '/var/log/jsonhome.log' {
    elasticsearch {
      [JSON file case]
      action => "update"
    }
  }
}

像这样设置将允许您根据事件发生的位置更改 ElasticSearch 行为。