Elasticsearch-Updates, -Einfügungen und -Löschungen: Funktionsweise und Einschränkungen

Einführung

Die Verwaltung von Streaming-Daten aus einem Quellsystem wie PostgreSQL, MongoDB oder DynamoDB in ein Downstream-System für Echtzeitsuche und -analyse ist für viele Teams eine Herausforderung. Der Datenfluss umfasst häufig komplexe ETL-Tools sowie selbstverwaltende Integrationen, um sicherzustellen, dass Schreibvorgänge mit hohem Volumen, einschließlich Aktualisierungen und Löschungen, weder die CPU belasten noch die Leistung der Endanwendung beeinträchtigen.

Für ein System wie Elasticsearch müssen Ingenieure über umfassende Kenntnisse der zugrunde liegenden Architektur verfügen, um Streaming-Daten effizient verarbeiten zu können. Elasticsearch wurde für die Protokollanalyse entwickelt, bei der sich die Daten nicht häufig ändern, was bei der Verarbeitung von Transaktionsdaten zusätzliche Herausforderungen mit sich bringt.

Rockset hingegen ist eine Cloud-native Datenbank, die viele Tools und den Aufwand für die Dateneingabe in das System überflüssig macht. Da Rockset speziell für die Suche und Analyse in Echtzeit entwickelt wurde, ist es auch auf Veränderlichkeit auf Feldebene ausgelegt, wodurch der CPU-Bedarf für die Verarbeitung von Einfügungen, Aktualisierungen und Löschungen verringert wird.

In diesem Blog vergleichen wir, wie Elasticsearch und Rockset mit der Datenaufnahme umgehen, und stellen praktische Techniken zur Verwendung dieser Systeme für Echtzeitanalysen vor.

Elasticsearch

Datenaufnahme in Elasticsearch

Es gibt viele Möglichkeiten, Daten in Elasticsearch einzuspeisen. Wir behandeln drei gängige Methoden für die Echtzeitsuche und -analyse:

• Daten aus einer relationalen Datenbank mithilfe des Logstash JDBC-Eingabe-Plugins in Elasticsearch aufnehmen

• Übernehmen Sie Daten aus Kafka in Elasticsearch mithilfe des Kafka Elasticsearch Service Sink Connector

• Übernehmen Sie Daten mithilfe der REST-API und Client-Bibliotheken direkt aus der Anwendung in Elasticsearch

  
  

Nehmen Sie Daten aus einer relationalen Datenbank mithilfe des Logstash JDBC-Eingabe-Plugins in Elasticsearch auf. Mit dem Logstash JDBC-Eingabe-Plugin können Sie Daten aus einer relationalen Datenbank wie PostgreSQL oder MySQL für Such- und Analysezwecke in Elasticsearch übertragen.

Logstash ist eine Ereignisverarbeitungs-Pipeline, die Daten aufnimmt und transformiert, bevor sie an Elasticsearch gesendet werden. Logstash bietet ein JDBC-Eingabe-Plugin , das eine relationale Datenbank wie PostgreSQL oder MySQL regelmäßig nach Einfügungen und Aktualisierungen abfragt. Um diesen Dienst nutzen zu können, muss Ihre relationale Datenbank mit Zeitstempeln versehene Datensätze bereitstellen, die von Logstash gelesen werden können, um festzustellen, welche Änderungen aufgetreten sind.

Dieser Aufnahmeansatz funktioniert gut für Einfügungen und Aktualisierungen, aber für Löschungen sind zusätzliche Überlegungen erforderlich. Das liegt daran, dass Logstash nicht feststellen kann, was in Ihrer OLTP-Datenbank gelöscht wurde. Benutzer können diese Einschränkung umgehen, indem sie Soft Deletes implementieren, bei denen ein Flag auf den gelöschten Datensatz angewendet wird, mit dem Daten zum Abfragezeitpunkt herausgefiltert werden. Oder sie können ihre relationale Datenbank regelmäßig scannen, um Zugriff auf die aktuellsten Datensätze zu erhalten und die Daten in Elasticsearch neu zu indizieren.

Übernehmen Sie Daten aus Kafka in Elasticsearch mithilfe des Kafka Elasticsearch Sink Connector . Häufig wird auch eine Event-Streaming-Plattform wie Kafka verwendet, um Daten aus Quellsystemen für die Echtzeitsuche und -analyse an Elasticsearch zu senden.

Confluent und Elastic haben gemeinsam den Kafka Elasticsearch Service Sink Connector herausgebracht, der Unternehmen zur Verfügung steht, die sowohl die verwalteten Confluent Kafka- als auch die Elastic Elasticsearch-Angebote nutzen. Für den Connector ist die Installation und Verwaltung des zusätzlichen Tools Kafka Connect erforderlich.

Mithilfe des Connectors können Sie jedes Thema in Kafka einem einzelnen Indextyp in Elasticsearch zuordnen. Wenn dynamische Typisierung als Indextyp verwendet wird, unterstützt Elasticsearch einige Schemaänderungen wie das Hinzufügen, Entfernen von Feldern und Ändern von Typen.

Eine der Herausforderungen, die bei der Verwendung von Kafka auftreten, besteht darin, dass die Daten in Elasticsearch neu indexiert werden müssen, wenn Sie den Analysator, den Tokenizer oder die indexierten Felder ändern möchten. Dies liegt daran, dass die Zuordnung nicht mehr geändert werden kann, wenn sie einmal definiert ist. Um eine Neuindexierung der Daten durchzuführen, müssen Sie doppelt in den ursprünglichen Index und den neuen Index schreiben, die Daten vom ursprünglichen Index in den neuen Index verschieben und dann den ursprünglichen Connector-Job stoppen.

Wenn Sie keine verwalteten Dienste von Confluent oder Elastic verwenden, können Sie das Open-Source-Kafka-Plugin für Logstash verwenden, um Daten an Elasticsearch zu senden.

Nehmen Sie Daten direkt aus der Anwendung in Elasticsearch auf, indem Sie die REST-API und Client-Bibliotheken verwenden . Elasticsearch bietet die Möglichkeit, unterstützte Client-Bibliotheken wie Java, Javascript, Ruby, Go, Python und mehr zu verwenden, um Daten über die REST-API direkt aus Ihrer Anwendung aufzunehmen. Eine der Herausforderungen bei der Verwendung einer Client-Bibliothek besteht darin, dass sie so konfiguriert werden muss, dass sie mit Warteschlangen und Gegendruck funktioniert, falls Elasticsearch die Aufnahmelast nicht bewältigen kann. Ohne ein vorhandenes Warteschlangensystem besteht die Gefahr eines Datenverlusts in Elasticsearch.

Aktualisierungen, Einfügungen und Löschungen in Elasticsearch

Elasticsearch verfügt über eine Update-API , mit der Aktualisierungen und Löschungen verarbeitet werden können. Die Update-API reduziert die Anzahl der Netzwerkausgänge und das Potenzial für Versionskonflikte. Die Update-API ruft das vorhandene Dokument aus dem Index ab, verarbeitet die Änderung und indexiert die Daten dann erneut. Allerdings bietet Elasticsearch keine direkten Aktualisierungen oder Löschungen. Das gesamte Dokument muss also erneut indexiert werden, ein CPU-intensiver Vorgang.

Im Hintergrund werden Elasticsearch-Daten in einem Lucene-Index gespeichert und dieser Index wird in kleinere Segmente unterteilt. Jedes Segment ist unveränderlich, sodass Dokumente nicht geändert werden können. Wenn eine Aktualisierung vorgenommen wird, wird das alte Dokument zum Löschen markiert und ein neues Dokument wird zusammengeführt, um ein neues Segment zu bilden. Um das aktualisierte Dokument verwenden zu können, müssen alle Analysatoren ausgeführt werden, was auch die CPU-Auslastung erhöhen kann. Bei Kunden mit sich ständig ändernden Daten kommt es häufig vor, dass Indexzusammenführungen einen erheblichen Teil ihrer gesamten Elasticsearch-Rechnerrechnung verschlingen.

Angesichts der erforderlichen Ressourcen empfiehlt Elastic, die Anzahl der Updates in Elasticsearch zu begrenzen. Bol.com , ein Referenzkunde von Elasticsearch, nutzte Elasticsearch für die Site-Suche als Teil seiner E-Commerce-Plattform. Bol.coms Angebote wurden täglich rund 700.000 Mal aktualisiert, darunter Änderungen bei Inhalt, Preisen und Verfügbarkeit. Ursprünglich wollte man eine Lösung, die mit allen Änderungen synchronisiert blieb, sobald diese auftraten. Angesichts der Auswirkungen von Updates auf die Systemleistung von Elasticsearch entschied man sich jedoch für Verzögerungen von 15 bis 20 Minuten. Die Stapelverarbeitung von Dokumenten in Elasticsearch stellte eine konsistente Abfrageleistung sicher.

Löschungen und Segmentzusammenführungsherausforderungen in Elasticsearch

Bei Elasticsearch kann es zu Herausforderungen im Zusammenhang mit der Löschung alter Dokumente und der Wiederherstellung von Speicherplatz kommen.

Elasticsearch führt eine Segmentzusammenführung im Hintergrund durch, wenn ein Index eine große Anzahl von Segmenten enthält oder wenn ein Segment viele Dokumente enthält, die zum Löschen markiert sind. Bei einer Segmentzusammenführung werden Dokumente aus vorhandenen Segmenten in ein neu erstelltes Segment kopiert und die verbleibenden Segmente gelöscht. Leider ist Lucene nicht gut darin, die Größe der zusammenzuführenden Segmente zu bestimmen, wodurch möglicherweise ungleichmäßige Segmente entstehen, die sich auf Leistung und Stabilität auswirken.

Das liegt daran, dass Elasticsearch davon ausgeht, dass alle Dokumente die gleiche Größe haben, und Zusammenführungsentscheidungen auf Grundlage der Anzahl der gelöschten Dokumente trifft. Beim Umgang mit heterogenen Dokumentgrößen, wie es häufig bei Multi-Tenant-Anwendungen der Fall ist, wachsen einige Segmente schneller als andere, was die Leistung für die größten Clients der Anwendung verlangsamt. In diesen Fällen besteht die einzige Abhilfe darin, eine große Datenmenge neu zu indizieren.

Replikationsherausforderungen in Elasticsearch

Elasticsearch verwendet für die Replikation ein primäres Backup-Modell . Das primäre Replikat verarbeitet einen eingehenden Schreibvorgang und leitet den Vorgang dann an seine Replikate weiter. Jedes Replikat empfängt diesen Vorgang und indiziert die Daten erneut lokal. Dies bedeutet, dass jedes Replikat unabhängig voneinander kostspielige Rechenressourcen aufwendet, um dasselbe Dokument immer wieder neu zu indizieren. Wenn n Replikate vorhanden sind, würde Elastic n-mal so viel CPU aufwenden, um dasselbe Dokument zu indizieren. Dies kann die Datenmenge erhöhen, die bei einer Aktualisierung oder Einfügung neu indiziert werden muss.

Bulk-API- und Warteschlangen-Herausforderungen in Elasticsearch

Obwohl Sie die Update-API in Elasticsearch verwenden können, empfiehlt es sich im Allgemeinen, häufige Änderungen mithilfe der Bulk-API zu bündeln. Bei Verwendung der Bulk-API müssen Entwicklungsteams häufig eine Warteschlange erstellen und verwalten, um Aktualisierungen im System zu optimieren.

Eine Warteschlange ist unabhängig von Elasticsearch und muss konfiguriert und verwaltet werden. Die Warteschlange konsolidiert die Einfügungen, Aktualisierungen und Löschungen im System innerhalb eines bestimmten Zeitintervalls, beispielsweise 15 Minuten, um die Auswirkungen auf Elasticsearch zu begrenzen. Das Warteschlangensystem wendet auch eine Drosselung an, wenn die Einfügungsrate hoch ist, um die Anwendungsstabilität sicherzustellen. Warteschlangen sind zwar hilfreich für Aktualisierungen, können jedoch nicht gut feststellen, wann viele Datenänderungen vorliegen, die eine vollständige Neuindizierung der Daten erfordern. Dies kann jederzeit passieren, wenn viele Aktualisierungen am System vorgenommen werden. Es ist üblich, dass Teams, die Elastic in großem Maßstab ausführen, dedizierte Betriebsmitglieder haben, die ihre Warteschlangen täglich verwalten und optimieren.

Neuindizierung in Elasticsearch

Wie im vorherigen Abschnitt erwähnt, werden die Daten neu indiziert, wenn es eine Menge Aktualisierungen gibt oder Sie die Indexzuordnungen ändern müssen. Die Neuindizierung ist fehleranfällig und kann einen Cluster zum Absturz bringen. Noch schlimmer ist, dass die Neuindizierung jederzeit erfolgen kann.

Wenn Sie Ihre Zuordnungen ändern möchten, haben Sie mehr Kontrolle über den Zeitpunkt der Neuindizierung. Elasticsearch verfügt über eine Neuindizierungs-API zum Erstellen eines neuen Index und eine Alias-API , um sicherzustellen, dass es beim Erstellen eines neuen Indexes zu keinen Ausfallzeiten kommt. Mit einer Alias-API werden Abfragen an den Alias oder den alten Index weitergeleitet, während der neue Index erstellt wird. Wenn der neue Index fertig ist, konvertiert die Alias-API, um Daten aus dem neuen Index zu lesen.

Mit der Alias-API ist es immer noch schwierig, den neuen Index mit den neuesten Daten synchron zu halten. Das liegt daran, dass Elasticsearch nur Daten in einen Index schreiben kann. Sie müssen also die Datenpipeline vorgelagert so konfigurieren, dass doppelt in den neuen und den alten Index geschrieben wird.

Rockset

Datenaufnahme in Rockset

Rockset verwendet integrierte Konnektoren, um Ihre Daten mit den Quellsystemen synchron zu halten. Die verwalteten Konnektoren von Rockset sind auf jeden Datenquellentyp abgestimmt, sodass Daten innerhalb von 2 Sekunden aufgenommen und abfragbar gemacht werden können. Dadurch werden manuelle Pipelines vermieden, die zu zusätzlichen Verzögerungen führen oder Daten nur in Mikro-Batches aufnehmen können, beispielsweise alle 15 Minuten.

Auf hohem Niveau bietet Rockset integrierte Konnektoren zu OLTP-Datenbanken, Datenströmen sowie Datenseen und -lagern. So funktionieren sie:

Integrierte Konnektoren zu OLTP-Datenbanken Rockset führt einen ersten Scan Ihrer Tabellen in Ihrer OLTP-Datenbank durch und verwendet dann CDC-Streams , um mit den neuesten Daten synchron zu bleiben. Dabei werden die Daten innerhalb von 2 Sekunden nach ihrer Generierung durch das Quellsystem zur Abfrage bereitgestellt.

Integrierte Konnektoren zu Datenströmen Mit Datenströmen wie Kafka oder Kinesis nimmt Rockset kontinuierlich alle neuen Themen auf und verwendet dazu eine Pull-basierte Integration, die keine Feinabstimmung in Kafka oder Kinesis erfordert.

Integrierte Konnektoren zu Data Lakes und Warehouses Rockset sucht ständig nach Updates und nimmt alle neuen Objekte aus Data Lakes wie S3-Buckets auf. Wir stellen im Allgemeinen fest, dass Teams Echtzeit-Streams mit Daten aus ihren Data Lakes verbinden möchten, um Echtzeitanalysen durchzuführen.

Aktualisierungen, Einfügungen und Löschungen in Rockset

Rockset verfügt über eine verteilte Architektur, die für die effiziente parallele Indizierung von Daten auf mehreren Maschinen optimiert ist.

Rockset ist eine in Dokumente aufgeteilte Datenbank , die ganze Dokumente auf eine einzige Maschine schreibt, anstatt sie aufzuteilen und die verschiedenen Felder an verschiedene Maschinen zu senden. Dadurch können neue Dokumente schnell hinzugefügt oder vorhandene Dokumente basierend auf dem Primärschlüssel _id für Aktualisierungen und Löschungen gefunden werden.

Ähnlich wie Elasticsearch verwendet Rockset Indizes, um Daten bei Abfragen schnell und effizient abzurufen. Im Gegensatz zu anderen Datenbanken oder Suchmaschinen indiziert Rockset Daten jedoch zum Zeitpunkt der Aufnahme in einem konvergenten Index , einem Index, der einen Spaltenspeicher, einen Suchindex und einen Zeilenspeicher kombiniert. Der konvergente Index speichert alle Werte in den Feldern als Reihe von Schlüssel-Wert-Paaren. Im folgenden Beispiel sehen Sie ein Dokument und wie es in Rockset gespeichert wird.

Unter der Haube verwendet Rockset RocksDB , einen hochleistungsfähigen Schlüssel-Wert-Speicher, der Mutationen trivial macht. RocksDB unterstützt atomare Schreib- und Löschvorgänge über verschiedene Schlüssel hinweg. Wenn ein Update für das name eines Dokuments eingeht, müssen genau 3 Schlüssel aktualisiert werden, einer pro Index. Indizes für andere Felder im Dokument sind davon nicht betroffen, was bedeutet, dass Rockset Updates effizient verarbeiten kann, anstatt jedes Mal Zyklen damit zu verschwenden, Indizes für ganze Dokumente zu aktualisieren.

Verschachtelte Dokumente und Arrays sind in Rockset ebenfalls erstklassige Datentypen, was bedeutet, dass für sie der gleiche Aktualisierungsprozess gilt. Dadurch eignet sich Rockset gut für Aktualisierungen von Daten, die in modernen Formaten wie JSON und Avro gespeichert sind.

Das Team bei Rockset hat außerdem mehrere benutzerdefinierte Erweiterungen für RocksDB entwickelt, um hohe Schreib- und Lesevorgänge zu bewältigen, ein häufiges Muster bei Echtzeitanalyse-Workloads. Eine dieser Erweiterungen ist die Remote-Komprimierung , die eine saubere Trennung von Abfrage- und Indexierungsvorgängen in RocksDB Cloud einführt. Dadurch kann Rockset vermeiden, dass Schreibvorgänge Lesevorgänge stören. Dank dieser Verbesserungen kann Rockset seine Schreibvorgänge entsprechend den Kundenanforderungen skalieren und neue Daten für Abfragen bereitstellen, selbst wenn im Hintergrund Mutationen auftreten.

Aktualisierungen, Einfügungen und Löschungen mithilfe der Rockset-API

Benutzer von Rockset können das Standardfeld _id verwenden oder ein bestimmtes Feld als Primärschlüssel angeben. Dieses Feld ermöglicht das Überschreiben eines Dokuments oder eines Teils eines Dokuments. Der Unterschied zwischen Rockset und Elasticsearch besteht darin, dass Rockset den Wert eines einzelnen Felds aktualisieren kann, ohne dass ein ganzes Dokument neu indexiert werden muss.

Um vorhandene Dokumente in einer Sammlung mithilfe der Rockset-API zu aktualisieren, können Sie Anfragen an den Endpunkt „Patch Documents“ stellen. Für jedes vorhandene Dokument, das Sie aktualisieren möchten, geben Sie einfach das Feld _id und eine Liste der Patch-Operationen an, die auf das Dokument angewendet werden sollen.

Die Rockset-API stellt auch einen Endpunkt zum Hinzufügen von Dokumenten bereit, sodass Sie Daten direkt aus Ihrem Anwendungscode in Ihre Sammlungen einfügen können. Um vorhandene Dokumente zu löschen, geben Sie einfach die _id-Felder der Dokumente an, die Sie entfernen möchten, und senden Sie eine Anfrage an den Endpunkt zum Löschen von Dokumenten der Rockset-API.

Umgang mit Replikaten in Rockset

Anders als bei Elasticsearch führt in Rockset nur ein Replikat die Indizierung und Komprimierung mithilfe von RocksDB-Remote-Kompaktierungen durch. Dies reduziert den für die Indizierung erforderlichen CPU-Bedarf, insbesondere wenn aus Gründen der Dauerhaftigkeit mehrere Replikate verwendet werden.

Neuindizierung in Rockset

Beim Aufnehmen in Rockset können Sie mithilfe einer Aufnahmetransformation die gewünschten Datentransformationen angeben, die auf Ihre Rohquelldaten angewendet werden sollen. Wenn Sie die Aufnahmetransformation zu einem späteren Zeitpunkt ändern möchten, müssen Sie Ihre Daten neu indizieren.

Allerdings ermöglicht Rockset eineschemalose Aufnahme und typisiert die Werte jedes Datenfelds dynamisch. Wenn sich Größe und Form der Daten oder Abfragen ändern, bleibt Rockset weiterhin leistungsfähig und erfordert keine Neuindizierung der Daten.

Rockset kann auf Hunderte von Terabyte an Daten skaliert werden, ohne dass es jemals neu indiziert werden muss. Dies geht auf die Sharding-Strategie von Rockset zurück. Wenn die Rechenleistung, die ein Kunde seiner virtuellen Instanz zuweist, zunimmt, wird eine Teilmenge der Shards neu gemischt, um eine bessere Verteilung im Cluster zu erreichen. Dies ermöglicht eine parallelisiertere, schnellere Indizierung und Abfrageausführung. Daher muss in diesen Szenarien keine Neuindizierung erfolgen.

Abschluss

Elasticsearch wurde für die Protokollanalyse entwickelt, bei der Daten nicht häufig aktualisiert, eingefügt oder gelöscht werden. Im Laufe der Zeit haben Teams ihre Nutzung von Elasticsearch erweitert und verwenden Elasticsearch häufig als sekundären Datenspeicher und Indizierungs-Engine für Echtzeitanalysen ständig wechselnder Transaktionsdaten. Dies kann ein kostspieliges Unterfangen sein, insbesondere für Teams, die die Echtzeitaufnahme von Daten optimieren, und es kann einen erheblichen Verwaltungsaufwand verursachen.

Rockset hingegen wurde für Echtzeitanalysen entwickelt und stellt neue Daten innerhalb von 2 Sekunden nach ihrer Generierung für Abfragen bereit. Um diesen Anwendungsfall zu lösen, unterstützt Rockset direkte Einfügungen, Aktualisierungen und Löschungen, wodurch Rechenleistung gespart und die Neuindizierung von Dokumenten eingeschränkt wird. Rockset erkennt auch den Verwaltungsaufwand von Konnektoren und Aufnahme und verfolgt einen Plattformansatz, indem es Echtzeit-Konnektoren in sein Cloud-Angebot integriert.

Insgesamt haben wir Unternehmen erlebt, die für Echtzeitanalysen von Elasticsearch zu Rockset migrieren und allein bei ihren Rechenkosten 44 % sparen. Schließen Sie sich der Welle von Entwicklungsteams an, die innerhalb weniger Tage von Elasticsearch zu Rockset wechseln. Starten Sie noch heute Ihre kostenlose Testversion .