Dokumente mit dem NEST Elasticsearch .NET-Client indexieren
Einführung
Es gibt verschiedene Möglichkeiten, Dokumente mit dem NEST Elasticsearch .NET-Client in Elasticsearch zu indexieren.
In diesem Blogbeitrag werden einige der einfachen Methoden vorgestellt, vom Indexieren einzelner Dokumente bis hin zu fortgeschritteneren Methoden unter Verwendung des BulkObservable-Helpers.
Einzelne Dokumente
Innerhalb von NEST wird ein Dokument als POCO (Plain Old CLR Object) modelliert; ein Beispiel finden Sie unten:
public class Person
{
public int Id { get; set; }
public Zeichenfolge FirstName { get; set; }
public Zeichenfolge LastName { get; set; }
}
Eine Instanz dieses Objekts, die ein einzelnes Dokument in Elasticsearch darstellt, kann dann mit verschiedenen Methoden indexiert werden. Verwenden wir die folgende Instanz als Beispiel:
var person = new Person
{
Id = 1,
FirstName = "Martijn",
LastName = "Laarman"
};
Die Methoden IndexDocument<T> und IndexDocumentAsync<T> bieten eine einfache Möglichkeit, ein einzelnes Dokument vom Typ T unter Verwendung von Standardparametern zu indexieren. Das Ergebnis dieses Methodenaufrufs kann inspiziert werden, um festzustellen, ob der Indexierungsvorgang erfolgreich war.
// synchrone Methode, die ein IIndexResponse-Objekt zurückgibt
var indexResponse = client.IndexDocument(person);
//asynchrone Methode, die ein Task<IIndexResponse> zurückgibt, auf das gewartet werden kann
var indexResponseAsync = await client.IndexDocumentAsync(person);
// Ergebnis des synchronen Vorgangs inspizieren
if (!indexResponse.IsValid)
{
// If the request isn't valid, we can take action here
}
Die Eigenschaft IsValid kann verwendet werden, um zu prüfen, ob eine Reaktion funktional gültig ist oder nicht. Dies ist eine NEST-Abstraktion, um einen zentralen Punkt für die Prüfung zu haben, ob bei der Anfrage ein Fehler aufgetreten ist.
Wenn Sie beim Indexieren eines Dokuments zusätzliche Parameter festlegen müssen, können Sie die Fluent- oder Object-Initializer-Syntax verwenden. Dies gibt Ihnen eine feinere Kontrolle über den Indexierungsprozess. Im folgenden Beispiel indexieren wir das Dokument in einen Index namens „people“.
// Fluent-Syntax
var fluentIndexResponse = client.Index(person, i => i.Index("people"));
// Object-Initializer-Syntax
var initializerIndexResponse = client.Index(new IndexRequest<Person>(person, "people"));
Ein naiver Ansatz zum Indexieren mehrerer Dokumente bestünde darin, einfach eine Schleife zu erstellen, um bei jeder Iteration ein einzelnes Dokument zu indexieren; dies ist jedoch ein sehr ineffizienter Ansatz, der bei großen Dokumentsammlungen nicht gut skaliert.
Mehrere Dokumente
Die Bulk-API kann zum Indexieren mehrerer Dokumente verwendet werden. Erstellen wir zunächst eine Sammlung von Dokumenten zum Indexieren:
var people = new []
{
new Person
{
Id = 1,
FirstName = "Martijn",
LastName = "Laarman"
},
new Person
{
Id = 2,
FirstName = "Stuart",
LastName = "Cam"
},
new Person
{
Id = 3,
FirstName = "Russ",
LastName = "Cam"
}
// snip
};
Mehrere Dokumente können mithilfe der Methoden IndexMany und IndexManyAsync entweder synchron oder asynchron indexiert werden. Diese Methoden sind spezifisch für den NEST-Client und kapseln Aufrufe der Bulk-Methode und der Bulk-API des Clients, was eine bequeme Abkürzung für das Indexieren vieler Dokumente darstellt.
Beachten Sie, dass diese Methoden alle Dokumente in einer einzigen HTTP-Anfrage indexieren. Bei sehr großen Dokumentsammlungen müssen Sie die Sammlung daher in viele kleinere Batches unterteilen und mehrere Bulk-Aufrufe ausführen. Wenn dies erforderlich ist, sollten Sie stattdessen den BulkAllObservable<T> -Helper verwenden, der später in diesem Beitrag beschrieben wird.
// synchrone Methode, die eine IBulkResponse zurückgibt
var indexManyResponse = client.IndexMany(people);
if (indexManyResponse.Errors)
{
// die Reaktion kann auf Fehler inspiziert werden
foreach (var itemWithError in indexManyResponse.ItemsWithErrors)
{
// falls Fehler vorliegen, können diese aufgelistet und inspiziert werden
Console.WriteLine("Fehler beim Indexieren des Dokuments {0}: {1}",
itemWithError.Id, itemWithError.Error);
}
}
// alternativ können Dokumente asynchron indexiert werden
var indexManyAsyncResponse = await client.IndexManyAsync(people);
Wenn Sie eine feinere Kontrolle über das Indexieren vieler Dokumente benötigen, können Sie die Methoden Bulk und BulkAsync verwenden und die Deskriptoren nutzen, um die Bulk-Aufrufe anzupassen.
Wie bei den oben genannten IndexMany-Methoden werden Dokumente in einer einzigen HTTP-Anfrage an den _bulk-Endpoint gesendet. Dies bedeutet, dass die Gesamtgröße der HTTP-Anfrage berücksichtigt werden muss. Zum Indexieren einer großen Anzahl von Dokumenten sollten Sie wahrscheinlich den BulkAllObservable<T> -Helper verwenden.
// gibt eine IBulkResponse zurück, die auf Fehler inspiziert werden kann
var bulkIndexResponse = client.Bulk(b => b
.Index("people")
.IndexMany(people)
);
// asynchrone Version
var asyncBulkIndexResponse = await client.BulkAsync(b => b
.Index("people")
.IndexMany(people)
);
BulkAllObservable<T> -Hilfsprogramm
Mit dem Helper BulkAllObservable<T> können Sie sich auf das übergeordnete Ziel konzentrieren, eine Sammlung von Dokumenten zu indexieren, ohne sich um die Mechanismen für Neuversuche, Backoff oder Batch-Verarbeitung kümmern zu müssen.
Mehrere Dokumente können mithilfe der BulkAll-Methode und der BlockingSubscribeExtensions Wait()-Erweiterungsmethode indexiert werden. Dieser Helper bietet Funktionen für automatische Wiederholungsversuche/Backoff bei einem Fehler beim Indexieren sowie zur Steuerung der Anzahl der in einer einzelnen HTTP-Anfrage indexierten Dokumente.
Im folgenden Beispiel indiziert jede Anfrage 1000 Dokumente, die aus dem ursprünglichen Eingang als Batch verarbeitet werden. Bei einer großen Anzahl von Dokumenten kann dies zu vielen HTTP-Anfragen führen, die jeweils 1000 Dokumente enthalten (die letzte Anfrage enthält je nach Gesamtzahl möglicherweise weniger).
Der Helper enumeriert eine IEnumerable<T> -Collection verzögert (lazily), was es Ihnen ermöglicht, eine große Anzahl von Dokumenten einfach zu indexieren, wie z. B. Dokumente, die aus paginierten Datenbankeinträgen materialisiert wurden.
var bulkAllObservable = client.BulkAll(people, b => b
.Index("people")
// Wartezeit zwischen Wiederholungsversuchen
.BackOffTime("30s")
// Anzahl der Wiederholungsversuche bei Fehlern
.BackOffRetries(2)
// Index nach Abschluss des Bulk-Vorgangs aktualisieren
.RefreshOnCompleted()
// Anzahl der gleichzeitigen Bulk-Anfragen
.MaxDegreeOfParallelism(Environment.ProcessorCount)
// Anzahl der Elemente pro Bulk-Anfrage
.Size(1000)
)
// Indexierung durchführen, bis zu 15 Minuten warten.
// Während die BulkAll-Aufrufe asynchron sind, handelt es sich hierbei um einen blockierenden Vorgang
.Wait(TimeSpan.FromMinutes(15), next =>
{
// do something on each response e.g. write number of batches indexed to console
});
Der BulkAllObservable<T> -Helper bietet eine Reihe fortschrittlicher Features.
- BufferToBulk ermöglicht die Anpassung einzelner Operationen innerhalb der Bulk-Anfrage, bevor diese an den Server gesendet wird.
- RetryDocumentPredicate ermöglicht eine feinkörnige Steuerung bei der Entscheidung, ob ein Dokument, das nicht indexiert werden konnte, erneut versucht werden soll.
- DroppedDocumentCallback: Falls ein Dokument auch nach mehreren Versuchen nicht indexiert werden kann, wird dieser Delegate aufgerufen.
client.BulkAll(people, b => b
.BufferToBulk((descriptor, list) =>
{
// individuelle Operationen im Bulk-Request anpassen,
// bevor er gesendet wird
foreach (var item in list)
{
// index each document into either even-index or odd-index
descriptor.Index<Person>(bi => bi
.Index(item.Id % 2 == 0 ? "even-index" : "odd-index")
.Document(item)
);
}
})
.RetryDocumentPredicate((item, person) =>
{
// decide if a document should be retried in the event of a failure
return item.Error.Index == "even-index" && person.FirstName == "Martijn";
})
.DroppedDocumentCallback((item, person) =>
{
// wenn ein Dokument nicht indexiert werden kann, wird dieser Delegate aufgerufen
Console.WriteLine($"Unable to index: {item} {person}");
})
);
Ingest-Knoten
Da Elasticsearch Ingest-Anfragen automatisch an Ingest-Knoten weiterleitet, müssen Sie keine Routing-Informationen angeben oder konfigurieren. Wenn Sie jedoch eine intensive Ingestion durchführen und über dedizierte Ingest-Knoten verfügen, ist es sinnvoll, Index-Anfragen direkt an diese Knoten zu senden, um zusätzliche Hops im Cluster zu vermeiden.
Der einfachste Weg, dies zu erreichen, ist das Erstellen einer dedizierten „Indexieren“-Client-Instanz, die für Anfragen zum Indexieren verwendet wird.
// Liste der Ingest-Knoten
var pool = new StaticConnectionPool(new []
{
new Uri("http://ingestnode1:9200"),
new Uri("http://ingestnode2:9200"),
new Uri("http://ingestnode3:9200")
});
var settings = new ConnectionSettings(pool);
var indexingClient = new ElasticClient(settings);
Bei komplexen Cluster-Konfigurationen kann es einfacher sein, einen Sniffing-Verbindungspool zusammen mit einem Node-Prädikat zu verwenden, um die Nodes herauszufiltern, die über Ingest-Funktionen verfügen. Dies ermöglicht es Ihnen, den Cluster anzupassen, ohne den Client neu konfigurieren zu müssen.
// Liste der Cluster-Nodes
var pool = new SniffingConnectionPool(new []
{
new Uri("http://node1:9200"),
new Uri("http://node2:9200"),
new Uri("http://node3:9200")
});
// Prädikat zur Auswahl nur der Nodes mit Ingest-Funktionen
var settings = new ConnectionSettings(pool).NodePredicate(n => n.IngestEnabled);
var indexingClient = new ElasticClient(settings);
Ingestionspipelines
Ändern wir unseren Person-Typ, um einige zusätzliche Informationen aufzunehmen:
public class Person
{
public int Id { get; set; }
public Zeichenfolge FirstName { get; set; }
public Zeichenfolge LastName { get; set; }
public Zeichenfolge IpAddress { get; set; }
public Zeichenfolge GeoIp { get; set; }
}
public class GeoIp
{
public Zeichenfolge CityName { get; set; }
public Zeichenfolge ContinentName { get; set; }
public Zeichenfolge CountryIsoCode { get; set; }
public GeoLocation Location { get; set; }
public Zeichenfolge RegionName { get; set; }
}
Wir können eine Ingestionspipeline erstellen, die die eingehenden Werte manipuliert, bevor sie indexiert werden. Nehmen wir an, dass unsere Anwendung immer erwartet, dass Nachnamen großgeschrieben werden und Initialen in ein eigenes Feld indexiert werden. Wir haben außerdem eine IP-Adresse, die wir in einen für Menschen lesbaren Standort umwandeln möchten.
Wir könnten diese Anforderung durch das Erstellen eines benutzerdefinierten Mappings und einer Ingest-Pipeline erfüllen. Der neue Person-Typ kann dann ohne weitere Änderungen verwendet werden.
Zuerst erstellen wir den Index und das benutzerdefinierte Mapping:
client.CreateIndex("people", c => c
.Mappings(ms => ms
.Map<Person>(p => p
// Mapping automatisch aus dem Typ erstellen
.AutoMap()
// alle von AutoMap() abgeleiteten Mappings überschreiben
.Eigenschaften(props => props
// ein zusätzliches Feld zum Speichern der Initialen erstellen
.Keyword(t => t.Name("initials"))
// Feld als IP-Adresse-Typ mappen
.Ip(t => t.Name(dv => dv.IpAddress))
// GeoIp als Objekt mappen
.Object<GeoIp>(t => t.Name(dv => dv.GeoIp))
)
)
)
);
Als Nächstes erstellen wir eine Ingest-Pipeline und nutzen dabei das mitgelieferte ingest-geoip-Plugin, das seit Version 6.7 enthalten ist.
client.PutPipeline("person-pipeline", p => p
.Processors(ps => ps
// Nachnamen in Großbuchstaben umwandeln
.Uppercase<Person>(s => s
.Field(t => t.LastName)
)
// Painless-Skript zum Befüllen des neuen Feldes verwenden
.Script(s => s
.Lang("painless")
.Source("ctx.initials = ctx.firstName.substring(0,1) + ctx.lastName.substring(0,1)")
)
// ingest-geoip-Plugin verwenden, um das GeoIp-Objekt aus der bereitgestellten IP-Adresse anzureichern
.GeoIp<Person>(s => s
.Field(i => i.IpAddress)
.TargetField(i => i.GeoIp)
)
)
);
Indizieren wir nun eine Person-Instanz unter Verwendung dieses neuen Index und der Ingest-Pipeline.
var person = new Person
{
Id = 1,
FirstName = "Martijn",
LastName = "Laarman",
IpAddress = "139.130.4.5"
};
// Dokument mithilfe der erstellten Pipeline indizieren
var indexResponse = client.Index(person, p => p
.Index("people")
.Pipeline("person-pipeline")
);
Die Suche zeigt nun das indizierte Dokument mit den angereicherten Werten an.
{
"took": 5,
"timed_out": false,
"_shards": {
"total": 1,
"successful": 1,
"skipped": 0,
"failed": 0
},
"hits": {
"total": 1,
"max_score": 1,
"hits": [
{
"_index": "people",
"_type": "person",
"_id": "1",
"_score": 1,
"_source": {
"firstName": "Martijn",
"lastName": "LAARMAN",
"initials": "ML",
"geoIp": {
"continent_name": "Oceania",
"region_iso_code": "AU-NSW",
"city_name": "Sydney",
"country_iso_code": "AU",
"region_name": "New South Wales",
"location": {
"lon": 151.2167,
"lat": -33.7333
}
},
"ipAddress": "139.130.4.5",
"id": 1
}
}
]
}
}
Wenn eine Pipeline angegeben ist, entsteht beim Indexieren ein zusätzlicher Aufwand für die Dokumentanreicherung; im obigen Beispiel ist dies die Ausführung der Großschreibung und des Painless-Skripts.
Bei großen Bulk-Anfragen kann es ratsam sein, das Standard-Timeout für das Indexieren zu erhöhen, um Ausnahmen zu vermeiden.
client.Bulk(b => b
.Index("people")
.Pipeline("person-pipeline")
// erhöht das Timeout auf der Elasticsearch-Serverseite
.Timeout("5m")
.IndexMany<Person>(people)
.RequestConfiguration(rc => rc
// erhöht das HTTP-Anfrage-Timeout auf dem Client, bevor die Anfrage abgebrochen wird
.RequestTimeout(TimeSpan.FromMinutes(5))
)
);
Zusammenfassung
In diesem Blogbeitrag haben wir den einfachen Fall des Indexierens eines einzelnen Dokuments bis hin zum stapelweisen Indexieren mehrerer Dokumente mit Ingest-Pipelines behandelt.
Probieren Sie es in Ihrem eigenen Cluster aus oder starten Sie eine 14-tägige, kostenlose Testversion des Elasticsearch-Dienstes auf Elastic Cloud. Falls Sie auf Probleme stoßen oder Fragen haben, können Sie diese jederzeit in unseren Discuss -Foren stellen.
Die vollständige Dokumentation zum Indexieren mit dem NEST Elasticsearch .NET-Client finden Sie in unserer Dokumentation.