Cómo ingerir datos en Elasticsearch a través de Apache Camel

Elasticsearch te permite indexar datos de manera rápida y flexible. Pruébalo gratis en el cloud o ejecútalo localmente para ver lo fácil que puede ser indexar.

La ingestión de datos en Elasticsearch usando Apache Camel es un proceso que combina la robustez de un motor de búsqueda con la flexibilidad de un marco de integración. En este artículo, exploraremos cómo Apache Camel puede simplificar y optimizar la ingestión de datos en Elasticsearch. Para ilustrar esta funcionalidad, implementaremos una aplicación introductoria que demuestra, paso a paso, cómo configurar y usar Apache Camel para enviar datos a Elasticsearch.

¿Qué es el camello apache?

Apache Camel es un framework de integración de código abierto que simplifica la conexión de sistemas diversos, permitiendo a los desarrolladores centrar en la lógica de negocio sin preocupar por las complejidades de la comunicación del sistema. El concepto central en Camel son las "rutas", que definen el camino que sigue un mensaje desde el origen hasta el destino, incluyendo potencialmente pasos intermedios como transformaciones, validaciones y filtrado.

Arquitectura del camello apache

Camel emplea "componentes" para conectarse a diferentes sistemas y protocolos, como bases de datos y servicios de mensajería, y "endpoints" para representar los puntos de entrada y salida de los mensajes. Estos conceptos proporcionan un diseño modular y flexible, facilitando la configuración y gestión de integraciones complejas de forma eficiente y escalable.

Uso de Elasticsearch y Apache Camel

Demostraremos cómo configurar una aplicación Java sencilla que emplea Apache Camel para ingirir datos en un clúster de Elasticsearch. También se cubrirán los procesos de creación, actualización y eliminación de datos en Elasticsearch usando rutas definidas en Apache Camel (Apache Camel).

1. Agregar dependencias

El primer paso para configurar esta integración es agregar las dependencias necesarias al archivo pom.xml de tu proyecto. Esto incluirá las bibliotecas Apache Camel y Elasticsearch. Vamos a usar la nueva biblioteca cliente de Java API, así que debemos importar el componente camel-elasticsearch y la versión debe ser la misma que la de la biblioteca camel-core .

Si quieres usar el cliente de descanso de bajo nivel de Java, debes emplear el componente de cliente de descanso de bajo nivel de Elasticsearch.

2. Configuración y ejecución del Camel Context

La configuración comienza creando un nuevo contexto Camel usando la clase DefaultCamelContext , que sirve como base para definir y ejecutar rutas. A continuación, configuramos el componente Elasticsearch, que permitirá a Apache Camel interactuar con un clúster de Elasticsearch. La instancia ESlasticsearchComponent está configurada para conectarse a la dirección localhost:9200, que es la dirección predeterminada para un clúster local de Elasticsearch. Para una configuración de entorno que requiera autenticación, deberías leer la documentación sobre cómo configurar el componente y habilitar la autenticación básica, conocida como "Configurar el componente y habilitar la autenticación básica".

Este componente se agrega al contexto de Camel, permitiendo que las rutas definidas empleen este componente para realizar operaciones en Elasticsearch.

Después, las rutas se agregan al contexto. Crearemos rutas para la indexación masiva, actualización y eliminación de documentos.

3. Configuración de rutas Camel

Indexación de datos

La primera ruta que configuraremos es para la indexación de datos. Emplearemos un archivo JSON que contiene un catálogo de películas. La ruta se configurará para leer el archivo ubicado en src/main/resources/movies.json, deserializar el contenido JSON en objetos Java y luego aplicar una estrategia de agregación para combinar múltiples mensajes en uno solo, permitiendo operaciones por lotes en Elasticsearch. Se configuró el tamaño de 500 elementos por mensaje, es decir, el volumen indexará 500 películas a la vez.

Operación de búsqueda elástica de ruta en masa

El lote de documentos se enviará al endpoint de operaciones masivas de Elasticsearch. Este enfoque garantiza eficiencia y rapidez al manejar grandes volúmenes de datos.

Actualización de datos

La siguiente opción será actualizar los documentos. Indexamos algunas películas en el paso anterior y ahora crearemos nuevas rutas para buscar un documento por código de referencia y luego actualizar el campo de calificación.

Configuramos un contexto Camel (DefaultCamelContext), donde se registra un componente de Elasticsearch y se agrega una ruta personalizada llamada IngestionRoute. La operación comienza enviando el código del documento a través de ProducerTemplate, que inicia la ruta desde el punto final direct:update-ingestión.

A continuación, tenemos el IngestionRoute, que es el punto final de entrada para este flujo. La ruta realiza varias operaciones por oleoductos. Primero, se realiza una búsqueda en Elasticsearch para localizar el documento por código (direct:search-by-id), donde el SearchByCodeProcessor ensambla la consulta a partir del código. Después, el documento recuperado es procesado por el UpdateRatingProcessor, que convierte el resultado en objetos Película, actualiza la calificación de la película a un valor específico y prepara el documento actualizado para ser enviado de nuevo a Elasticsearch para su actualización.

El procesador SearchByCodeProcessor estaba configurado solo para ejecutar la consulta de búsqueda:

El procesador UpdateRatingProcessor es responsable de actualizar el campo de calificación.

Eliminación de datos

Finalmente, la ruta para eliminar documentos está configurada. Aquí, eliminaremos un documento usando su ID. En Elasticsearch, para eliminar un documento necesitamos conocer el identificador del documento, el índice donde se almacena el documento y ejecutar una solicitud de borrado. En Apache Camel realizaremos esta operación creando una nueva ruta como se muestra a continuación.

La ruta comienza desde el punto final direct:op-delete, que sirve como punto de entrada. Cuando un documento necesita ser eliminado, su identificador (_id) se recibe en el cuerpo del mensaje. La ruta entonces establece el encabezado indexId con el valor de este identificador usando("${body}")simple , que extrae el _id del cuerpo del mensaje.

Finalmente, el mensaje se dirige al punto final especificado por URI_DELETE_OPERATION, que se conecta con Elasticsearch para realizar la operación de eliminación de documentos en el índice correspondiente.
Ahora que creamos la ruta, podemos crear un contexto Camel (DefaultCamelContext), que está configurado para incluir el componente Elasticsearch.

A continuación, se agrega la ruta de borrado, definida por la clase OperationDeleteRoute , al contexto. Con el contexto inicializado, se emplea un ProducerTemplate para pasar el identificador del documento que debe eliminar al punto final direct:op-delete , lo que activa la ruta de eliminación.

Conclusión

La integración entre Apache Camel y Elasticsearch permite una ingesta de datos robusta y eficiente, aprovechando la flexibilidad de Camel para definir rutas que pueden manejar diferentes escenarios de manipulación de datos, como indexación, actualización y eliminación. Con esta configuración, puedes orquestar y automatizar procesos complejos de forma escalable, cerciorando que tus datos se gestionen de forma eficiente en Elasticsearch. Este ejemplo demostró cómo estas herramientas pueden emplear juntas para crear una solución eficiente y adaptable para la ingestión de datos.

Referencias

• Camello apache
• Arquitectura del camello apache
• Camello apache agregado
• Componente de archivo
• Componente Elasticsearch

Preguntas frecuentes