Lidar com a necessidade de atualizar grandes volumes de dados no Elasticsearch pode ser um desafio de desempenho, especialmente quando as atualizações não são pontuais e afetam milhões de documentos. Uma abordagem comum, como recuperar documentos via cursor ou paginação e depois atualizá-los individualmente ou em pequenos lotes, torna-se inviável devido à sobrecarga de rede e processamanto. Nesses cenários, a funcionalidade de atualização em lote por consulta (_update_by_query) combinada com scripts se destaca como uma solução altamente eficiente.

Este método permite que o Elasticsearch execute a lógica de atualização direetamente no lado do servidor, minimizando a transferência de dados e maximizando a velocidade. O exemplo a seguir demonstra como implementar uma atualização em massa de documentos, modificando campos de data/timestamp, usando o cliente NEST em uma aplicação .NET Core.

Exemplo de Serviço de Atualização em Lote em C# com NEST

Considere uma situação onde precisamos atualizar periodicamente um campo de "data de processamanto" e seu respectivo timestamp para todos os documentos que atendam a um determinado critério, como uma categoria específica e uma data de última modificação antiga. O serviço abaixo ilustra como realizar essa operação em lotes de forma assíncrona.

using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Elasticsearch.Net;
using NEST;
using Microsoft.Extensions.Logging;

public class DocumentUpdaterService
{
    private readonly IElasticClient _elasticClient;
    private readonly ILogger<DocumentUpdaterService> _logger;

    public DocumentUpdaterService(IElasticClient elasticClient, ILogger<DocumentUpdaterService> logger)
    {
        _elasticClient = elasticClient ?? throw new ArgumentNullException(nameof(elasticClient));
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
    }

    /// <summary>
    /// Realiza uma atualização em massa de documentos no Elasticsearch usando _update_by_query com script.
    /// </summary<
    /// <param name="indexName">O nome do índice onde os documentos serão atualizados.</param>
    /// <param name="categoryFilterId">ID da categoria para filtrar documentos.</param>
    /// <returns>O número total de documentos atualizados.</returns>
    public async Task<long> UpdateDocumentsProcessingTimeAsync(string indexName, int categoryFilterId)
    {
        long totalDocumentsUpdated = 0;
        int batchCount = 0;

        // Definimos uma data limite para identificar documentos "antigos" que precisam de atualização.
        // Aqui, consideramos documentos com "LastModified" há mais de um dia.
        var modificationThreshold = DateTime.UtcNow.AddDays(-1);

        _logger.LogInformation("Iniciando processo de atualização em massa no índice '{Index}' para categoria '{Category}'.", indexName, categoryFilterId);

        while (true)
        {
            batchCount++;
            var newProcessedTimestamp = DateTimeOffset.UtcNow.ToUnixTimeSeconds();
            var newProcessDate = DateTime.UtcNow;

            var updateRequest = new UpdateByQueryRequest<MyDocument>(indexName)
            {
                Query = new QueryContainer(new BoolQuery
                {
                    Must = new List<QueryContainer>
                    {
                        new TermQuery { Field = new Field(d => d.CategoryId), Value = categoryFilterId },
                        new DateRangeQuery { Field = new Field(d => d.LastModified), LessThan = modificationThreshold }
                    }
                }),
                Script = new ScriptDescriptor()
                    .Source("ctx._source.processedTimestamp = params.newTimestamp; ctx._source.processDate = params.newDate;")
                    .Params(new Dictionary<string, object>
                    {
                        { "newTimestamp", newProcessedTimestamp },
                        { "newDate", newProcessDate }
                    }),
                WaitForCompletion = true, // Espera a conclusão da operação
                ScrollSize = 2000, // Número de documentos por lote de scroll
                Conflicts = Conflicts.Proceed // Continua mesmo em caso de conflitos de versão
            };

            var response = await _elasticClient.UpdateByQueryAsync(updateRequest);

            if (!response.IsValid)
            {
                _logger.LogError("Falha na execução do lote {Batch} da atualização em massa: {DebugInfo}", batchCount, response.DebugInformation);
                break; // Interrompe o processo em caso de erro
            }

            _logger.LogInformation("Lote {Batch} concluído. Documentos atualizados neste lote: {UpdatedCount}.", batchCount, response.Updated);
            totalDocumentsUpdated += response.Updated;

            if (response.Updated == 0)
            {
                // Nenhum documento foi correspondido ou atualizado neste lote,
                // indicando que todos os documentos relevantes já foram processados.
                break;
            }

            // Uma pequena pausa para evitar sobrecarregar o cluster, se necessário.
            await Task.Delay(TimeSpan.FromSeconds(5));
        }

        _logger.LogInformation("Processo de atualização em massa finalizado. Total de {Total} documentos atualizados em {Batches} lotes.", totalDocumentsUpdated, batchCount);
        return totalDocumentsUpdated;
    }
}

// Modelo de exemplo para os documentos no Elasticsearch
public class MyDocument
{
    public int CategoryId { get; set; }
    public long ProcessedTimestamp { get; set; }
    public DateTime ProcessDate { get; set; }
    public DateTime LastModified { get; set; } // Usado para filtrar documentos antigos
    public string Name { get; set; }
    // Outros campos do seu documento...
}

Estrutura DSL Equivalente

O exemplo em C# acima se traduz para uma requisição DSL (Domain Specific Language) do Elasticsearch de forma semelhante ao seguinte:

POST /my_documents-2024.03/_update_by_query?wait_for_completion=true
{
  "query": {
    "bool": {
      "must": [
        {
          "term": {
            "categoryId": {
              "value": 123
            }
          }
        },
        {
          "range": {
            "lastModified": {
              "lt": "2024-03-01T00:00:00Z"
            }
          }
        }
      ]
    }
  },
  "script": {
    "source": "ctx._source.processedTimestamp = params.newTimestamp; ctx._source.processDate = params.newDate;",
    "params": {
      "newTimestamp": 1710172800,
      "newDate": "2024-03-11T12:00:00"
    }
  },
  "conflicts": "proceed",
  "scroll_size": 2000
}