Como ingerir dados no Elasticsearch usando o Apache Camel

O Elasticsearch permite que você indexe dados de maneira rápida e flexível. Experimente gratuitamente na nuvem ou execute-o localmente para ver como a indexação pode ser fácil.

A ingestão de dados no Elasticsearch usando o Apache Camel é um processo que combina a robustez de um mecanismo de busca com a flexibilidade de uma estrutura de integração. Neste artigo, exploraremos como o Apache Camel pode simplificar e otimizar a ingestão de dados no Elasticsearch. Para ilustrar essa funcionalidade, implementaremos uma aplicação introdutória que demonstra, passo a passo, como configurar e usar o Apache Camel para enviar dados para o Elasticsearch.

O que é o Apache Camel?

O Apache Camel é uma estrutura de integração de código aberto que simplifica a conexão de diversos sistemas, permitindo que os desenvolvedores se concentrem na lógica de negócios sem se preocuparem com as complexidades da comunicação entre sistemas. O conceito central do Camel é o de "rotas", que definem o caminho que uma mensagem percorre da origem ao destino, podendo incluir etapas intermediárias como transformações, validações e filtragem.

Arquitetura Apache Camel

O Camel usa "componentes" para se conectar a diferentes sistemas e protocolos, como bancos de dados e serviços de mensagens, e "pontos de extremidade" para representar os pontos de entrada e saída das mensagens. Esses conceitos proporcionam um design modular e flexível, facilitando a configuração e o gerenciamento de integrações complexas de forma eficiente e escalável.

Utilizando Elasticsearch e Apache Camel

Vamos demonstrar como configurar uma aplicação Java simples que utiliza o Apache Camel para ingerir dados em um cluster Elasticsearch. Os processos de criação, atualização e exclusão de dados no Elasticsearch usando rotas definidas no Apache Camel também serão abordados.

1. Adicionando dependências

O primeiro passo para configurar esta integração é adicionar as dependências necessárias ao arquivo pom.xml do seu projeto. Isso incluirá as bibliotecas Apache Camel e Elasticsearch. Usaremos a nova biblioteca Java API Client, portanto, devemos importar o componente camel-elasticsearch e a versão deve ser a mesma da biblioteca camel-core .

Se você deseja usar o cliente REST de baixo nível em Java, deve usar o componente Cliente REST de baixo nível do Elasticsearch.

2. Configurando e executando o contexto do Camel

A configuração começa com a criação de um novo contexto Camel usando a classe DefaultCamelContext , que serve como base para definir e executar rotas. Em seguida, configuramos o componente Elasticsearch, que permitirá que o Apache Camel interaja com um cluster Elasticsearch. A instância ESlasticsearchComponent está configurada para se conectar ao endereço localhost:9200, que é o endereço padrão para um cluster Elasticsearch local. Para uma configuração de ambiente que requer autenticação, você deve ler a documentação sobre como configurar o componente e habilitar a autenticação básica, referida como "Configurar o componente e habilitar a autenticação básica".

Em seguida, esse componente é adicionado ao contexto do Camel, permitindo que as rotas definidas o utilizem para realizar operações no Elasticsearch.

Em seguida, as rotas são adicionadas ao contexto. Criaremos rotas para indexação, atualização e exclusão em massa de documentos.

3. Configurando rotas do Camel

Indexação de dados

A primeira rota que configuraremos é para indexação de dados. Utilizaremos um arquivo JSON contendo um catálogo de filmes. A rota será configurada para ler o arquivo localizado em src/main/resources/movies.json, desserializar o conteúdo JSON em objetos Java e, em seguida, aplicar uma estratégia de agregação para combinar várias mensagens em uma só, permitindo operações em lote no Elasticsearch. O tamanho configurado para 500 itens por mensagem é o seguinte: o processo em lote indexará 500 filmes por vez.

Roteamento Elasticsearch Operação em massa

O lote de documentos será enviado para o endpoint de operações em lote do Elasticsearch. Essa abordagem garante eficiência e rapidez no processamento de grandes volumes de dados.

Atualização de dados

O próximo passo será atualizar os documentos. Na etapa anterior, indexamos alguns filmes e agora criaremos novas rotas para pesquisar um documento por código de referência e, em seguida, atualizar o campo de classificação.

Configuramos um contexto Camel (DefaultCamelContext), onde um componente Elasticsearch é registrado e uma rota personalizada IngestionRoute é adicionada. A operação começa com o envio do código do documento através do ProducerTemplate, que inicia a rota a partir do endpoint direct:update-ingestion.

Em seguida, temos o IngestionRoute, que é o ponto de extremidade de entrada para este fluxo. A rota executa diversas operações em dutos. Primeiro, é feita uma busca no Elasticsearch para localizar o documento pelo código (direct:search-by-id), onde o SearchByCodeProcessor monta a consulta com base no código. Em seguida, o documento recuperado é processado pelo UpdateRatingProcessor, que converte o resultado em objetos Movie, atualiza a classificação do filme para um valor específico e prepara o documento atualizado para ser enviado de volta ao Elasticsearch para atualização.

O processador SearchByCodeProcessor foi configurado apenas para executar a consulta de pesquisa:

O processador UpdateRatingProcessor é responsável por atualizar o campo de classificação.

Exclusão de dados

Por fim, a rota para exclusão de documentos está configurada. Aqui, vamos excluir um documento usando seu ID. No Elasticsearch, para excluir um documento, precisamos saber o identificador do documento, o índice onde o documento está armazenado e executar uma solicitação de exclusão. No Apache Camel, realizaremos essa operação criando uma nova rota, conforme mostrado abaixo.

A rota começa no endpoint direct:op-delete, que serve como ponto de entrada. Quando um documento precisa ser excluído, seu identificador (_id) é recebido no corpo da mensagem. A rota então define o cabeçalho indexId com o valor deste identificador usando simples("${body}"), que extrai o _id do corpo da mensagem.

Por fim, a mensagem é direcionada para o endpoint especificado por URI_DELETE_OPERATION, que se conecta ao Elasticsearch para executar a operação de remoção do documento no índice correspondente.
Agora que criamos a rota, podemos criar um contexto Camel (DefaultCamelContext), que está configurado para incluir o componente Elasticsearch.

Em seguida, a rota de exclusão, definida pela classe OperationDeleteRoute , é adicionada ao contexto. Com o contexto inicializado, um ProducerTemplate é usado para passar o identificador do documento que deve ser excluído para o endpoint direct:op-delete , que aciona a rota de exclusão.

Conclusão

A integração entre o Apache Camel e o Elasticsearch permite uma ingestão de dados robusta e eficiente, aproveitando a flexibilidade do Camel para definir rotas que podem lidar com diferentes cenários de manipulação de dados, como indexação, atualização e exclusão. Com essa configuração, você pode orquestrar e automatizar processos complexos de forma escalável, garantindo que seus dados sejam gerenciados com eficiência no Elasticsearch. Este exemplo demonstrou como essas ferramentas podem ser usadas em conjunto para criar uma solução eficiente e adaptável para ingestão de dados.

Referências

• Camelo Apache
• Arquitetura Apache Camel
• Apache Camel agregado
• Componente de arquivo
• componente Elasticsearch

Perguntas frequentes