Learn how to use Logstash to safely route a subset of production data to pre-production clusters by using efficient sampling techniques and by leveraging the UDP protocol’s fire-and-forget functionality.

Risks of driving data into multiple destinations

Overcoming the drawbacks of the output isolator pattern

Caveats: When to use the pipeline-to-pipeline with UDP solution

Pipeline-to-pipeline with UDP overview

Sampling the production data with the Ruby filter

Pipeline configuration

Running the code

Example output

Common pipeline output (simulating production data flow)

Pre-production pipeline output (simulating pre-production data flow)

Benefits of the pipeline-to-pipeline with UDP solution

Easy debugging

Alternative approach: Sampling with the Drop filter

Discover efficient sampling and fire-and-forget pipelines

Safely route a subset of production data to pre-production clusters by leveraging UDP’s fire-and-forget functionality

158175 - Blog header image_V1.jpg

Safely sample production data into pre-production environments with Logstash

減量とリサイクルで、コードもサステナブルに。モジュールコンポーネントでLogstash投入パイプラインを作成する方法をご紹介します。

<a href="/jp/products/logstash">Logstash</a>は、オープンソースのデータ処理パイプラインです。1つ以上のインプットからくるイベントを投入および変換し、各イベントを1つ以上のアウトプット先に送ることができます。ケースによっては、複数のインプットソースから来るイベントを処理するため、Logstashに多数のコード行を追加して実装していることもあります。この記事では、保守性にすぐれた実装にするため、モジュールコンポーネントでパイプラインを作成して、コードを再利用しやすくする方法をご紹介します。
 <h2>背景</h2>Logstashで、複数のインプットソースからくるイベントに共通のロジックサブセットを適用することが必要となるケースはよくあります。こうした適用は一般的に、次の2つの方法で行われています。&nbsp;
<ol>
	<li>異なる複数のインプットソースからくるイベントを1つのパイプラインで処理し、すべてのソースのイベントに共通のロジックを簡単に適用できるようにします。このパターンの実装では通常、共通のロジックのほかに、かなり多くの条件付きロジックが用いられます。したがってこのアプローチによるLogstashの実装は、複雑でわかりにくいものになる可能性があります。</li>
	<li>イベントの処理を、インプットソースごとに個別のパイプラインで実行します。このアプローチでは、各パイプラインに共通の機能をコピーして複製を保つ必要があり、コードの共通部分を維持することが困難です。&nbsp;</li>
</ol>今回ご紹介するのは、上記のアプローチの欠点を解決するテクニックです。モジュールコンポーネントを異なるファイルに格納し、それらのコンポーネントを組み合わせてパイプラインを構築するというやり方です。このテクニックを使うとパイプラインはさほど複雑にならず、コードの複製も回避することができます。
 <h2>モジュール式のパイプライン構築</h2>Logstash<a href="/guide/en/logstash/current/configuration.html">設定ファイル</a>は、Logstashパイプラインが実行するインプット、フィルター、およびアウトプットで構成されています。&nbsp; 高度なセットアップでは、<a href="/guide/en/logstash/current/multiple-pipelines.html">複数のパイプラインを実行するLogstashインスタンス</a>を使用することが一般的です。引数を使わずデフォルトでLogstashを起動すると、Logstashは<code>pipelines.yml</code>と呼ばれるファイルを読み込み、指定されたパイプラインをインスタンス化します。&nbsp;
Logstashのインプット、フィルター、およびアウトプットは複数のファイルに格納でき、<a href="/guide/en/logstash/current/glob-support.html">glob表現</a>を指定することで、パイプラインに含める選択ファイルとすることができます。glob表現に一致するファイルは、アルファベット順に組み合わされます。しかし、フィルターを実行する順番が重要な意味を持つことは少なくありません。そこで、ファイル名に数的な識別子を含めておき、ファイルの組み合わせが確実に望ましい順番で行われるようにすると便利です。
以下は、複数のLogstashモジュールコンポーネントを組み合わせて、2つの独立のパイプラインを定義する場合の例です。Logstashコンポーネントを、次のファイルに格納します。
<ul>
	<li>インプット宣言：<code>01_in.cfg</code>, <code>02_in.cfg</code></li>
	<li>フィルター宣言：<code>01_filter.cfg</code>, <code>02_filter.cfg</code>, <code>03_filter.cfg</code></li>
	<li>アウトプット宣言：<code>01_out.cfg</code></li>
</ul>以下のように<a href="/guide/en/logstash/current/glob-support.html">glob表現</a>を使用し、次に<code>pipelines.yml</code>に定義するパイプラインが、望ましいコンポーネントで構成されるようにします。&nbsp;
<pre class="prettyprint">- pipeline.id: my-pipeline_1
 path.config: "&lt;path&gt;/{01_in,01_filter,02_filter,01_out}.cfg"
- pipeline.id: my-pipeline_2
 path.config: "&lt;path&gt;/{02_in,02_filter,03_filter,01_out}.cfg"
</pre>上記のパイプライン設定で、<code>02_filter.cfg</code>というファイルが双方のパイプラインに存在しています。これが、2つのパイプラインに共通するコードを1つのファイルで定義および保守し、さらに複数のパイプラインで実行させる仕組みです。
 <h2>パイプラインをテストする</h2>このセクションでは、先ほど<code>pipelines.yml</code>で定義した個別のパイプラインで、組み合わさるファイルの具体的な例を示します。次に、そのファイルを使ってLogstashを実行し、生成されたアウトプットを見てみます。&nbsp;
 <h3>設定ファイル</h3>インプットファイル：01_in.cfg
このファイルは、<a href="/guide/en/logstash/current/plugins-inputs-generator.html">ジェネレーター</a>によるインプットを定義します。ジェネレーターインプットはLogstashのテスト用に設計されています。今回のケースでは、単一のイベントを生成します。&nbsp;
<pre class="prettyprint">input { 
 generator { 
 lines =&gt; ["Generated line"] 
 count =&gt; 1 
 } 
}
</pre>インプットファイル：02_in.cfg
このファイルは、stdinをリッスンするLogstashインプットを定義します。
<pre class="prettyprint">input { 
 stdin {}
}
</pre>フィルターファイル：01_filter.cfg
<pre class="prettyprint">filter { 
 mutate { 
 add_field =&gt; { "filter_name" =&gt; "Filter 01" } 
 } 
}
</pre>フィルターファイル：02_filter.cfg
<pre class="prettyprint">filter { 
 mutate { 
 add_field =&gt; { "filter_name" =&gt; "Filter 02" } 
 } 
}
</pre>フィルターファイル：03_filter.cfg
<pre class="prettyprint">filter { 
 mutate { 
 add_field =&gt; { "filter_name" =&gt; "Filter 03" } 
 } 
}
</pre>アウトプットファイル：01_out.cfg
<pre class="prettyprint">output { 
 stdout { codec =&gt; "rubydebug" } 
}
</pre> <h3>パイプラインを実行する</h3>オプションを使用せずLogstashを起動すると、上で定義した<code>pipelines.yml</code>ファイルが実行されます。次のようにLogstashを実行します：&nbsp;
<pre class="prettyprint">./bin/logstash
</pre><code>my-pipeline_1</code>というパイプラインは、インプットイベントをシミュレーションするジェネレーターを実行しています。Logstashが初期化を完了するとすぐに、以下のようにアウトプットを確認できるはずです。この結果からは、<code>01_filter.cfg</code>および<code>02_filter.cfg</code>のコンテンツがこのパイプラインによって期待通りに実行されたことがわかります。
<pre class="prettyprint">{ 
 "sequence" =&gt; 0,
 "host" =&gt; "alexandersmbp2.lan",
 "message" =&gt; "Generated line",
 "@timestamp" =&gt; 2020-02-05T22:10:09.495Z,
 "@version" =&gt; "1",
 "filter_name" =&gt; [ 
 [0] "Filter 01",
 [1] "Filter 02" 
 ] 
}
</pre>もう一方のパイプライン、<code>my-pipeline_2</code>はstdinのインプットを待機しており、このパイプラインで処理されたイベントはまだ確認できません。Logstashを実行している端末に何か入力し、[Return]（または[Enter]）キーを押して、このパイプライン用のイベントを作成してみましょう。この手順を完了すると、次のような結果になるはずです。&nbsp;
<pre class="prettyprint">{ 
 "filter_name" =&gt; [ 
 [0] "Filter 02",
 [1] "Filter 03" 
 ],
 "host" =&gt; "alexandersmbp2.lan",
 "message" =&gt; "I’m testing my-pipeline_2",
 "@timestamp" =&gt; 2020-02-05T22:20:43.250Z,
 "@version" =&gt; "1" 
}
</pre>上の結果から、<code>02_filter.cfg</code>および<code>03_filter.cfg</code>のロジックが期待通りに適用されていることがわかります。&nbsp;&nbsp;
 <h3>実行の順番</h3>Logstashはglob表現の中のファイルの順番を考慮しない点に注意してください。glob表現は、どのファイルを含めるかの判断に使用され、ファイルをアルファベット順に並べるだけです。たとえば<code>my-pipeline_2</code>の定義を変えて、glob表現で<code>03_filter.cfg</code>が<code>02_filter.cfg</code>より先に現れるようにしても、各イベントは<code>03_filter.cfg</code>に定義されたフィルターより先に<code>02_filter.cfg</code>のフィルターを通過します。
 <h2>まとめ</h2>glob表現を使うと、個別のファイルとして格納できるモジュールコンポーネントでLogstashパイプラインを構成することができます。この方法で、コードの保守性や、再利用のしやすさ、可読性が向上します。&nbsp;
補足として、今回の記事で取り上げたテクニックのほかに、<a href="/guide/en/logstash/current/pipeline-to-pipeline.html">パイプラインツーパイプライン通信</a>も検討すべきポイントです。パイプラインツーパイプライン通信は、Logstashの実装のモジュール性をさらに高める可能性を秘めています。

blog-pipe-sky-banner.jpg

blog-pipe-sky-thumb.jpg

How to create maintainable and reusable Logstash pipelines

保守性と再利用性にすぐれたLogstashパイプラインの作り方

Elasticsearchクラスター内のデータと既存のリレーショナルデータベースの同期を維持する方法について説明します。

es-database-sync-banner.jpg

es-database-sync-thumb.jpg

How to keep Elasticsearch synchronized with a relational database using Logstash and JDBC

LogstashおよびJDBCを使用してElasticsearchとリレーショナルデータベースの同期を維持する方法

<a href="/jp/logstash">Logstash</a>は、オープンソースのサーバーサイドデータ処理パイプラインです。Logstashでデータを投入、変換し、1つ以上のアウトプット先に送ることができます。本ブログ記事ではLogstashを使って複数の株式市場のデータを投入し、各市場に対応する個別のアウトプットに送る方法をご紹介します。実行する手順は次の通りです：

<ol>
	<li>株式市場のインプットストリームから来るドキュメントのコピーを作成する。</li>
	<li>各コピーをフィルタリングし、該当の市場で有効なフィールドのみを持たせる。</li>
	<li>各コピーにメタデータを追加し、どの株式市場のデータが入っているかわかるようにする。</li>
	<li>各ドキュメントのメタデータから判断し、ドキュメントを適切なアウトプットに送る</li>
</ol>
留意点として、本ブログ記事が説明する機能の一部をアーカイブすると考えられる<a href="/guide/en/logstash/6.5/pipeline-to-pipeline.html">パイプラインツーパイプライン通信（6.5でベータリリース）</a>は、本記事で紹介する手順では使用していません。

<h2>インプットファイル例</h2>
Logstashにインプットするファイルには、株式市場でベンチーマークとなる指数を含むCSVファイルを使用します。CSVエントリの一例として、以下のようなものがあります：

<pre class="prettyprint">1483230600,1628.75,1678.1,1772.8,2443.6
1483232400,1613.63,1688.5,1750.5,2460.2
1483234200,1606.51,1678.6,1718,2448.2
1483236000,1621.04,1684.1,1708.1,2470.4
</pre>
コンマで区切られた値は、"time"（時刻）と、"DAX"、"SMI"、"CAC"、"FTSE"の各市場の指数となる値を表します。上の4行を"stocks.csv"と呼ばれるCSVファイルにコピー＆ペーストすると、このLogstashパイプライン例でインプットとして使うことができます。

<h2>Logstashパイプライン例</h2>
本セクションで後ほど示す記述は、次の動作を実行するLogstashパイプラインです。

<ol>
	<li>CSVファイルから、CSV形式のインプットとして株式市場の値を読み込む。</li>
	<li>CSVインプットの各行をJSONドキュメントにマップする（CSV列を"time"、"DAX"、"SMI"、"CAC"、"FTSE"の各JSONフィールドにマップする）。</li>
	<li>timeフィールドをUnixフォーマットに変換する。</li>
	<li><a href="/guide/en/logstash/6.5/plugins-filters-clone.html">clone filterプラグイン</a>を使用し、（元のドキュメントに加えて）各ドキュメントにつき2つコピーを作成する。"clone filter"は各ドキュメントのコピーに新規の"type"フィールドを自動で追加する。"type"は所与のクローンの配列名に対応する。（今回はtypeが"clone_for_SMI"や"clone_for_FTSE"となるよう定義し、各クローンが最終的に"SMI"または"FTSE"のいずれかの市場から来るデータだけを持つように設定しています。）</li>
	<li>各クローンに対して次の動作を実行する：</li>
	<ol>
		<li><a href="/guide/en/logstash/6.5/plugins-filters-prune.html">prune filterプラグイン</a>を使用して、特定の市場向けにホワイトリスト化されたフィールドを除くすべてのフィールドを削除する。</li>
		<li>各ドキュメントに、クローン関数が追加した"type"に対応する<a href="/jp/blog/logstash-metadata">メタデータ</a>を追加する。（prune関数を使ってclone関数が挿入した"type"を削除していますが、"type"情報は正しいアウトプットにドキュメントを送る際に必要となることから、この手順が必要となります。）</li>
	</ol>
	<li><a href="/guide/en/logstash/6.5/plugins-outputs-elasticsearch.html">Logstash用のElasticsearchアウトプットプラグイン</a>を使用して、異なるElasticsearchアウトプットに各市場のドキュメントを書き出す。アウトプットの定義は、手順5で追加したメタデータフィールドの値に基づく。（コードをシンプル化するため、各ElasticsearchアウトプットはローカルのElasticsearchクラスター内で一意のインデックスに書き込む設定となっています。複数のクラスターをアウトプットとして使用する必要がある場合、各Elasticsearchアウトプットの宣言は、一意の<a href="/guide/en/logstash/6.5/plugins-outputs-elasticsearch.html#plugins-outputs-elasticsearch-hosts">Elasticsearchホスト</a>を指定して簡単に修正することができます。）</li>
</ol>
次の記述が、上の手順を実行するLogstashパイプラインです（対応する手順の番号が、コメント形式で挿入されています）。&nbsp;このパイプラインを実行するには、"clones.conf"というファイルに次の記述をコピーします。

<pre class="prettyprint">## 手順1
input {
 file {
 # stocks.csvファイルを確実に使用できるよう、パスを編集してください
 path =&gt; "${HOME}/stocks.csv"
 # 以下でLogstashの実行のたびに完全なインプットの 
 # 再読み込みを確実に実行します（デバッグに役立ちます）
 start_position =&gt; "beginning"
 sincedb_path =&gt; "/dev/null"
 }
}
## 手順2
filter {
 csv {
 columns =&gt; ["time","DAX","SMI","CAC","FTSE"]
 separator =&gt; ","
 convert =&gt; { 
 'DAX' =&gt; 'float'
 'SMI' =&gt; 'float'
 'CAC' =&gt; 'float'
 'FTSE' =&gt; 'float'
 }
 }
## 手順3 
date {
 match =&gt; ['time', 'UNIX']
 }
## 手順4
 # 以下の行は各ドキュメントにつき2つずつ 
 # コピーを作成します（オリジナルを含め合計3つ）。 
 # 各コピーには自動的に配列の
 # 名前に対応する"type"が 
 # 追加されます
 clone {
 clones =&gt; ['clone_for_SMI', 'clone_for_FTSE']
 }
## 手順5
 if [type] == 'clone_for_SMI' {
 # "SMI"以外のすべてを削除します
 prune {
 whitelist_names =&gt; [ "SMI"]
 }
 mutate {
 add_field =&gt; { "[@metadata][type]" =&gt; "only_SMI" } 
 }
 } 
 else if [type] == 'clone_for_FTSE' {
 prune {
 whitelist_names =&gt; [ "FTSE"]
 }
 mutate {
 add_field =&gt; { "[@metadata][type]" =&gt; "only_FTSE" } 
 }
 } 
}
## 手順6
output {
 # 以下のstdoutへのアウトプットは 
 # デバッグ目的であり、削除できます
 stdout { 
 codec =&gt; rubydebug {
 metadata =&gt; true
 }
 }
 if [@metadata][type] == 'only_SMI' {
 elasticsearch {
 index =&gt; "smi_data"
 }
 }
 else if [@metadata][type] == 'only_FTSE' {
 elasticsearch {
 index =&gt; "ftse_data"
 }
 }
 else {
 elasticsearch {
 index =&gt; "stocks_original"
 }
 }
}
</pre>
<h2>Logstashパイプラインをテストする</h2>
事例のCSVデータを投入するパイプラインをテストするには、次のコマンドを実行します。その際、必ずパスを修正し、お使いのシステムで実際に使用した正しいパスとなるようにしてください。留意点として、"config.reload.automatic"の指定はオプションです。このオプションには、Logstashを再起動せずに"clones.conf"を自動でリロードできるというメリットがあります。

<pre class="prettyprint">./logstash -f ./clones.conf --config.reload.automatic
</pre>
Logstashが"stocks.csv"ファイルを読み込んで処理を完了すると、"smi_data"、"ftse_data"、"stocks_original"の3つのインデックスを確認することができます。

<h2>SMIインデックスを確認する</h2>
<pre class="prettyprint">GET /smi_data/_search
</pre>
このコマンドで、ドキュメントが次の構造で表示されます。"smi_data"インデックスで、"SMI"市場のデータのみが表示されることに注目しましょう。

<pre class="prettyprint"> {
 "_index": "smi_data",
 "_type": "doc",
 "_id": "_QRskWUBsYalOV9y9hGJ",
 "_score":1,
 "_source": {
 "SMI":1688.5 
 }
 }
</pre>
<h2>FTSEインデックスを確認する</h2>
<pre class="prettyprint">GET /ftse_data/_search
</pre>
このコマンドで、ドキュメントが次の構造で表示されます。"ftse_data"インデックスのドキュメントで、"FTSE"フィールドのみが表示されることに注目しましょう。

<pre class="prettyprint"> {
 "_index": "ftse_data",
 "_type": "doc",
 "_id":"AgRskWUBsYalOV9y9hL0",
 "_score":1,
 "_source": {
 "FTSE":2448.2
 }
 }
</pre>
<h2>オリジナルドキュメントインデックスを確認する</h2>
<pre class="prettyprint">GET /stocks_originals/_search
</pre>
このコマンドで、ドキュメントが次の構造で表示されます。"stocks_original"インデックスで、元のドキュメントがフィルタリングされていないことがわかります。

<pre class="prettyprint"> {
 "_index": "stocks_original",
 "_type": "doc",
 "_id": "-QRskWUBsYalOV9y9hFo",
 "_score":1,
 "_source": {
 "host":"Alexanders-MBP",
 "@timestamp":"2017-01-01T00:30:00.000Z",
 "SMI":1678.1,
 "@version":"1",
 "message":"1483230600,1628.75,1678.1,1772.8,2443.6",
 "CAC":1772.8,
 "DAX":1628.75,
 "time":"1483230600",
 "path": "/Users/arm/Documents/ES6.3/datasets/stocks_for_clones.csv",
 "FTSE":2443.6
 }
 }
</pre>
<h2>まとめ</h2>
本ブログ記事では、<a href="/jp/logstash">Logstash</a>の機能の一部をご紹介しました。具体的には、複数の株式市場のデータをLogstashに投入し、処理を行った後にデータを個別のアウトプットに送る使い方を説明しています。Logstashや<a href="/jp/products/">Elastic Stack</a>の試用に関して質問がおありの場合は、<a href="http://discuss.elastic.co">公開ディスカッションフォーラム</a>を活用されることをおすすめしています。

log-split-route-banner.jpg

log-split-route-thumb.jpg

Using Logstash to Split Data and Send it to Multiple Outputs

Logstashを使ってデータを区分し、複数のアウトプットに送る

データをElasticsearchに送信している多くのシステムでは、 新たに挿入されるドキュメントに対して<a href="/guide/en/elasticsearch/reference/current/docs-index_.html#_automatic_id_generation">Elasticsearchが自動で生成するID値</a>を活用しています。しかし、データソースが誤って同じドキュメントを複数回Elasticsearchに送信し、それぞれのドキュメントにそのような自動生成の_id値が使用された場合、この同じドキュメントが複数回、異なる_id値でElasticsearchに保存されることになります。これが発生した場合は、そのような重複を見つけ、削除することが必要になる場合があります。そこでこのブログでは、（1）Logstashを使用して、または（2）Pythonで記述されたカスタムコードを使用して、Elasticsearchの重複ドキュメントを検出し、削除する方法について説明します。
<h2>ドキュメント構造の例</h2>
このブログ用として、Elasticsearchクラスター内のドキュメントが次のような構造を持っているとします。これは、株式市場の取引を表すドキュメントを含むデータセットに対応しています。<pre class="prettyprint"> { "_index": "stocks", "_type": "doc", "_id":"6fo3tmMB_ieLOlkwYclP", "_version":1, "found": true, "_source": { "CAC":1854.6, "host":"Alexanders-MBP", "SMI":2061.7, "@timestamp":"2017-01-09T02:30:00.000Z", "FTSE":2827.5, "DAX":1527.06, "time":"1483929000", "message":"1483929000,1527.06,2061.7,1854.6,2827.5\r", "@version":"1" } } </pre>
このブログ用として、このドキュメント構造の例を使用します。そして、複数のドキュメントで["CAC", "FTSE", "SMI"]フィールドの値が同じ（お互いに重複している）場合を想定します。
<h2>Logstashを使用してElasticsearchドキュメントの重複を排除</h2>
Logstashを使用して、重複しているドキュメントを検出し、Elasticsearchインデックスから削除することができます。そのテクニックについては、<a href="/jp/blog/logstash-lessons-handling-duplicates">Logstashで重複に対処する方法に関するこちらのブログ</a>で説明しているため、このセクションではそのアプローチに該当する具体例を説明することにします。
下記の例は、私が記述したシンプルなLogstash構成です。これは、Elasticsearchクラスターのインデックスからドキュメントを読み取り、そして<a href="/guide/en/logstash/current/plugins-filters-fingerprint.html">フィンガープリントフィルター</a>を使用して、["CAC", "FTSE", "SMI"]フィールドのハッシュに基づいて各ドキュメントの一意の_id値を計算し、最後に、各ドキュメントを同じElasticsearchクラスターの新しいインデックスに書き込みます。これにより、重複したドキュメントが同一の_id値となり、重複が排除されます。
さらに、同じLogstashフィルターにいくつかの簡単な修正を加えることで、新たに作成されるインデックスに書き込まれる今後のドキュメントに適用することも可能です。これにより、ほぼリアルタイムで重複を回避することができます。これは、既存のインデックスからドキュメントを取得するのではなく、リアルタイムの入力ソースからドキュメントを受け入れるように入力セクションを変更（下記の例を参照）することで達成できます。
注意が必要なのは、カスタムの_id値（つまりElasticsearchで生成されていない_id）を使用する場合は、インデックス操作の<a href="/guide/en/elasticsearch/reference/current/tune-for-indexing-speed.html#_use_auto_generated_ids">書き込みパフォーマンスに影響が及ぶ</a>ことです。
また、使用するハッシュアルゴリズムによっては、このアプローチでは理論的に、_id値に対してゼロ以外の数の<a href="https://en.wikipedia.org/wiki/Collision_(computer_science)">ハッシュ衝突</a>が発生する可能性があることに注意が必要です。これによって理論的に、異なる2つのドキュメントが同一の_idがマッピングされ、どちらかのドキュメントが失われる可能性があります。ただし、実際のケースのほとんどにおいては、ハッシュ衝突の可能性は非常に低いものです。さまざまなハッシュ関数の詳細な分析についてはこのブログでは扱いませんが、フィンガープリントフィルターで使用されるハッシュ関数は、入力パフォーマンスとハッシュ衝突数に影響を与えるため、慎重に考慮するべきです。
フィンガープリントフィルターを使用して既存のインデックスの重複を排除する簡単なLogstash構成は、以下のようになります。<pre class="prettyprint">input { # Elasticsearchからすべてのドキュメントを読み込みます elasticsearch { hosts =&gt; "localhost" index =&gt; "stocks" query =&gt; '{ "sort": [ "_doc" ] }' } } # このフィルターは2019年2月18日に更新されています filter { fingerprint { key =&gt; "1234ABCD" method =&gt; "SHA256" source =&gt; ["CAC", "FTSE", "SMI"] target =&gt; "[@metadata][generated_id]" &nbsp;&nbsp;&nbsp;&nbsp;concatenate_sources =&gt; true # &lt;-- 最初の記述日以降に追加された新しい行 } } output { stdout { codec =&gt; dots } elasticsearch { index =&gt; "stocks_after_fingerprint" document_id =&gt; "%{[@metadata][generated_id]}" } } </pre>
<h2>カスタムのPythonスクリプトを使用してElasticsearchドキュメントの重複を排除</h2>
<h3>メモリー効率の高いアプローチ</h3>
Logstashを使用しない場合は、カスタムのPythonスクリプトで効率的に重複の排除を達成できます。このアプローチでは、定義済みの["CAC", "FTSE", "SMI"]フィールドのハッシュを計算して、ドキュメントを一意に特定します。そして、そのハッシュをPythonの辞書のキーとして使用します。辞書の各エントリーに関連付けられた値は、同じハッシュにマッピングされるドキュメントの_idsの配列となります。
2つ以上のドキュメントが同一のハッシュを持っている場合、同じハッシュにマッピングされた重複ドキュメントは削除可能です。または、<a href="https://en.wikipedia.org/wiki/Collision_(computer_science)">ハッシュの衝突</a>の可能性に懸念がある場合は、同じハッシュにマッピングされているドキュメントのコンテンツを精査して、それらのドキュメントが本当に同一かどうかを確認し、同一の場合は重複を排除することができます。
<h3>検出アルゴリズム分析</h3>
50 GBのインデックスの場合、そのインデックスに平均0.4 kBのサイズのドキュメントが含まれていると仮定すると、そのインデックスには1億2,500万件のドキュメントが含まれていることになります。128ビットのmd5ハッシュを使用した場合、重複を排除したデータ構造を保存するのに必要なメモリー量はおよそ128ビットx 1億2,500万 = 2 GBのメモリーとなり、さらに160ビットの_idsには別途、160ビットx1億2,500万 = 2.5 GBのメモリーが必要になります。したがってこのアルゴリズムでは、関連するすべてのデータ構造をメモリーに保存するために、およそ4.5 GBのRAMが必要になります。このメモリーフットプリントは、次のセクションで説明するアプローチを適用すれば大幅に削減できます。
<h3>アルゴリズムの強化</h3>
このセクションでは、メモリー使用量を減らし、新たな重複ドキュメントを継続的に削除するためのアルゴリズムの強化について説明します。
時系列データを保存していて、重複するドキュメントが互いにわずかな時間内でのみ発生することが分かっている場合、インデックス内のドキュメントのサブセット（異なる期間に対応しているサブセット）に対してアルゴリズムを順次実行していくことで、このアルゴリズムのメモリーフットプリントを改善できる場合があります。たとえば、1年間のデータを保有している場合、日時フィールドに対して<a href="/guide/en/elasticsearch/reference/current/query-dsl-range-query.html">範囲クエリー</a>を使用（最良のパフォーマンスを得るためには<a href="/guide/en/elasticsearch/reference/current/query-filter-context.html">フィルターコンテキスト</a>内で使用）して、 一度に1週間ずつのデータセットで実行します。各週に対して1度実行すると、このアルゴリズムの実行は52回必要になります。この場合、このアプローチによってメモリーフットプリントは最低でも52分の1に削減されます。
上記の例では、1週間を超える間隔で発生した重複ドキュメントを検出できないと心配する方もいるでしょう。そこで、2時間以上の間隔でドキュメントの重複が発生しないことが分かっている場合を考えてみます。この場合、アルゴリズムの各実行は、前回の実行によって分析された最後のドキュメントセットから2時間分のドキュメントを含むようにして実行する必要があります。したがって週ごとの場合は、重複ドキュメントを確実に確認するために、170時間（1週間プラス2時間）分の時系列ドキュメントに対してクエリーを実行する必要があります。
インデックスから定期的かつ継続的に重複ドキュメントを削除する場合は、最近受領したドキュメントに対してこのアルゴリズムを実行します。上記と同じ論理が当てはまります。つまり、最近受領したドキュメントと、上記のように対象期間を十分に取ってその期間内の古いドキュメントを含めて分析し、重複をうっかり見逃さないようにすることが重要です。
<h3>重複ドキュメントを検出するためのPythonコード</h3>
以下のコードは、ドキュメントが同一かどうかを効率的に評価し、必要な場合にはその重複を排除する方法です。ただし、ドキュメントを誤って削除してしまうことを防ぐために、この例では実際に削除操作は実行しないようにしてあります。削除操作は簡単に含めることができます。
Elasticsearchからドキュメントの重複を排除するコードについては、<a href="https://github.com/alexander-marquardt/deduplicate-elasticsearch">GitHub</a>でも見つけることができます。<pre class="prettyprint">#!/usr/local/bin/python3 import hashlib from elasticsearch import Elasticsearch es = Elasticsearch(["localhost:9200"]) dict_of_duplicate_docs = {} # 次の行は、ドキュメントが重複しているかどうかを 判断するために使用するフィールドを定義します keys_to_include_in_hash = ["CAC", "FTSE", "SMI"] # 現在の検索/スクロールで返されたドキュメントを処理します def populate_dict_of_duplicate_docs(hits): for item in hits: combined_key = "" for mykey in keys_to_include_in_hash: combined_key += str(item['_source'][mykey]) _id = item["_id"] hashval = hashlib.md5(combined_key.encode('utf-8')).digest() # hashval（ハッシュ値）が新しい場合は、 # dict_of_duplicate_docsに新しいキーを作成します # これには空の配列の値が割り当てられます # その直後に_idを配列にプッシュします # hashvalがすでに存在している場合は、 # 新しい_idを既存の配列にプッシュします dict_of_duplicate_docs.setdefault(hashval, []).append(_id) # インデックスのすべてのドキュメントに実行し、 # dict_of_duplicate_docsデータ構造を投入します def scroll_over_all_docs(): data = es.search(index="stocks", scroll='1m', body={"query": {"match_all": {}}}) # スクロールIDを取得します sid = data['_scroll_id'] scroll_size = len(data['hits']['hits']) # スクロールする前に、現在のヒットのバッチを処理します populate_dict_of_duplicate_docs(data['hits']['hits']) while scroll_size &gt; 0: data = es.scroll(scroll_id=sid, scroll='2m') # 現在のヒットのバッチを処理します populate_dict_of_duplicate_docs(data['hits']['hits']) # スクロールIDを更新します sid = data['_scroll_id'] # 最後のスクロールで返された結果数を取得します scroll_size = len(data['hits']['hits']) def loop_over_hashes_and_remove_duplicates(): # ドキュメント値のハッシュを検索し、 # 重複したハッシュが見つかったどうかを確認します for hashval, array_of_ids in dict_of_duplicate_docs.items(): if len(array_of_ids) &gt; 1: print("********** Duplicate docs hash=%s **********" % hashval) # 現在のhashvalにマッピングされたドキュメントを取得します matching_docs = es.mget(index="stocks", doc_type="doc", body={"ids": array_of_ids}) for doc in matching_docs['docs']: # この例では重複ドキュメントの出力のみを行っています # このコードは、重複を出力するのではなく、 削除するように簡単に修正できます print("doc=%s\n" % doc) def main(): scroll_over_all_docs() loop_over_hashes_and_remove_duplicates() main() </pre>
<h2>まとめ</h2>
このブログでは、Elasticsearchでのドキュメントの重複を排除する方法を2つ説明しました。最初の方法ではLogstashを使用して重複ドキュメントを削除し、2つ目の方法ではカスタムのPythonスクリプトを使用して重複ドキュメントを見つけ、削除します。
Elasticsearchでのドキュメントの重複排除について、またはElasticsearch関連のトピックについてご質問がある場合は、<a href="https://discuss.elastic.co/">ディスカッションフォーラム</a>にアクセスし、重要なインサイトや情報を入手してください。

memory-elasticsearch-compute-storage-computer-thumbnail.jpg

memory-computer-storage-elasticsearch-compression-full-bleed.jpg

How to Find and Remove Duplicate Documents in Elasticsearch

Elasticsearchでの重複ドキュメントの発見と削除方法

Start free trial

Blog Footer CTA

Sign up for Elastic Cloud free trial

Consulting Engineer

Alex Marquardt

By submitting you acknowledge that you've read and agree to our <a href="/legal/elastic-cloud-account-terms" target="_self">Terms of Service</a>, and you agree to receive the <a href="/legal/privacy-statement#how-we-use-the-information" target="_self">selected communications</a> at the contact details you provided above. See <a href="/legal/privacy-statement/" target="_self">Elastic’s Privacy Statement</a> for more details or to opt-out at any time.

Communication Preferences - ONLY for /communication-preferences

送信することにより、Elasticの<a href="/jp/legal/elastic-cloud-account-terms" target="_self">利用規約</a>を読み、同意したものと見なされます。また、上記で提供された連絡先に<a href="/jp/legal/privacy-statement#how-we-use-the-information" target="_self">特定の通知</a>が送信されることを認めたことになります。詳細については、<a href="/jp/legal/privacy-statement/" target="_self">Elasticのプライバシーポリシー</a>をご覧ください。これらの同意についてはいつでもオプトアウトすることが可能です。

By submitting you acknowledge that you've read and agree to our <a href="/legal/elastic-cloud-account-terms" target="_blank">Terms of Service</a>, and that Elastic may <a href="/legal/privacy-statement#how-we-use-the-information" target="_blank">contact you</a> about our related products and services, using the details you provide above. See <a href="/legal/privacy-statement/" target="_blank">Elastic’s Privacy Statement</a> for more details or to opt-out at any time.By submitting this form, I agree to Elastic sharing my contact information with Google Cloud. I also agree to Elastic giving my information to the vendor for the purpose of shipping my food package.

GDPR Text + Google Cloud

送信することにより、Elasticの<a href="/jp/legal/elastic-cloud-account-terms" target="_self">利用規約</a>を読み、同意したものと見なされます。また、関連する製品やサービスについて、上記で提供された情報を使用してElasticから<a href="/jp/legal/privacy-statement#how-we-use-the-information" target="_self">連絡する</a>場合があることを認めたことになります。詳細については、<a href="/jp/legal/privacy-statement/" target="_self">Elasticのプライバシー</a>ポリシーをご覧ください。これらの同意についてはいつでもオプトアウトすることが可能です。このフォームを送信することにより、Elasticが私の連絡先情報をGoogle Cloudに共有することに同意します。また、Elasticが私に食材パッケージを発送する目的で私の情報をベンダーに提供することにも同意します。

By submitting you acknowledge that you've read and agree to our <a href="/legal/serverless-preview-terms" target="_blank">Terms of Service</a>, and that Elastic may <a href="/legal/privacy-statement#how-we-use-the-information" target="_blank">contact you</a> about our related products and services, using the details you provide above. See <a href="/legal/privacy-statement/" target="_blank">Elastic’s Privacy Statement</a> for more details or to opt-out at any time.

Serverless GDPR Text

送信することにより、Elasticの<a href="/jp/legal/serverless-preview-terms" target="_self">利用規約</a>を読み、同意したものと見なされます。また、関連する製品やサービスについて、上記で提供された情報を使用してElasticから<a href="/jp/legal/privacy-statement#how-we-use-the-information" target="_self">連絡する</a>場合があることを認めたことになります。詳細については、<a href="/jp/legal/privacy-statement/" target="_self">Elasticのプライバシー</a>ポリシーをご覧ください。これらの同意についてはいつでもオプトアウトすることが可能です。

By submitting you agree to <a href="/agreements/global/cloud-services-monthly" rel="nofollow">Elastic Cloud Terms of Service</a>. Your personal data will be processed in accordance with <a href="/legal/privacy-statement">Elastic's Privacy Statement</a>.

Cloud Trial GDPR Text

送信することにより、<a href="/jp/agreements/global/cloud-services-monthly" rel="nofollow">Elastic Cloud利用規約</a>に同意したものとみなされます。個人情報は<a href="/jp/legal/privacy-statement">Elasticのプライバシーポリシー</a>に基づいて取り扱われます。

By submitting you acknowledge that you've read and agree to our <a href="/legal/terms-of-use" target="_blank">Terms of Service</a>, and that Elastic may <a href="/legal/privacy-statement#how-we-use-the-information" target="_blank">contact you</a> about our related products and services, using the details you provide above. See <a href="/legal/privacy-statement/" target="_blank">Elastic’s Privacy Statement</a> for more details or to opt-out at any time.

Newsletter GDPR Text

送信することにより、Elasticの<a href="/jp/legal/terms-of-use" target="_self">利用規約</a>を読み、同意したものと見なされます。また、関連する製品やサービスについて、上記で提供された情報を使用してElasticから<a href="/jp/legal/privacy-statement#how-we-use-the-information" target="_self">連絡する</a>場合があることを認めたことになります。詳細については、<a href="/jp/legal/privacy-statement/" target="_self">Elasticのプライバシー</a>ポリシーをご覧ください。これらの同意についてはいつでもオプトアウトすることが可能です。

Explore similar demos

Overview

Speakers

Close

See more insights

The content on this page is not available in the selected language. As Elastic grows globally, we continue to support content in multiple languages.

The content on this page is not available in the selected language.

このページのコンテンツは、選択された言語ではご利用いただけません。Elasticでは現在、コンテンツをさまざまな言語でご利用いただけるよう取り組んでいます。お使いの言語が準備されるまでお待ちくださるようお願いいたします。

Author

Articles by

Watch now (no PT)

Watch now

See all top stories

すべてのトップストーリーを見る

All (no PT translation)

Contact information

Press Release

Share on Reddit

Share this story

Share by Email

Thank you for your interest!

Contact Info

Follow us on Youtube

Load More

Share on LinkedIn

Search Integrations

All Solutions

Video for

See when this webinar starts in my time zone

現在のタイムゾーンでこのウェビナー開始時間を見る

Register to Watch

See All Posts

Can't make it? Register and we'll send you the recording. You'll also receive an email with related content

他の予定と重なってしまった…という場合も、問題ありません。参加登録されると、録画配信をご利用いただけます。関連コンテンツのご案内もメールでお届けします。