脱Excel Elastic Stack（docker-compose）でcsvログを分析・可視化 - Elasticsearchのデータ重複エラー対処方法

はじめに

どうも！生産技術部で製品の検査工程を担当しているエンジニアです。Elasticsearchにデータを取り込む際にデータ重複のエラーが出ることがあります。その対処方法が公式ブログに記載されていましたので紹介します。

脱Excel Elastic Stack（docker-compose）でcsvログを分析・可視化 - Elastic Stackとはの続きになります。

対象読者

この記事では、Elastic Stackをそもそも知らない方、これから試そうと思っている方を対象としています。

この記事の内容

公式ブログ「Elasticsearchでのイベントベースのデータ重複を効果的に防止」の内容を実践したものです。

Filterの設定

FilterでユニークなドキュメントIDを生成します。IDはメッセージフィールドをMD5でハッシュ化し、タイムスタンプフィールドを１６進数に変換したものをprefixフィールドにセットしておき、これらをOutput時に連結すれば完了です。

注意すべきは、ブログのままだとInvalid FieldReference:'@metadata[prefix]'とエラーが出るため、以下のように変更しています。

logstash/pipeline/filter/filebeat_filter.cfg

  fingerprint {
    source => "message"
    target => "[@metadata][fingerprint]"
    method => "MD5"
    key => "test"
  }

  # ruby { code => "event.set('@metadata[prefix]', event.get('@timestamp').to_i.to_s(16))" }
  ruby { code => 'event.set("[@metadata][prefix]", event.get("@timestamp").to_i.to_s(16))' }

Outputの設定

先述の通りに、document_idに連結して設定します。

logstash/pipeline/output/filebeat_out.cfg

output {
  elasticsearch {
    hosts    => [ 'elasticsearch' ]
    index    => "%{[@metadata][beat]}-csv-%{+YYYY.MM.dd}"
    document_id => "%{[@metadata][prefix]}%{[@metadata][fingerprint]}"
  }
}

最後に

単純ですが、幅広いシチュエーションで適応できるので、重宝すると思います。