Instrucciones

Indexar documentos con el cliente NEST Elasticsearch .NET

By
Stuart Cam

Introducción

Hay varias formas en las que puedes indexar documentos en Elasticsearch usando el cliente NEST Elasticsearch .NET.

Esta publicación de blog demostrará algunos de los métodos sencillos, desde indexar un solo documento a la vez hasta métodos más avanzados que usan el asistente BulkObservable.

Documentos únicos

Dentro de NEST, un documento se modela como POCO (objeto CLR simple), a continuación se muestra un ejemplo:

public class Person
{
    public int Id { get; set; }
    public texto FirstName { get; set; }
    public texto LastName { get; set; }
}

Una instancia de este objeto, que representa un único documento en Elasticsearch, puede indexarse utilizando algunos métodos diferentes. Usemos la siguiente instancia como ejemplo:

var person = new Person
{
    Id = 1,
    FirstName = "Martijn",
    LastName = "Laarman"
};

Los métodos IndexDocument<T> e IndexDocumentAsync<T> proporcionan una forma sencilla de indexar un solo documento de tipo T, usando parámetros predeterminados. El resultado de esta llamada al método puede inspeccionarse para determinar si la operación de indexación fue exitosa.

// método síncrono que devuelve un objeto IIndexResponse
var indexResponse = client.IndexDocument(person);
// método asíncrono que devuelve una Task<IIndexResponse> que puede ser esperada
var indexResponseAsync = await client.IndexDocumentAsync(person);
// Inspeccionar el resultado de la operación síncrona
if (!indexResponse.IsValid)
{
    // If the request isn't valid, we can take action here
}

La propiedad IsValid puede usarse para comprobar si una respuesta es funcionalmente válida o no. Esta es una abstracción de NEST para tener un punto único donde verificar si ocurrió algo incorrecto con la solicitud.

Si necesitas establecer parámetros adicionales al indexar un documento, puedes usar la sintaxis fluida o la del inicializador de objetos. Esto te dará un control más preciso sobre el proceso de indexación. En el ejemplo a continuación, indexaremos el documento en un índice llamado “people”.

// sintaxis fluida
var fluentIndexResponse = client.Index(person, i => i.Index("people"));
// sintaxis del inicializador de objetos
var initializerIndexResponse = client.Index(new IndexRequest<Person>(person, "people"));

Un enfoque ingenuo para indexar varios documentos sería simplemente crear un bucle para indexar un solo documento en cada iteración; sin embargo, este es un enfoque muy ineficiente que no escalará bien para grandes colecciones de documentos.

Varios documentos

La API de bulk se puede usar para indexar múltiples documentos. Primero, creemos una colección de documentos para indexar:

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
};

Se pueden indexar varios documentos usando los métodos IndexMany e IndexManyAsync, ya sea de forma síncrona o asíncrona, respectivamente. Estos métodos son específicos del cliente NEST y envuelven las llamadas al método Bulk y a la API de bulk del cliente, proporcionando un atajo conveniente para indexar muchos documentos.

Ten en cuenta que estos métodos indexan todos los documentos en una única solicitud HTTP, por lo que para colecciones de documentos muy grandes, debes particionar la colección en muchos batches más pequeños y emitir múltiples llamadas Bulk. Cuando necesites hacer esto, considera usar el asistente BulkAllObservable<T> en su lugar, descrito más adelante en la publicación.

// método síncrono que devuelve una IBulkResponse
var indexManyResponse = client.IndexMany(people);
if (indexManyResponse.Errors)
{
    // la respuesta puede inspeccionarse en busca de errores
    foreach (var itemWithError in indexManyResponse.ItemsWithErrors)
    {
        // si hay errores, pueden enumerarse e inspeccionarse
        Console.WriteLine("Error al indexar el documento {0}: {1}",
            itemWithError.Id, itemWithError.Error);
    }
}
// alternativamente, los documentos pueden indexarse de forma asíncrona
var indexManyAsyncResponse = await client.IndexManyAsync(people);

Si requieres un control más preciso sobre indexar muchos documentos, puedes usar los métodos Bulk y BulkAsync y utilizar los descriptores para personalizar las llamadas bulk.

Al igual que con los métodos IndexMany anteriores, los documentos se envían al endpoint _bulk en una sola solicitud HTTP. Esto significa que será necesario considerar el tamaño total de la solicitud HTTP. Para indexar una gran cantidad de documentos, probablemente querrás usar el asistente BulkAllObservable<T>.

// devuelve una IBulkResponse que puede inspeccionarse en busca de errores
var bulkIndexResponse = client.Bulk(b => b
    .Index("people")
    .IndexMany(people)
);
// versión asíncrona
var asyncBulkIndexResponse = await client.BulkAsync(b => b
    .Index("people")
    .IndexMany(people)
);

Asistente BulkAllObservable<T>

Usar el ayudante BulkAllObservable<T> te permite concentrarte en el objetivo general de indexar una recopilación de documentos, sin tener que preocuparte por los mecanismos de reintento, interrupción o batch.

Se pueden indexar varios documentos usando el método BulkAll y el método de extensión BlockingSubscribeExtensions Wait(). Este helper expone la funcionalidad para reintentar / hacer backoff automáticamente en caso de una falla de indexación, y para controlar el número de documentos indexados en una sola solicitud HTTP.

En el siguiente ejemplo, cada solicitud indexa 1000 documentos, procesados en batch desde la entrada original. En caso de una gran cantidad de documentos, esto podría resultar en muchas solicitudes HTTP, cada una con 1000 documentos (la última solicitud puede contener menos, dependiendo del número total).

El asistente enumera de forma diferida una colección IEnumerable<T>, lo que te permite indexar una gran cantidad de documentos fácilmente, como documentos materializados a partir de registros de base de datos paginados.

var bulkAllObservable = client.BulkAll(people, b => b
    .Index("people")
    // cuánto tiempo esperar entre reintentos
    .BackOffTime("30s") 
    // cuántos reintentos se intentan si ocurre un error
    .BackOffRetries(2) 
    // actualizar el índice una vez que se completa la operación masiva
    .RefreshOnCompleted()
    // cuántas solicitudes masivas concurrentes realizar
    .MaxDegreeOfParallelism(Environment.ProcessorCount)
    // número de elementos por solicitud masiva
    .Size(1000)
)
// Realizar la indexación, esperando hasta 15 minutos. 
// Aunque las llamadas BulkAll son asíncronas, esta es una operación de bloqueo
.Wait(TimeSpan.FromMinutes(15), next =>
{
    // do something on each response e.g. write number of batches indexed to console
});

El asistente BulkAllObservable<T> expone una serie de características avanzadas.

  1. BufferToBulk permite la personalización de operaciones individuales dentro de la solicitud bulk antes de que se envíe al servidor.
  2. RetryDocumentPredicate permite un control detallado para decidir si un documento que no pudo ser indexado debe reintentarse.
  3. DroppedDocumentCallback: en caso de que un documento no se indexe, incluso después de reintentar, se llama a este delegado.
client.BulkAll(people, b => b
    .BufferToBulk((descriptor, list) =>
    {
        // personaliza las operaciones individuales en la solicitud
        // bulk antes de que se envíe
        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) =>
    {
        // si un documento no se puede indexar, se llama a este delegado
        Console.WriteLine($"No se puede indexar: {item} {person}");
    })
);

Nodos de ingesta

Dado que Elasticsearch redirigirá automáticamente las solicitudes de ingesta a los nodos de ingestión, no tienes que especificar ni configurar ninguna información de enrutamiento. Sin embargo, si realizas una ingesta intensiva y tienes nodos de ingestión dedicados, tiene sentido enviar las solicitudes de índice directamente a estos nodos para evitar saltos adicionales en el cluster.

La forma más sencilla de lograr esto es crear una instancia de cliente dedicada a la "indexar" y usarla para las solicitudes para indexar.

// lista de nodos de ingestión
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);

En configuraciones de cluster complejas, puede ser más fácil usar un grupo de conexiones de autodescubrir junto con un predicado de Node para filtrar los Node que tienen capacidades de ingesta. Esto te permite personalizar el cluster y no tener que reconfigurar el cliente.

// lista de Node del cluster
var pool = new SniffingConnectionPool(new []
{
    new Uri("http://node1:9200"),
    new Uri("http://node2:9200"),
    new Uri("http://node3:9200")
});
// predicado para seleccionar solo Node con capacidades de ingesta
var settings = new ConnectionSettings(pool).NodePredicate(n => n.IngestEnabled);
var indexingClient = new ElasticClient(settings);

Pipelines de ingesta

Modifiquemos nuestro tipo Person para incluir información adicional:

public class Person
{
    public int Id { get; set; }
    public texto FirstName { get; set; }
    public texto LastName { get; set; }
    public texto IpAddress { get; set; }
    public GeoIp GeoIp { get; set; }
}
public class GeoIp
{
    public texto CityName { get; set; }
    public texto ContinentName { get; set; }
    public texto CountryIsoCode { get; set; }
    public GeoLocation Location { get; set; }
    public texto RegionName { get; set; }
}

Podemos crear un pipeline de ingesta que manipule los valores entrantes antes de que se indexen. Supongamos que nuestra aplicación siempre espera que los apellidos estén en mayúsculas y que las iniciales se indexen en su propio campo. También tenemos una dirección IP que nos gustaría convertir en una ubicación legible para humanos.

Podríamos lograr este requisito creando un mapping personalizado y creando un pipeline de ingesta. El nuevo tipo Person puede utilizarse entonces sin realizar más cambios.

Primero, crearemos el índice y el mapping personalizado:

client.CreateIndex("people", c => c
 .mapping(ms => ms
 .Map(p<Person>=> p
 //crear automáticamente el mapping a partir del tipo
 .AutoMap()
 //sobrescribir cualquier mapping inferido de AutoMap() 
 .propiedades(props => props
 // crear un campo adicional para almacenar las iniciales
 .Keyword(t => t.Name("initials"))
 //mapear campo como tipo dirección IP
 .Ip(t => t.Name(dv => dv.IpAddress))
 
<GeoIp>.Object(t => t.Name(dv => dv.GeoIp)) ) ) ));
 


[[ ##
completed ##]]

A continuación, crearemos un pipeline de ingesta, aprovechando el plugin ingest-geoip incluido, ahora integrado en la versión 6.7.

client.PutPipeline("person-pipeline", p => p
    .Processors(ps => ps
        //convertir el apellido a mayúsculas
        .Uppercase<Person>(s => s
            .Field(t => t.LastName)
        )
        // usar un script de Painless para completar el nuevo campo
        .Script(s => s
            .Lang("painless") 
            .Source("ctx.initials = ctx.firstName.substring(0,1) + ctx.lastName.substring(0,1)")
        )
        // usar el plugin ingest-geoip para enriquecer el objeto GeoIp a partir de la dirección IP proporcionada
        .GeoIp<Person>(s => s 
            .Field(i => i.IpAddress)
            .TargetField(i => i.GeoIp)
        )
    )
);

Ahora vamos a indexar una instancia de Person usando este nuevo índice y pipeline de ingesta.

var person = new Person
{
    Id = 1,
    FirstName = "Martijn",
    LastName = "Laarman",
    IpAddress = "139.130.4.5"
};
// indexar el documento usando el pipeline creado
var indexResponse = client.Index(person, p => p
    .Index("people")
    .Pipeline("person-pipeline")
);

La búsqueda ahora muestra el documento indexado con los valores enriquecidos.

{
  "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
        }
      }
    ]
  }
}

Cuando se especifica un pipeline, habrá el coste adicional de enriquecer documentos al indexar; en el ejemplo anterior, la ejecución de la conversión a mayúsculas y el script de Painless.

Para solicitudes masivas grandes, podría ser prudente aumentar el tiempo de espera de indexar predeterminado para evitar excepciones.

client.Bulk(b => b
    .Index("people")
    .Pipeline("person-pipeline")
    //aumenta el tiempo de espera en el lado del servidor de Elasticsearch
    .Timeout("5m") 
    .IndexMany<Person>(people)
    .RequestConfiguration(rc => rc
        // aumenta el tiempo de espera de la solicitud HTTP en el cliente, antes de abortar la solicitud
        .RequestTimeout(TimeSpan.FromMinutes(5)) 
    )
);

En resumen

En esta publicación de blog hemos cubierto desde el caso sencillo de indexar un solo documento hasta indexar masivamente múltiples documentos con pipelines de ingesta.

Pruébalo en tu propio cluster o activa una prueba gratuita de 14 días de Elasticsearch Service en Elastic Cloud. Y si ejecutas algún problema o pregunta, consulta los foros de Discuss.

Para obtener la documentación completa sobre cómo indexar usando el cliente NEST Elasticsearch .NET, consulta nuestra documentación.