Elasticsearch in detail

Tras haberle dedicado tantos artículos en los últimos, parece más que evidentemente que Elasticsearch se merece un post en exclusiva en el que desgranar su secretos más oscuros. ¿Es realmente tan fantástica como parece?

Como ya es costumbre en la casa, en el presente artículo se pretende describir la arquitectura y características de la citada base de datos, no sin antes invitaros a leer la primera y segunda parte acerca de los conceptos básicos de las bases de datos distribuidas.

Introduction

Elasticsearch se describe oficialmente como un motor de analítica y análisis distribuido para todo tipo de datos, incluidos textuales, numéricos, geoespaciales, estructurados y desestructurados. Una definición un tanto comercial para el enfoque que se le pretende dar al escrito.

Dicho de otro modo, Elasticsearch es una base de datos distribuida orientada a documentos, escalable y altamente disponible, optimizada para la búsqueda y recuperación de los datos. Aunque estas características puedan resultar similares a las de otros productos analizados previamente, pronto descubriréis que su posicionamiento es completamente distinto.

Comencemos con un poco de historia. Allá por 2004, un jovencísimo Shay Banon, actual CEO de Elastic, se acababa de mudar a Londres para apoyar a su esposa en su sueño de convertirse en chef. Desesperado por encontrar trabajo, comenzó a trastear y formarse en las nuevas tecnologías del momento, lo que desembocaría en la construcción de un motor de búsquedas basado en Lucene, con el que ayudar a su pareja a capturar todo el conocimiento que estaba acumulando durante sus lecciones de cocina. A esta primera iteración la llamo Compas y fue liberado como proyecto open source unos pocos meses después de su creación.

Por desgracia, lograr que Compass fuera una solución escalable requería reescribir gran parte del código fuente, por lo que Banon decidió comenzar de nuevo y diseñar una solución de búsqueda distribuida que vería la luz en febrero de 2010 bajo el nombre de Elasticsearch.

Así, con el paso de los años se ha convertido en una de las bases de datos mas utilizadas del mundo, especialmente como medio de almacenamiento y explotación de logs y métricas.

Architecture

Elasticsearch se basa una arquitectura descentralizada peer to peer (P2P) de nodos simétrico, en la que, a diferencia de otras soluciones, cada nodo puede desempeñar una o mas funciones.

Por defecto, todos los nodos son de tipo master-eligible, data, coordinator o ingest, si bien para entornos productivos se recomienda encarecidamente configurar un rol especifico a cada uno, con el objetivo de obtener el mejor rendimiento posible.

Master elegible

Nodos candidatos a ser elegidos como maestro. Solo puede existir un nodo maestro activo en todo el cluster, que será el responsable de crear o eliminar índices, decidir qué shards asignar a cada nodo o registrar qué nodos formar parte del clúster, entre otros.

Al igual que otros tantos sistemas, Elasticsearch emplea una estrategia de quorum para determinar cual de todos los nodos elegibles se convierte en el maestro, proceso en el que se requiere que la mitad +1 de los nodos master elegibles participen en la votación (mínimo 3). De esta forma, se evita el “split-brain problem“, escenario en el que un cluster tiene más de un nodo maestro.

Imaginad sino que ocurriría si se perdiera la conectividad en un cluster formado únicamente por dos nodos, con un “minimum_master_nodes” a 1. Efectivamente, en ese momento ambas instancias creerían que son los únicos participantes del cluster y, por tanto, los nodos maestros.

Dicho esto, cuando un nodo es configurado exclusivamente como master elegible, automáticamente dejar de participar en las operaciones de búsqueda e ingesta, con el objetivo de no saturarlo en momentos de alta carga. Por el mismo motivo, se recomienda encarecidamente no utilizarlos para enrutar las solicitudes de búsqueda e indexación de los clientes a los nodos data, a pesar de que puedan desempeñar dicha función. Lo nodos coordinadores son la solución a esto.

Finalmente, y a diferencia de los nodos data, el número nodos master elegible no debe escalar en función del volumen de datos, ya que únicamente puede haber uno activo. De hecho, tampoco es estrictamente necesario que funcionen sobre un hardware extremadamente potente, si bien esto depende del caso de uso. Como siempre, no hay una formula mágica con la que determinar su tamaño, por lo que lo mejor es realizar pruebas de carga y tomar la decisión en función de los resultados.

Data

Nodos encargados de almacenar los datos y realizar las operaciones relacionadas con ellos, como búsquedas, inserciones, actualizaciones, borrados o agregaciones.

Aunque pueda resultar evidentemente, estas operaciones pueden llegar a realizar un uso intensivo de CPU, memoria, disco y ancho de banda, por lo que se recomienda encarecidamente dedicar un tiempo a seleccionar el tipo de instancia y almacenamiento oportuno, acompañado como no, de las pertinentes pruebas de carga.

En esta ocasión sí que existe una formula mágica oficial para el calculo del número de nodos data, en base a la cantidad de información a almacenar y su periodo de retención.

Total Data (GB) = Raw data (GB) per day * Number of days retained * (Number of replicas + 1)
Total Storage (GB) = Total data (GB) * (1 + 0.15 disk Watermark threshold + 0.1 Margin of error)
Total Data Nodes = ROUNDUP(Total storage (GB) / Memory per data node / Memory:Data ratio)

Llevado esto a un sencillo ejemplo, imaginad un sistema de observabilidad basado en Elasticsearch en el que se ingestan 100GB de logs y métricas diarios, con un periodo de retención obligatorio de una semana.

Total Data (GB) = 100GB x 7 x 2 = 1400GB
Total Storage (GB)= 1400GB x (1+0.15+0.1) = 1750GB
Total Data Nodes = ROUNDUP1750GB disk / 16GB RAM /32 ratio) = 4 nodes

En conclusión, harían falta 4 nodos con 16GB de memoria RAM y 512GB de almacenamiento para satisfacer el caso de uso descrito. A modo de curiosidad, desde Elastic recomiendan asignar el 50% de la memoria total a Elasticsearch y el otro 50% restante al sistema operativo, de tal forma que pueda ser utilizada para cachear datos y evitar accesos a disco en operaciones complejas.

Coordinator

Nodos encargados de redirigir las peticiones de los clientes a los nodos data que almacenan los datos, descargando así al resto de integrantes del cluster de esta tarea. Es decir, no almacenan datos ni pueden convertirse en master, simplemente actúan como un enrutador inteligente.

Aunque a primera vista su uso pueda resultar intranscendente, la naturaleza distribuida de Elasticsearch hace que las peticiones de búsqueda o de indexación masiva puedan involucrar a datos almacenados en distintos nodos, llevándose a cabo dicha operación en paralelo, por lo que toda ayuda en bienvenida para no saturar los nodos master elegible o data.

Ingest

Nodos encargados de aplicar pipelines de preprocesado sobre los documentos, con los que transformar y/o enriquecer su contenido antes de indexarlo. Entonces, ¿vienen a suplir a otras herramientas como Logstash o Fluentd? La respuesta correcta ya la podéis intuir y es que depende del caso de uso.

Es decir, es cierto que existe una superposición en la funcionalidad que Logstash/FluentD y el nodo de ingesta de Elasticsearch ofrecen, si bien el abanico de opciones que este ultimo brinda es algo mas limitado.

Dado que no es la intención del escrito realizar una comparativa entre ambas soluciones, en el siguiente enlace dispones un articulo de lo más interesante, en el que se detallan sus principales diferencias.

Por otro lado, e independientemente de su rol, cada nodo dispone de toda la información de enrutamiento necesaria para redirigir cada petición directamente al nodo data apropiado, sin necesidad de comunicarse con un nodo master elegible para conocer la ubicación de los datos o como el estado de salud del resto de los instancias que conforman el cluster.

Esto es gracias al modulo ZenDiscovery, que permite a los nodos cuando arrancan, descubrir al resto de integrantes de la malla y determinar el nodo master, sin tener que depender de una herramienta externa, como, por ejemplo, ZooKeeper.

Para ello, se establece mediante la propiedad “discovery.zen.ping.unicast.hosts” un listado de direcciones de los nodos, preferiblemente master elegibles, con los que inicialmente contactar y posteriormente emplea el protocolo gossip, también conocido como epidemic protocol, para descubrir al resto de participantes.

En definitiva, si bien este diseño de arquitectura en la que los nodos tienen distintos roles añade una ligera complejidad al proceso de aprovisionamiento y mantenimiento del sistema, logra optimizar al máximo la función de cada uno y en una ultima instancia, garantizar un escalado horizontal teóricamente infinito.

Data Modeling

Elasticsearch almacena los datos como documentos JSON en índices, que no son mas que colecciones de documentos relacionados entre sí. Así, un documento correlaciona un conjunto de claves o nombres de campos, con sus valores correspondientes de tipo string, number, Boolean, date, arrays, o geolocations, entre otros.

Hasta hace bien poquito, y con el objetivo de proporcionar multi-tenancy dentro de un índice, cada documento de Elasticsearch disponía de un campo “Type” que permitía diferenciar distintos tipos de documentos en un mismo índice, como si de distintas tablas de una RDBMS se trataran. Por desgracia, esta solución causó mas problemas de lo que realmente llegó a solventar y desde la versión 7.0 esta deprecada, siendo directamente incompatible con la versión 8.0.

La siguiente tabla describe a las mil maravillas lo comentado anteriormente, al comparar la forma en la que Elasticsearch y una BD relacional organizan los datos.

Elasticsearch	SQL Database
Index	Database
Type	Table
Document	Row
Field	Column

A diferencia de otras soluciones como Cassandra o MongoDB, los datos se almacenan internamente en indices invertidos de Apache Lucene, una librería open source escrita en Java por Doug Cutting (creador de Apache Hadoop), que integra potentes funciones de indexación y búsquedas de información textual.

Se podría decir que es el corazón de Elasticsearch y a la postre, lo que le permite ofrecer sencillas APIs para sacarle el mayor rédito posible sin mucho esfuerzo. Pero, ¿Qué es un índice invertido y cómo funciona?

Tal y como describe Wikipedia, un índice invertido es una forma de estructurar la información que va a ser recuperada por un motor de búsqueda, para poder llevar a cabo una búsqueda de texto completa.

Así, un índice invertido se construye mediante la tokenización de los términos en el documento, creando una lista ordenada de todos los términos únicos y asociando una lista de documentos con el lugar donde se puede encontrar la palabra. Se podría decir que es muy similar a un índice al final de un libro, que contiene todas las palabras únicas del mismo y una lista de páginas donde se pueden encontrar dichos términos.

El siguiente ejemplo del blog oficial de Elasticsearch lo ilustra a las mil maravillas.

Se desean indexar 3 documentos simples que contienen las frases “Winter is coming”, “Our is the fury” y “The choice is yours”. Cuando los datos llegan a Elasticsearch, Apache Lucene aplica un procesamiento de texto simple basado en convertir todos los caracteres a minúsculas, eliminar los signos de puntuación y tokenizacion, con el que construir el índice invertido ordenado mostrado en la imagen.

Una vez que los datos han sido correctamente almacenados, operaciones como buscar todos documentos que contengan un determinado termino se vuelven trivibales. Por ejemplo, para obtener aquellos documentos que contienen la palabra “is”, Apache Lucene escanea el índice invertido, buscar el termino solicitado y devuelve los ID de documento que contienen esta palabra, que en este caso serían el Doc 1, 2 y 3.

Para las búsquedas con varios términos se aplica exactamente la misma estrategia: Se buscan todos los términos y sus ocurrencias, y se toma la intersección (para búsquedas AND) o la unión (para búsquedas OR) de los conjuntos de ocurrencias para obtener la lista de documentos resultante.

Partitioning

En Elasticsearch los datos se organizan en índices, que a su vez se componen de uno o más shards, que se distribuyen lo largo de los distintos nodos que forman el clusters. Cada shard es un índice completo de Lucene encargado de almacenar los datos, hasta un máximo de 2,147,483,519 documentos por cada uno, así como de atender las operaciones de consulta, para lo que dispone su propio motor de búsqueda incorporado.

A su vez, cada índice de Lucene se compone de un conjunto de segmentos inmutables creados en cada operación de escritura sobre Elasticsearch. Es decir, cada vez que se agregan nuevos documentos a un índice de Elasticsearch, Lucene crea un nuevo segmento y lo escribe. ¿Y que ocurre entones con las actualizaciones y los borrados? Exactamente lo mismo, Lucene crea un nuevo segmento, en el primer caso con los valores actualizamos y en el segundo marcándolo como eliminado (borrado lógico).

Dado que Lucene realiza las búsquedas en todos los segmentos de forma secuencial, a medida que crece el número segmentos, lo hace también el tiempo de respuesta, motivo por el cual forma periódica fusiona segmentos más pequeños en uno más grande, en el que solo se copian las ultimas versiones de los documentos, excluyendo los eliminados claro está.

En lo que a la distribución de los datos se refiere, los documentos su ubican en un determinado shard en base a un algoritmo de hashing consistente aplicado sobre el campo routing o shard-key, por defecto el ID del documento. Concretamente, se aplica el algoritmo de hashing mumur3 de 32 bits y se divide entre el número de particiones del tópico, logrando así una distribución equitativa de los datos.

shard_num = hash(_routing) % num_primary_shards

Como ya se comentó en anteriores entregas, esta estrategia de distribución de datos tiene un aspecto negativo, ya que el tiempo de respuesta se ve muy mermado al realizar búsquedas complejas en los que datos se encuentran desperdigados entre varios nodos. Para solventar este problema, y tal y como se menciono previamente, el nodo coordinador realiza peticiones en paralelo a todos los shards que puedan verse involucrados en la operación, agrupa los resultados y devuelve la respuesta al usuario.

Dicho esto, Elasticsearch permite configurar el número de shards de los que se compone cada índice, siendo 5 el valor por defecto. La parte negativa reside en que, una vez creado el índice, no es posible modificar dicho valor sin recrearlo, es decir, sin borrarlo, crearlo y reindexar los datos. Esto obliga a realizar un análisis previo, especialmente en entornos productivos, pero raro es encontrar un sistema distribuido que no lo requiera.

Por desgracia, no hay un valor por defecto que se adapte a todos los casos de uso, tan solo una serie de directrices a tener en cuenta. El hecho de definir elevado número de pequeños shards puede ser positivo desde el punto de vista de que es mas sencillo ubicarlos en los nodos, son más eficientes al trabajar con un conjunto de datos mas pequeño o reducen el tiempo de arranque o recuperación del cluster.

Ahora bien, las operaciones de búsqueda tendrán mas posibilidades de involucrar a un mayor número de shard y dado que cada uno de ellos se ejecuta sobre un hilo de CPU, puede acabar agotando el pool de hilos del nodo y en consecuencia, aumentar el tiempo de respuesta de las peticiones y disminuir el throughtput total del sistema. Esta estrategia también penaliza o añade sobrecarga sobre el nodo maestro, el coordinador encargado de gestionarlos.

Por el contrario, configurar un reducido número de shards por índice, reduce esta sobrecarga, pero limita el número máximo de documento que puede almacenar. Además, limita la paralelización en las operaciones de búsqueda y obliga a Lucene a trabajar con segmentos más grandes, lo que en ultima instancia también puede penalizar los tiempos de respuesta.

Visto lo visto, calcular el número adecuado de shards por índice se vuelve crucial si se quiere extraer el mayor rendimiento posible del sistema. Para ello, desde Elastic recomiendan un tamaño de shard comprendido entre 10 y 50 GB, ya que a partir de ahí se complica su recolocación en otros nodos, y alojar un máximo de 20 shard por GB de memoria de la maquina. Es decir, un nodo data con 30GB de RAM debería albergar menos de 600 shards.

Replication

Llega el momento de hablar de la replicación, factor que impacta de lleno en el nivel de disponibilidad y consistencia que es capaz de ofrecer el producto.

La replicación en Elasticsearch no es mas que el proceso de mantener un número configurable de copias de lo shards (por defecto 1) que conforman un índice, distribuidas a lo largo de los distintos nodos del cluster, con el propósito de garantizar la disponibilidad en el caso de que alguno de ellos dejara de funcionar.

Para ello, emplea el primary-backup model descrito en el PacificA paper de Microsoft Research, en el que una de las réplicas de cada shard se designa como líder mientras que el resto actuan de replicas. Así, la replica líder es la encargada de llevar a cabo las operaciones de indexación y su posterior replicación al resto de shards, pero a diferencia de otras soluciones, los shard réplicas si que pueden atender operaciones de consulta.

De esta forma, no solo se garantiza que, el caso de caso de que el shard primario deje de funcionar, otra de las replicas sea designada por el nodo master como la nueva partición líder y pueda continuar atendiendo peticiones de indexación, sino que también permite aprovechar los shard replica para mejorar el rendimiento global del cluster en operaciones de lectura.

Para ello, el nodo maestro mantiene una lista de shards replica in-sync por cada shard primario, sobre los que se deben replicar los datos, estrategia muy similar a la empleada por Kafka. Tal y como su nombre indica, se trata de un conjunto de replicas que han procesado todas las operaciones de indexación y borrado emitidas por el usuario, lo que les permite ser candidatas para ser consideradas como primarias en caso de fallo o atender solicitudes de consulta.

Dicho esto, el proceso de escritura y replicación se compone de las siguientes acciones secuenciales:

El nodo coordinador recibe una petición por parte del usuario y la rediríge al shard primario correspondiente.
El shard primario valida la operación entrante y la rechaza si es estructuralmente inválida, como, por ejemplo, si el documento contiene un campo de tipo texto donde se espera un número.
El shard primario ejecuta la operación localmente.
El shard primario reenvía la operación en paralelo a cada réplica in-sync.
Una vez que todas las réplicas completan con éxito la operación, el shard primario notifica el resultado al usuario.

Para mejorar la resiliencia en las escrituras en el sistema, Elasticsearch permite configurar mediante la propiedad wait_for_active_shards (por defecto uno), la cantidad se shards activos necesarios para llevar a cabo las operaciones de indexación. Si el número requerido es mayor a la cantidad de shards disponibles, la operación de escritura deberá esperar hasta que se hayan iniciado las replicas requeridas o se agote el tiempo de espera.

Pero…¿Que ocurre cuando se produce un error? En el caso de que se produjera un fallo en el shard primario, el nodo que lo aloja enviaría un mensaje al nodo maestro, para que pudiera determinar como primaria una de las réplicas in-sync. Luego, la operación se reenviaría al nuevo shard primario para su procesamiento.

Por el contrario, si el error se produjera en una replica, el shard primario enviaría un mensaje al nodo maestro para solicitar que se elimine del grupo in-sync, evitando así una posible inconsistencia en los datos. Una vez recibida la confirmación por parte del nodo maestro, el shard primario devolvería la respuesta al usuario y se comenzaría la construcción de un shard replica en otro nodo.

Consistency

Tal y como se detallaba en la anterior sección, Elasticsearch emplea un modelo de replicación sicrono con un shard lider encargado de atender las operaciones de indexación y consulta, junto a un conjunto de replicas in-sync, siempre listas para ser promocionadas y atender operaciones de lectura, lo que en teoría garantiza la alta disponibilidad del sistema.

¿Se trata entonces de una base de datos que cumple con las capacidades CP del teorema de CAP? No exactamente, de una forma algo enrevesada se podría decir que su modo de actuación es configurable.

Aunque no se haya mencionada previamente, cuando se realiza una operación de indexación sobre Elasticsearch esta no pasa a ser visible de forma inmediata para el motor de búsquedas. Es decir, Lucene escribe los datos en memoria, pero no crea un nuevo segmento hasta que se ejecuta una operación refresh, algo que la propia base de datos se encarga de realizar por defecto una vez por segundo. Por tanto, esto habilita los dirty-reads y en consecuencia, el sistema deja de garantizar la consistencia.

Entonces, ¿Elasticsearch no es capaz la garantizar la consistencia a pesar de su replicación síncrona? Si, si puede.

Tal y como se detallaba en el artículo Time based conditional update, todos los documentos disponen del campo SeqNoPrimaryTerm, un número de secuencia que identifica cada operación realizada en un documento, para garantizar que el dato leído no ha sido modificado durante el transcurso de la operación. Si a este se le suma la opción de forzar un refresh al finalizar una operación de indexación, se obtiene la tan ansiada consistencia, a costa de una importante penalización en el rendimiento.

En ese caso, ¿Cumpliría con todas las propiedades del teorema de CAP? Tampoco, porque al añadir este tiempo de espera se estaría perdiendo por el camino el “Partition Tolerant”.

Sea como fuere, lo que parece claro es que se trata de un producto AP o AC, dependiendo de su configuración, ¿No? Bueno… En aphyr realizaron un fabuloso artículo en el que recopilaban toda la información facilitada por Elastic para determinar en que categoría del CAP se podía clasificar. En el, se detallaba como la disponibilidad era una de las propiedades que se podía ver comprometida, en el caso en el que fallaran los suficientes nodos, ya que el estado del clúster se volvería rojo y dejaría de atender peticiones tanto de indexación como de consulta sobre dicho índice.

Recordar que para poder cumplir con la propiedad “Availability”, cada solicitud a un nodo que no falla se completada con éxito y devolver un mensaje de error no es valido.

En definitiva, si se tuviera que clasificar Elasticsearch es una categoría de concreta de CAP sería como una base de datos Available and Partition Tolerant. Al fin y al cabo, es bajo la modalidad en la que mejor se desenvuelve y a su vez, la que mejor se ajusta cuando se quiere utilizar como medio de almacenamiento de logs y métricas.

Conclusiones

En conclusión, Elasticsearch es una excelente base de datos distribuida, especialmente diseñada para sacarle todo el provecho posible a su potente motor de búsquedas, aspecto en el que no tiene rival, con el permiso de Apache Solr.

Ahora bien, al igual que el resto de competidores, también tiene otras cualidades en las que no brilla con tanta fuerza y la consistencia es uno de ellas. Por lo que si pretendes utilizarla como sistema de almacenamiento primario, analiza previamente en profundidad el caso de uso, al menos si no quieres llevarte una decepción cuando sea demasiado tarde.

REFERENCIAS

Se recomienda encarecidamente leer los siguientes artículos que han servido de base para el escrito:

Elasticsearch in detail

Introduction