elasticsearch - 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}"
  }
}

当我们使用这个时，我们正在使用上面的配置，空值字段被删除而不是跳过空值更新。

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

XML 日志格式:GEO-1|CD|23|John|892|Canada|31-01-2017|QC|-|-|-|-|- JSON 日志格式:GEO-1|AS|33|-|-|-|-|-|Mike|123|US|31-01-2017|QC

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

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

Answer 1

对于 Elasticsearch 输出执行除 index 之外的任何操作 you need to tell it to do something else .

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 行为。