Code reduzieren, wiederverwenden und recyclen. Wir zeigen Ihnen, wie Sie modulare Komponenten für Ihre Logstash-Ingest-Pipelines erstellen können.

<a href="/de/products/logstash">Logstash</a> ist eine Open-Source-Pipeline zur Datenverarbeitung, die Ereignisdaten aus einer oder mehreren Quellen (Inputs) ingestiert und transformiert und das Ergebnis dann an einen oder mehrere Zielorte (Outputs) sendet. Es gibt Logstash-Implementierungen mit einer großen Menge von Codezeilen, die Ereignisdaten aus mehreren Quellen verarbeiten. Ich werde Ihnen zeigen, wie Sie mithilfe von Pipelines für modulare Komponenten die Wiederverwendbarkeit von Code verbessern können, damit sich solche Implementierungen leichter pflegen lassen.
 <h2>Motivation</h2>Bei Logstash kommt es häufig vor, dass auf Ereignisdaten aus mehreren Quellen eine bestimmte Menge gemeinsamer Logik angewendet werden muss. Dies lässt sich in der Regel auf eine der beiden folgenden Arten erledigen:&nbsp;
<ol>
	<li>Es gibt eine gemeinsame Pipeline, in der die Ereignisdaten aus allen Quellen zusammengefasst werden. Das macht es einfach, die gemeinsame Logik auf alle Ereignisdaten aus allen Quellen anzuwenden. Derartige Implementierungen weisen zusätzlich zur gemeinsamen Logik in der Regel eine beträchtliche Menge konditionaler Logik auf. Diese Herangehensweise kann daher dazu führen, dass die resultierenden Logstash-Implementierungen kompliziert und schwer durchschaubar sind.</li>
	<li>Es gibt für jede Quelle eine eigene Pipeline zur Verarbeitung der Ereignisdaten aus genau dieser Quelle. Bei dieser Methode müssen die für alle Daten benötigten Funktionen dupliziert und in jede einzelne Pipeline kopiert werden, was den Aufwand für die Pflege der gemeinsamen Teile des Codes erhöht.&nbsp;</li>
</ol>Das hier beschriebene Verfahren hat zum Ziel, den Mängeln der beiden oben aufgeführten Verfahren entgegenzuwirken, indem wir modulare Komponenten in unterschiedlichen Dateien speichern und dann diese Komponenten zu Pipelines kombinieren. So lässt sich die Komplexität der Pipelines reduzieren und es muss kein Code mehr dupliziert werden.
 <h2>Aufbau einer modularen Pipeline</h2>Eine Logstash-<a href="/guide/en/logstash/current/configuration.html">Konfigurationsdatei</a> besteht aus Inputs, Filtern und Outputs, die von einer Logstash-Pipeline ausgeführt werden.&nbsp; In komplexeren Setups <a href="/guide/en/logstash/current/multiple-pipelines.html">übernimmt eine Logstash-Instanz die Ausführung mehrerer Pipelines</a>. Wenn man Logstash ohne Argumente startet, liest diese Instanz standardmäßig eine Datei namens <code>pipelines.yml</code> und instanziiert die angegebenen Pipelines.&nbsp;
Logstash-Inputs, ‑Filter und ‑Outputs können in mehreren Dateien gespeichert und über einen <a href="/guide/en/logstash/current/glob-support.html">Glob-Ausdruck</a> für die Aufnahme in eine Pipeline ausgewählt werden. Die Dateien, die mit dem Glob-Ausdruck übereinstimmen, werden dann in alphabetischer Reihenfolge miteinander kombiniert. Da es bei der Ausführung von Filtern häufig auf die Reihenfolge ankommt, kann es sinnvoll sein, in den Dateinamen Nummerierungen zu verwenden, damit die Dateien in der gewünschten Reihenfolge kombiniert werden.
Im Folgenden definieren wir zwei getrennte Pipelines, in denen mehrere modulare Logstash-Komponenten miteinander kombiniert sind. Wir speichern unsere Logstash-Komponenten in den folgenden Dateien:
<ul>
	<li>Input-Deklarationen: <code>01_in.cfg</code>, <code>02_in.cfg</code></li>
	<li>Filter-Deklarationen:<code> 01_filter.cfg</code>, <code>02_filter.cfg</code>, <code>03_filter.cfg</code></li>
	<li>Output-Deklarationen:<code> 01_out.cfg</code></li>
</ul>Anschließend nutzen wir <a href="/guide/en/logstash/current/glob-support.html">Glob-Ausdrücke</a>, um in <code>pipelines.yml</code> Pipelines zu definieren, die aus den gewünschten Komponenten bestehen:&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>In der oben beschriebenen Pipeline-Konfiguration gibt es in beiden Pipelines die Datei <code>02_filter.cfg</code>. Sie zeigt, wie der in beiden Pipelines vorhandene Code in einer gemeinsamen Datei definiert und verwaltet und von mehreren Pipelines ausgeführt werden kann.
 <h2>Testen der Pipelines</h2>In diesem Abschnitt stellen wir ein konkretes Beispiel für die Dateien in den Pipelines vor, die in der oben erwähnten Datei <code>pipelines.yml</code> definiert sind. Anschließend führen wir Logstash mit diesen Dateien aus und zeigen, was Logstash für einen Output generiert.&nbsp;
 <h3>Konfigurationsdateien</h3>Input-Datei: 01_in.cfg
Diese Datei definiert einen Input, der als <a href="/guide/en/logstash/current/plugins-inputs-generator.html">Generator</a> fungiert. Der Generator-Input dient zum Testen von Logstash und generiert in unserem Fall ein einziges Ereignis.&nbsp;
<pre class="prettyprint">input { 
 generator { 
 lines =&gt; ["Generated line"] 
 count =&gt; 1 
 } 
}
</pre>Input-Datei: 02_in.cfg
Diese Datei definiert einen Logstash-Input, der den Datenstrom „stdin“ abhört.
<pre class="prettyprint">input { 
 stdin {} 
}
</pre>Filter-Datei: 01_filter.cfg
<pre class="prettyprint">filter { 
 mutate { 
 add_field =&gt; { "filter_name" =&gt; "Filter 01" } 
 } 
}
</pre>Filter-Datei: 02_filter.cfg
<pre class="prettyprint">filter { 
 mutate { 
 add_field =&gt; { "filter_name" =&gt; "Filter 02" } 
 } 
}
</pre>Filter-Datei: 03_filter.cfg
<pre class="prettyprint">filter { 
 mutate { 
 add_field =&gt; { "filter_name" =&gt; "Filter 03" } 
 } 
}
</pre>Output-Datei: 01_out.cfg
<pre class="prettyprint">output { 
 stdout { codec =&gt; "rubydebug" } 
}
</pre> <h3>Ausführen der Pipeline</h3>Wenn Logstash ohne Optionen gestartet wird, wird die zuvor von uns definierte Datei <code>pipelines.yml</code> ausgeführt. Zum Starten von Logstash geben wir Folgendes ein:&nbsp;
<pre class="prettyprint">./bin/logstash
</pre>Die Pipeline <code>my-pipeline_1</code> führt zum Simulieren eines Input-Ereignisses einen Generator aus. Daher wird nach der Initialisierung von Logstash Folgendes ausgegeben, was zeigt, dass die Inhalte von <code>01_filter.cfg</code> und <code>02_filter.cfg</code> so ausgeführt werden, wie dies von dieser Pipeline erwartet wird:
<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>Da die andere Pipeline namens <code>my-pipeline_2</code> auf stdin auf Input wartet, haben wir bisher noch keine Ereignisse gesehen, die von dieser Pipeline verarbeitet wurden. Wir können jetzt etwas in das Terminal für Logstash eingeben und durch Drücken der Eingabetaste ein Ereignis für diese Pipeline erstellen. Daraufhin dürfte in etwa Folgendes angezeigt werden:&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>Das zeigt uns, dass die Logik aus <code>02_filter.cfg</code> und <code>03_filter.cfg</code> erwartungsgemäß angewendet wird.&nbsp;&nbsp;
 <h3>Reihenfolge der Ausführung</h3>Zu beachten ist, dass Logstash die Reihenfolge der Dateien im Glob-Ausdruck komplett egal ist. Der Glob-Ausdruck wird nur verwendet, um zu bestimmen, welche Dateien einbezogen werden sollen. Die Dateien werden dann alphabetisch geordnet. Das bedeutet, dass selbst dann, wenn wir die Definition von <code>my-pipeline_2</code> so ändern würden, dass <code>03_filter.cfg</code> im Glob-Ausdruck vor <code>02_filter.cfg</code> erscheint, jedes einzelne Ereignis vor dem Durchlaufen des in <code>03_filter.cfg</code> definierten Filters den Filter in <code>02_filter.cfg</code> passieren muss.
 <h2>Fazit</h2>Mit Glob-Ausdrücken lassen sich Logstash-Pipelines aus modularen Komponenten zusammensetzen, die als einzelne Dateien gespeichert werden. Das kann die Pflege, Wiederverwendbarkeit und Lesbarkeit von Code erleichtern.&nbsp;
Nebenbei bemerkt: Bei den Überlegungen dazu, wie sich die Modularität von Logstash-Implementierungen verbessern lässt, lohnt es sich, zusätzlich zum hier beschriebenen Verfahren auch einen Blick auf die <a href="/guide/en/logstash/current/pipeline-to-pipeline.html">Pipeline-zu-Pipeline-Kommunikation</a> zu werfen.

blog-pipe-sky-banner.jpg

blog-pipe-sky-thumb.jpg

How to create maintainable and reusable Logstash pipelines

Erstellen gut pflegbarer und wiederverwendbarer Logstash-Pipelines

Lesen Sie, wie Sie dafür sorgen können, dass die Daten in Ihrem Elasticsearch-Cluster mit Ihren relationalen Datenbanken synchronisiert bleiben.

es-database-sync-banner.jpg

es-database-sync-thumb.jpg

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

RDBMS- und Elasticsearch-Daten mit Logstash und JDBC synchron halten

<a href="/de/logstash">Logstash</a> ist eine Open-Source-Software zur serverseitigen Verarbeitung von Datenströmen, die Daten ingestiert und transformiert, bevor sie sie an einen oder mehrere Zielorte (Outputs) sendet. In diesem Blog zeige ich Ihnen anhand eines Beispiels, wie man mit Logstash Daten für mehrere Aktienindizes ingestieren und sie dann je nach Aktienindex jeweils an einen anderen Zielort senden lassen kann. Dafür sind die folgenden Schritte erforderlich:

<ol>
	<li>Von jedem Dokument im eingehenden Aktienindex-Datenstream werden Kopien erstellt.</li>
	<li>Die Kopien werden so gefiltert, dass sie nur die für den jeweiligen Aktienindex gültigen Felder enthalten.</li>
	<li>Den Kopien werden Metadaten hinzugefügt, aus denen hervorgeht, zu welchem Aktienindex die Daten gehören.</li>
	<li>Es wird eine Auswertung der Metadaten in jedem Dokument eingerichtet, damit das Dokument an den richtigen Zielort gesendet wird.</li>
</ol>
Beachten Sie, dass ich in diesem Blogpost nicht die <a href="/guide/en/logstash/6.5/pipeline-to-pipeline.html">Pipeline-zu-Pipeline-Kommunikation (in 6.5 in der Betaphase)</a> nutze, mit der sich einige der hier beschriebenen Funktionen ebenfalls umsetzen lassen dürften.

<h2>Beispiel für eine eingehende Datei</h2>
Als eingehende Datei verwenden wir eine CSV-Datei, die Aktienindexwerte enthält. Die Werte in dieser CSV-Datei sehen wie folgt aus (als Dezimaltrennzeichen wird hier der Punkt verwendet):

<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>
Die Werte stehen jeweils für einen Zeitpunkt und den Wert der folgenden Aktienindizes: DAX, SMI, CAC und FTSE. Wir kopieren die Zeilen oben in eine CSV-Datei namens „stocks.csv“, die wir dann an die Logstash-Pipeline in unserem Beispiel senden können.

<h2>Beispiel für eine Logstash-Pipeline</h2>
Die unten dargestellte Logstash-Pipeline führt Folgendes aus:

<ol>
	<li>Sie liest Aktienindexwerte aus einer CSV-Datei (Datei mit kommagetrennten Werten).</li>
	<li>Sie ordnet jede Zeile in der CSV-Datei einem JSON-Dokument zu, wobei die Spalten in der CSV-Datei für die folgenden JSON-Felder stehen: „time“ (Zeitpunkt), „DAX“, „SMI“, „CAC“ und „FTSE“.</li>
	<li>Sie wandelt das Feld „time“ ins Unix-Format um.</li>
	<li>Sie erstellt mithilfe des <a href="/guide/en/logstash/6.5/plugins-filters-clone.html">Filter-Plugins „clone“</a> von jedem Dokument zwei Kopien (diese Kopien werden zusätzlich zum Originaldokument angelegt). Der Filter „clone“ fügt jeder neuen Dokumentkopie automatisch ein neues Feld „type“ hinzu, wobei „type“ mit den Namen korrespondiert, die im Klone-Array festgelegt wurden. Als Typennamen haben wir „clone_for_SMI“ oder „clone_for_FTSE“ definiert und jeder Klon enthält am Ende nur Daten entweder für den SMI oder aber für den FTSE. </li>
	<li>Für jeden Klon gilt Folgendes:</li>
	<ol>
		<li>Mit dem <a href="/guide/en/logstash/6.5/plugins-filters-prune.html">Filter-Plugin „prune“</a> werden die Felder entfernt, die nicht für den konkreten Aktienindex gültig sind.</li>
		<li>Jedes Dokument wird entsprechend dem Wert des von der Funktion „clone“ hinzugefügten Feldes „type“ mit <a href="/de/blog/logstash-metadata">Metadaten</a> versehen. Dies ist notwendig, da wir die Funktion „prune“ verwenden werden, mit der die von „clone“ hinzugefügte Angabe zum Typ entfernt wird. Wir benötigen aber den Typ, um beim Senden der Daten an den Zielort entscheiden zu können, wohin das Dokument gehen soll.</li>
	</ol>
	<li>Mit dem <a href="/guide/en/logstash/6.5/plugins-outputs-elasticsearch.html">Elasticsearch-Output-Plugin für Logstash</a> werden die Dokumente gemäß ihrer Aktienindexzugehörigkeit jeweils an einen anderen Elasticsearch-Zielort gesendet. Welcher das ist, richtet sich nach dem Wert im Metadatenfeld, den wir in Schritt&nbsp;5 hinzugefügt haben. Zur Vereinfachung des Codes unten wird jede Elasticsearch-Ausgabe in einen eigenen Index in einem lokalen Elasticsearch-Cluster geschrieben. Wenn mehrere Cluster als Zielorte verwendet werden sollen, lassen sich in den Elasticsearch-Output-Deklarationen ganz einfach eindeutige <a href="/guide/en/logstash/6.5/plugins-outputs-elasticsearch.html#plugins-outputs-elasticsearch-hosts">Elasticsearch-Hosts</a> angeben.</li>
</ol>
Das Beispiel unten zeigt eine Logstash-Pipeline, die die oben genannten Schritte ausführt (mit als Kommentar hinzugefügten entsprechenden Schrittnummern).&nbsp;Diese Pipeline kopieren wir zur Ausführung in eine Datei namens „clones.conf“:

<pre class="prettyprint">## SCHRITT 1
input {
 file {
 # Im Pfad auf die Datei stocks.csv verweisen
 path =&gt; "${HOME}/stocks.csv"
 # Festlegen, dass bei jeder Logstash-Ausführung der 
 # gesamte Input gelesen wird (fürs Debugging hilfreich):
 start_position =&gt; "beginning"
 sincedb_path =&gt; "/dev/null"
 }
}
## SCHRITT 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'
 }
 }
## SCHRITT 3 
date {
 match =&gt; ['time', 'UNIX']
 }
## SCHRITT 4
 # Mit der folgenden Zeile werden von jedem 
 # Dokument 2 Kopien erstellt, sodass wir am Ende 
 # insgesamt 3 Exemplare (Original + 2 Kopien) haben. 
 # Jedem Exemplar wird automatisch ein Feld „type“ 
 # entsprechend dem Namen im Array hinzugefügt.
 clone {
 clones =&gt; ['clone_for_SMI', 'clone_for_FTSE']
 }
## SCHRITT 5
 if [type] == 'clone_for_SMI' {
 # Alles außer „SMI“ entfernen
 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" } 
 }
 } 
}
## SCHRITT 6
output {
 # Die folgende Ausgabe für „stdout“ dient nur zum Debugging 
 # und kann entfernt werden
 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>Testen der Logstash-Pipeline</h2>
Zum Testen dieser Pipeline mit den Beispiel-CSV-Daten kann der folgende Befehl verwendet werden, wobei sicherzustellen ist, dass die korrekten Pfade fürs jeweilige System verwendet werden. Die Festlegung von „config.reload.automatic“ ist optional, erlaubt es uns aber, „clones.conf“ neu zu laden, ohne Logstash neu starten zu müssen:

<pre class="prettyprint">./logstash -f ./clones.conf --config.reload.automatic
</pre>
Nachdem Logstash die Datei „stocks.csv“&nbsp;gelesen und die Verarbeitung abgeschlossen hat, können wir die drei Ergebnisindizes („smi_data“, „ftse_data“ und „stocks_original“) prüfen.

<h2>Prüfen des Index für den SMI</h2>
<pre class="prettyprint">GET /smi_data/_search
</pre>
Damit werden Dokumente mit der folgenden Struktur angezeigt. Wie Sie sehen, erscheinen im Index „smi_data“ nur Daten für den SMI.

<pre class="prettyprint"> {
 "_index": "smi_data",
 "_type": "doc",
 "_id": "_QRskWUBsYalOV9y9hGJ",
 "_score": 1,
 "_source": {
 "SMI": 1688.5 
 }
 }
</pre>
<h2>Prüfen des Index für den FTSE</h2>
<pre class="prettyprint">GET /ftse_data/_search
</pre>
Damit werden Dokumente mit der folgenden Struktur angezeigt. Wie Sie sehen, erscheint in Dokumenten im Index „ftse_data“ nur das Feld „FTSE“.

<pre class="prettyprint"> {
 "_index": "ftse_data",
 "_type": "doc",
 "_id": "AgRskWUBsYalOV9y9hL0",
 "_score": 1,
 "_source": {
 "FTSE": 2448.2
 }
 }
</pre>
<h2>Prüfen des Index der Originaldokumente</h2>
<pre class="prettyprint">GET /stocks_originals/_search
</pre>
Damit werden Dokumente mit der folgenden Struktur angezeigt. Wie Sie sehen, erscheint im Index „stocks_original“ die originale ungefilterte Version der Dokumente.

<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>Fazit</h2>
In diesem Blogpost habe ich einen kleinen Teil der Funktionen demonstriert, die <a href="/de/logstash">Logstash</a> zu bieten hat. Anhand eines Beispiels habe ich gezeigt, wie man mit Logstash Daten für mehrere Aktienindizes ingestieren, sie verarbeiten und die verarbeiteten Daten dann an verschiedene Zielorte senden kann. Wenn Sie gerade dabei sind, Logstash und den <a href="/de/products/">Elastic Stack</a> auszuprobieren, und Fragen haben, sind unsere <a href="http://discuss.elastic.co">öffentlichen Diskussionsforen</a> die beste Anlaufstelle.

log-split-route-banner.jpg

log-split-route-thumb.jpg

Using Logstash to Split Data and Send it to Multiple Outputs

Aufteilen von Daten und Senden der Daten an unterschiedliche Zielorte mit Logstash

Viele der Systeme, die Elasticsearch mit Daten versorgen, nutzen die <a href="/guide/en/elasticsearch/reference/current/docs-index_.html#_automatic_id_generation">automatisch generierten ID-Werte von Elasticsearch</a> für neu eingefügte Dokumente. Wenn die Datenquelle jedoch versehentlich dasselbe Dokument mehrmals an Elasticsearch sendet und jedes in Elasticsearch eingefügte Dokument eine solche automatisch generierte _id erhält, dann wird dasselbe Dokument in Elasticsearch mehrfach mit unterschiedlichen _id-Werten gespeichert. In diesem Fall kann es notwendig sein, solche Duplikate zu finden und zu entfernen. Daher befassen wir uns in diesem Blogeintrag damit, wie Sie doppelte Dokumente aus Elasticsearch (1) mit Logstash oder (2) mit benutzerdefiniertem Python-Code erkennen und eliminieren können.
<h2>Beispielstruktur für Dokumente</h2>
Für diesen Blogeintrag gehen wir davon aus, dass die Dokumente im Elasticsearch-Cluster die folgende Struktur haben. Diese Struktur entspricht einem Datensatz mit Dokumenten, die verschiedene Aktienkurse abbilden.<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>
Angesichts dieser Dokumentstruktur legen wir für diesen Blogeintrag fest, dass Dokumente Duplikate sind, wenn sie jeweils dieselben Werte in den Feldern ["CAC", "FTSE", "SMI"] haben.
<h2>Elasticsearch-Dokumente mit Logstash deduplizieren</h2>
Sie können Logstash verwenden, um doppelte Dokument in einem Elasticsearch-Index zu finden und zu entfernen. Diese Technik wird in <a href="/de/blog/logstash-lessons-handling-duplicates">unserem Blogeintrag zum Umgang mit Duplikaten in Logstash</a> beschrieben, und in diesem Abschnitt zeige ich Ihnen ein konkretes Beispiel.
Im folgenden Beispiel habe ich eine einfache Logstash-Konfiguration geschrieben, die Dokumente aus einem Index in einem Elasticsearch-Cluster liest, mit dem <a href="/guide/en/logstash/current/plugins-filters-fingerprint.html">Blueprint-Filter</a> einen eindeutigen _id-Wert für die einzelnen Dokumente auf Basis eines Hashs über die Felder ["CAC", "FTSE", "SMI"] berechnet und die Dokumente anschließend zurück in einen neuen Index im gleichen Elasticsearch-Cluster schreibt. Dabei werden doppelte Dokumente in dieselbe _id geschrieben und somit eliminiert.
Mit wenigen Änderungen können wir denselben Logstash-Filter auch auf zukünftige Dokumente anwenden, die in den neu erstellten Index geschrieben werden, um Duplikate praktisch in Echtzeit zu entfernen. Dazu können wir den Eingabebereich im folgenden Beispiel so ändern, dass Dokumente aus Ihrer Echtzeit-Eingabequelle akzeptiert werden, anstatt Dokumente aus einem vorhandenen Index abzurufen.
Beachten Sie dabei, dass benutzerdefinierte _id-Werte (_id-Werte, die nicht von Elasticsearch generiert werden) <a href="/guide/en/elasticsearch/reference/current/tune-for-indexing-speed.html#_use_auto_generated_ids">die Schreibleistung Ihrer Indexoperationen beeinträchtigen können</a>.
Je nach verwendetem Hashalgorithmus können bei diesem Ansatz theoretisch auch <a href="https://en.wikipedia.org/wiki/Collision_(computer_science)">Hash-Kollisionen</a> für den _id-Wert vorkommen, was dazu führen kann, dass zwei nicht identische Dokument zur gleichen _id zugeordnet werden. In diesem Fall würde eines dieser Dokumente verloren gehen. In der Praxis ist die Wahrscheinlichkeit von Hash-Kollisionen jedoch sehr gering. Eine ausführliche Analyse der verschiedenen Hashfunktionen übersteigt den Rahmen dieses Blogeintrags, aber die im Fingerprint-Filter verwendete Hashfunktion sollte sorgfältig ausgewählt werden, da sie sich auf die Ingestionsleistung und die Anzahl der Hash-Kollisionen auswirkt.
Hier sehen Sie eine einfache Logstash-Konfiguration zum Deduplizieren eines vorhandenen Index mit dem Fingerprint-Filter.<pre class="prettyprint">input { # Alle Dokumente aus Elasticsearch lesen elasticsearch { hosts =&gt; "localhost" index =&gt; "stocks" query =&gt; '{ "sort": [ "_doc" ] }' } } # Dieser Filter wurde am 18. Februar 2019 aktualisiert 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;-- Neue Zeile seit dem ursprünglichen Beitragsdatum hinzugefügt } } output { stdout { codec =&gt; dots } elasticsearch { index =&gt; "stocks_after_fingerprint" document_id =&gt; "%{[@metadata][generated_id]}" } } </pre>
<h2>Benutzerdefiniertes Python-Skript zum Deduplizieren von Elasticsearch-Dokumenten</h2>
<h3>Ein speicherschonender Ansatz</h3>
Falls Sie nicht mit Logstash arbeiten, können Sie Ihre Dokumente mit einem benutzerdefinierten Python-Skript deduplizieren. Für diesen Ansatz berechnen wir den Hash über die Felder ["CAC", "FTSE", "SMI"], die laut unserer Definition ein Dokument eindeutig identifizieren. Anschließend verwenden wir diesen Hash als Schlüssel in einem Python-Wörterbuch, in dem der zugeordnete Wert für jeden Wörterbucheintrag ein Array der _ids der Dokumente enthält, die denselben Hash haben.
Wenn mehr als ein Dokument denselben Hash hat, können die doppelten Dokumente, die denselben Hash haben, gelöscht werden. Falls Sie mögliche <a href="https://en.wikipedia.org/wiki/Collision_(computer_science)">Hash-Kollisionen</a> ausschließen möchten, können Sie den Inhalt der Dokumente mit gleichem Hash analysieren, um herauszufinden, ob die Dokumente wirklich identisch sind, und die Duplikate anschließend eliminieren.
<h3>Analyse des Erkennungsalgorithmus</h3>
Wenn ein Index mit 50 GB Dokumente mit einer durchschnittlichen Größe von 0,4 KB enthält, dann befinden sich etwa 125 Millionen Dokumente im Index. Wenn wir einen MD5-Hash mit 128 Bit verwenden, dann beträgt der Speicherverbrauch für die Deduplizierungsdatenstrukturen im Arbeitsspeicher also 128 Bits x 125 Millionen = 2 GB Arbeitsspeicher, plus 160 Bits x 125 Millionen = 2,5 GB Arbeitsspeicher für die _ids mit jeweils 160 Bits. Dieser Algorithmus benötigt also etwa 4,5 GB Arbeitsspeicher für alle relevanten Datenstrukturen. Dieser Speicherbedarf kann dramatisch reduziert werden, wenn der im nächsten Abschnitt beschriebene Ansatz eingesetzt werden kann.
<h3>Erweiterter Algorithmus</h3>
In diesem Abschnitt stelle ich Ihnen eine Erweiterung des Algorithmus vor, mit dem Sie den Speicherbedarf reduzieren und neue Duplikate fortlaufend entfernen können.
Wenn Sie Zeitreihendaten speichern und wissen, dass doppelte Dokumente nur in kurzen Abständen zueinander auftreten werden, können Sie den Speicherbedarf des Algorithmus verbessern, indem Sie den Algorithmus immer wieder über eine Teilmenge der Dokumente im Index ausführen, wobei jede Teilmenge einem anderen Zeitfenster entspricht. Wenn Sie beispielsweise Daten für ein Jahr haben, können Sie <a href="/guide/en/elasticsearch/reference/current/query-dsl-range-query.html">Bereichsabfragen</a> über Ihr datetime-Feld verwenden (in einem <a href="/guide/en/elasticsearch/reference/current/query-filter-context.html">Filterkontext</a> für optimale Leistung), um Ihre Daten Woche für Woche zu durchlaufen. In diesem Fall führen Sie den Algorithmus 52-mal aus (einmal pro Woche), und der Speicherbedarf für den schlechtesten Fall würde um den Faktor 52 reduziert.
Im obigen Beispiel haben Sie sich vielleicht gefragt, was mit Duplikaten passiert, die mehrere Wochen umspannen. Angenommen, Sie wissen, dass doppelte Dokumente nicht mehr als zwei Stunden voneinander getrennt auftreten können. In diesem Fall stellen Sie sicher, dass jede Ausführung des Algorithmus alle Dokumente enthält, die sich um jeweils zwei Stunden mit den Dokumenten aus dem letzten Durchlauf des Algorithmus überschneiden. Für das wöchentliche Beispiel müssten Sie also 170 Stunden (1 Woche + 2 Stunden) an Zeitreihendokumenten abfragen, um sicherzugehen, dass keine Duplikate übersehen werden.
Wenn Sie doppelte Dokumente in regelmäßigen Abständen aus Ihren Indizes entfernen möchten, können Sie diesen Algorithmus für die zuletzt empfangenen Dokumente ausführen. Dabei gilt dieselbe Logik wie oben: Die zuletzt empfangenen Dokumente müssen zusammen mit einem Teil der etwas älteren Dokumente analysiert werden, um sicherzustellen, dass keine Duplikate übersehen werden.
<h3>Python-Code zum Erkennen von Duplikaten</h3>
Der folgende Code demonstriert, wie Sie Dokumente effizient auf Gleichheit prüfen und bei Bedarf eliminieren können. In diesem Beispiel führen wir jedoch keinen tatsächlichen Löschvorgang aus, um zu verhindern, dass Dokumente versehentlich gelöscht werden. Diese Funktion kann jedoch mühelos hinzugefügt werden.
Sie finden den Code zum Deduplizieren von Dokumenten in Elasticsearch auch <a href="https://github.com/alexander-marquardt/deduplicate-elasticsearch">auf GitHub</a>.<pre class="prettyprint">#!/usr/local/bin/python3 import hashlib from elasticsearch import Elasticsearch es = Elasticsearch(["localhost:9200"]) dict_of_duplicate_docs = {} # Die folgende Zeile definiert die Felder, # mit denen ermittelt wird, ob ein Dokument ein Duplikat ist keys_to_include_in_hash = ["CAC", "FTSE", "SMI"] # Von der aktuellen Suche zurückgegebene Dokumente verarbeiten/scrollen 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() # Wenn hashval neu ist, generieren wir einen neuen Schlüssel # in dict_of_duplicate_docs und weisen ihm # ein leeres Array als Wert zu. # Anschließend schreiben wir sofort die _id in das Array. # Wenn hashval bereits existiert, # schreiben wir einfach die neue _id in das vorhandene Array dict_of_duplicate_docs.setdefault(hashval, []).append(_id) # Alle Dokumente im Index durchlaufen und die # Datenstruktur dict_of_duplicate_docs füllen. def scroll_over_all_docs(): data = es.search(index="stocks", scroll='1m', body={"query": {"match_all": {}}}) # Scroll-ID abrufen sid = data['_scroll_id'] scroll_size = len(data['hits']['hits']) # Aktuellen Treffer-Batch verarbeiten und anschließend scrollen populate_dict_of_duplicate_docs(data['hits']['hits']) while scroll_size &gt; 0: data = es.scroll(scroll_id=sid, scroll='2m') # Aktuellen Treffer-Batch verarbeiten populate_dict_of_duplicate_docs(data['hits']['hits']) # Scroll-ID aktualisieren sid = data['_scroll_id'] # Anzahl der im letzten Scroll zurückgegebenen Ergebnisse abrufen scroll_size = len(data['hits']['hits']) def loop_over_hashes_and_remove_duplicates(): # Hashes der Dokumentwerte nach # doppelt vorhandenen Hashes durchsuchen for hashval, array_of_ids in dict_of_duplicate_docs.items(): if len(array_of_ids) &gt; 1: print("********** Duplicate docs hash=%s **********" % hashval) # Dokumente abrufen, die zum aktuellen hashval zugeordnet wurden matching_docs = es.mget(index="stocks", doc_type="doc", body={"ids": array_of_ids}) for doc in matching_docs['docs']: # In diesem Beispiel geben wir die doppelten Dokumente nur aus. # Dieser Code kann problemlos geändert werden, # um die Duplikate zu löschen anstatt sie auszugeben. print("doc=%s\n" % doc) def main(): scroll_over_all_docs() loop_over_hashes_and_remove_duplicates() main() </pre>
<h2>Fazit</h2>
In diesem Blogeintrag habe ich Ihnen zwei Methoden zum Deduplizieren von Dokumenten in Elasticsearch vorgestellt. Die erste Methode verwendet Logstash zum Entfernen von doppelten Dokumenten, und die zweite Methode verwendet ein benutzerdefiniertes Python-Skript, um doppelte Dokumente zu finden und zu entfernen.
Falls Sie Fragen zum Deduplizieren von Elasticsearch-Dokumenten oder zu anderen Themen rund um Elasticsearch haben, finden Sie hilfreiche Einblicke und Informationen in unseren <a href="https://discuss.elastic.co/">Diskussionsforen</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

Doppelte Dokumente in Elasticsearch finden und entfernen

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

Durch das Absenden erklären Sie, dass Sie unsere <a href="/de/legal/elastic-cloud-account-terms" target="_self">Nutzungsbedingungen</a> gelesen haben und mit ihnen einverstanden sind und dass Sie Elastic erlauben, <a href="/de/legal/privacy-statement#how-we-use-the-information" target="_self">ausgewählte Nachrichten und Informationen</a> an die oben von Ihnen bereitgestellten Kontaktangaben zu senden. Weitere Informationen, darunter auch dazu, wie Sie Ihre Einwilligung zurückziehen können, finden Sie in der <a href="/de/legal/privacy-statement/" target="_self">Datenschutzerklärung von Elastic</a>.

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

Durch das Absenden erklären Sie, dass Sie unsere <a href="/de/legal/serverless-preview-terms" target="_self">Nutzungsbedingungen</a> gelesen haben und mit ihnen einverstanden sind und dass Sie Elastic erlauben, bezüglich unserer zugehörigen Produkte und Dienstleistungen und unter Nutzung der von Ihnen oben bereitgestellten Kontaktangaben <a href="/de/legal/privacy-statement#how-we-use-the-information" target="_self">mit Ihnen in Kontakt zu treten</a>. Weitere Informationen, darunter auch dazu, wie Sie Ihre Einwilligung zurückziehen können, finden Sie in der <a href="/de/legal/privacy-statement/" target="_self">Datenschutzerklärung von 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

Beim Absenden akzeptieren Sie die <a href="/de/agreements/global/cloud-services-monthly" rel="nofollow">Nutzungsbedingungen von Elastic Cloud</a>. Die Verarbeitung Ihrer personenbezogenen Daten unterliegt der <a href="/de/legal/privacy-statement">Datenschutzerklärung von 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

Durch das Absenden erklären Sie, dass Sie unsere <a href="/de/legal/terms-of-use" target="_self">Nutzungsbedingungen</a> gelesen haben und mit ihnen einverstanden sind und dass Sie Elastic erlauben, bezüglich unserer zugehörigen Produkte und Dienstleistungen und unter Nutzung der von Ihnen oben bereitgestellten Kontaktangaben <a href="/de/legal/privacy-statement#how-we-use-the-information" target="_self">mit Ihnen in Kontakt zu treten</a>. Weitere Informationen, darunter auch dazu, wie Sie Ihre Einwilligung zurückziehen können, finden Sie in der <a href="/de/legal/privacy-statement/" target="_self">Datenschutzerklärung von 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.

Der Inhalt dieser Seite ist in der ausgewählten Sprache nicht verfügbar. Wir bei Elastic arbeiten daran, die bereitgestellten Inhalte in verschiedenen Sprachen anzubieten. Bis dahin bitten wir Sie um etwas Geduld und hoffen auf Ihr Verständnis!

Author

Watch now (no PT)

Watch now

See all top stories

All (no PT translation)

Contact information

Press Release

Share on Reddit

Articles by

Share this story

Share by Email

Thank you for your interest!

Contact Info

Follow us on Youtube

Load More

Share on LinkedIn

Follow us on LinkedIn

Search Integrations

All Solutions

Video for

Follow us on Twitter

See when this webinar starts in my time zone

Register to Watch

Register now

See All Posts

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

Sie können nicht dabei sein? Registrieren Sie sich, damit wir Ihnen die Aufzeichnung zukommen lassen können. Außerdem erhalten Sie eine E‑Mail mit Begleitmaterial.

Small image for

On-demand webinar

Featured webinar

Register to Attend

Share on Facebook

Reset all

Upcoming webinar

View more posts

Print

Hosted by

Sign In to Attend

View next

Follow us on Facebook

Share

Highlights

Learn More

Read less

You'll also receive an email with related content

Wir schicken Ihnen zudem relevante Informationen, die von Interesse sein könnten.

Thank you for registering. We will send you a confirmation email soon.

Vielen Dank für Ihre Registrierung. Sie erhalten in Kürze eine Bestätigungs-E‑Mail von uns.

Author

Articles by Alex Marquardt

Consulting Engineer, Elastic

Erstellen gut pflegbarer und wiederverwendbarer Logstash-Pipelines

RDBMS- und Elasticsearch-Daten mit Logstash und JDBC synchron halten

Aufteilen von Daten und Senden der Daten an unterschiedliche Zielorte mit Logstash

Doppelte Dokumente in Elasticsearch finden und entfernen