Designing Data-Intensive Applications : Chapter 6

We must state relationships, not procedures.

Para grandes volúmenes de datos o tasas de lectura muy altas, la replicación no es suficiente, por lo cual es necesario separar los datos en lo que se conoce como Particiones o Sharding

Tecnologia	Nomenclatura
MongoDB	shard
Elasticsearch	shard
HBase	region
Bigtable	tablet
Cassandra	vnode
Couchbase	vBucket

Particionado se le llama a intencionalmente separar una base de datos en versiones mas pequeñas de si misma, cada partición se define de una manera que asegura que una parte de los datos solo pertenece a una partición. En la practica cada partición funciona como una base de datos independiente. La principal razón para hacer esto es escalabilidad, cada partición se puede colocar en diferentes nodos, permitiendo distribuir los datos en varios discos y distribuir la carga de consultas en varios procesadores.

Las queries que operan sobre una sola partición, pueden correr independientemente en cada nodo, permitiendo escalar rápidamente agregando más nodos.

Particionado y Replicación

Si bien cada registro pertenece a una sola partición, es probable que queramos tener copias de esa partición en otros nodos, para dar tolerancia a fallos.

Cada nodo puede ser Lider de una partición y seguidor en otras:

Particionado de datos clave-valor

El objetivo es particionar los datos de la manera mas uniforme entre los nodos disponibles, permitiendo que se pueda escalar linealmente. Si no logramos esa uniformidad, podríamos tener 10 nodos pero solo 1 trabajando constantemente (hot spot), por lo cual no estamos distribuyendo la carga.

Randomico

Una primera aproximación a como lograr que se distribuya correctamente, es repartir randomicamente los datos en los nodos. De esta manera van a estar distribuidos correctamente, pero al momento de hacer una query no podríamos saber en cual partición esta, por lo que deberíamos consultar en todas.

Rango de claves

Si tenemos un conjunto de datos que tienen una clave primaria, podriamos partir el intervalo de las claves primarias (de la A a la Z o cualquier intervalo con mínimo y máximo conocido) en N particiones. Luego al insertar un dato, calculamos a que intervalo corresponde, con ese intervalo sabemos a que nodo corresponde esa partición y lo insertamos.

En cada lectura repetimos el proceso y podemos dirigir la consulta directamente al nodo que tiene los datos para responderla.

Los rangos de cada partición no tienen porque ser iguales, dado que nuestra data podría no distribuirse parejo por cada clave, entonces hay que tener particiones con intervalos diferentes para evitar tener una partición con un volumen demasiado grande.

Este tipo de particionado es el que usan BigTable, HBase y MongoDB (hasta la 2.4).

Dentro de cada partición podemos mantener las claves ordenadas (como en SSTables y LSM-Trees), permitiendo que las búsquedas por rango sean fáciles, además las claves pueden ser tratadas como indices concatenados para levantar multiples registros en una sola query. Por ejemplo si la clave es la fecha en la que ocurrió un evento (year-month-day-hour-minute-second) una query por rango es muy util para traer todos los datos que hay para un mes particular.

El problema con esta técnica es, que si no se tiene cuidado, y por ejemplo se usa una clave que sea year-month-day, las inserciones se harán todas en la misma partición (la del día de hoy) generando un hot spot.

Una manera de evitar esto es poner al inicio de la clave algún valor que distribuya los datos, por ejemplo el nombre del generador del dato, pero generara que para levantar todos los datos de un mes sin importar quien los genero, hay que hacer una query por cada generador.

Hash de la clave

Con una buena función de Hash podemos llevar datos con distribuciones no uniformes hacia una distribución sin tanto sesgos. Puede ser una función de hash de 32 bits, que tome un String cualquiera y devuelva un numero randomico entre 0 y 2³²-1, incluso para 2 strings muy parecidos, los hashes estarán distribuidos uniformemente.

No es necesario que las funciones de hash sean criptograficamente fuertes, Cassandra y MongoDB utilizan MD5 por ejemplo. Es importante tener en cuenta que las funciones de Hash incluidas en algunos lenguajes de programación no son aptas para participando, por ejemplo Java (o Ruby) con el Object.hashCode() genera para una misma key diferentes valores en diferentes procesos.

Una vez que tenemos una función que cumple con esto, asignamos las particiones a rangos de hashes de claves en vez de rangos de clave. Esto permite que los rangos de las particiones estén uniformemente separados también.

El problema de este enfoque, es que perdemos la capacidad de buscar por rangos eficientemente, dos claves que antes estaban contiguas, ahora estarán en diferentes particiones por lo que se pierde el orden. En MongoDB las consultas de tipo rango se envían a todas las particiones, mientras que en CouchBase o Voldemort no están soportadas para la clave primaria.

Cassandra utiliza una mezcla de ambas técnicas de particionado, una tabla puede tener una clave primaria compuesta por varias columnas, siendo la primera un hash (que se usa para seleccionar la partición), pero el resto son usadas como un indice concatenado para poder ordenar los datos dentro de la SSTable. Esto implica que no vamos a poder realizar búsquedas por rangos de la primer columna, pero si por las otras manteniendo constante la primera.

Un ejemplo seria en el modelado de una red social, donde cada comentario se indexa por userId-timestamp, podremos buscar fácilmente todos los comentarios de un usuario (accediendo directamente a la partición) y luego por rango de tiempo en la segunda columna. Cada usuario podrá estar en una partición diferente, pero los comentarios de cada uno estarán en la misma partición ordenados por fecha.

Carga de trabajo sesgada y puntos calientes

Hay escenarios donde por más que hacemos hash de la clave, la carga sigue siendo sesgada, por ejemplo en una red social donde un famoso puede tener muchísimos comentarios en su cuenta, por lo cual si usamos como clave el userId, tendríamos un gran volumen de trafico en el la partición de un famoso.

Por ejemplo 3% del datacenter de Twitter se utilizaba para Justin Bieber solamente.

Para esto normalmente se hacen parches a nivel de la aplicación, donde para estos puntos calientes se utilizan un sufijo o prefijo randomico para ir balanceando la carga entre varias particiones. Esto genera un problema a la hora de leer todos los comentarios que le hicieron al usuario puesto que hay que unir N consultas, además de saber que solo ciertos usuarios tienen este tratamiento en la clave.

Particionando indices secundarios por documento

Se utiliza un indice primario para la partición y luego cada partición maneja su indice secundario (como indices locales), donde guarda para cada filtro, la lista de indices primarios que lo tienen a ese valor.

Esto permite independencia de las particiones para las operaciones de escritura, dado que solo afectan a la partición dueña de ese ID que estamos modificando. Pero para las operaciones de consulta por indice secundario hay que hacer una búsqueda en paralelo en todas las particiones y una union de los resultados, esto es normalmente una operacion costosa, y proclive a tener picos de latencia.

Este es el formato que usan Cassandra, MongoDB y Elasticsearch. Muchos recomiendan que se particióne de una manera que haga que las queries con indice secundario se respondan con una sola partición, pero esto no es normalmente posible.

Particionando indices secundarios por termino

En vez de un indice local a cada partición, se utiliza un indice global que cubre los datos de todas las particiones. Como siempre, no podemos tenerlo en un solo nodo porque seria un cuello de botella (razón por la cual participamos los datos), por lo tanto este indice global también estará particionado, pero no con el mismo criterio que los datos.

La idea es particionar el rango del indice secundario (en este caso los colores de la A a la W en la partición 0 y el resto en la 1), para repartirlo sobre los nodos disponibles.

Se le llama particionado por termino porque es el termino de búsqueda el que determina en que partición esta el indice secundario. Este indice secundario como siempre puede ser usado plano o hacheado, con las ventajas que cada uno trae.

Al usar indices globales las lecturas son mas eficientes que en los indices locales, pero penalizamos las escrituras que deben impactarse en multiples particiones.

Es importante tener en cuenta que al no existir transaccionalidad, la escritura en indices de multiples particiones no es en simultáneo (se realiza asincrónicamente), por lo que una lectura apenas escribimos puede no ver el ultimo cambio realizado en el indice. Amazon DynamoDB asegura que la actualización de los indices secundarios demora menos de 1 segundo, pero solo en circunstancias normales, pudiendo demorar más si hay problemas en la infraestructura.

Oracle es otro que utiliza este sistema, permitiendo elegir que tipo de indices secundarios usar.

Rebalanceo de las particiones

Las bases de datos van cambiando con el tiempo:

• Aumenta la carga y por lo tanto se agregan CPUs para soportarla.
• Aumenta el volumen de datos y por lo tanto es necesario mas disco y RAM para guardarlos.
• Una maquina se rompe y es necesario que otras maquinas tomen los datos de esa.

Cualquiera de estos cambios requieren que los datos y las solicitudes se muevan de un nodo al otro, a esto se le conoce como rebalanceo.

Algunos requerimientos que esperamos cumpla el rebalanceo son:

• Al terminar el rebalanceo, la carga debería quedar distribuida parejo en el cluster.
• Mientras ocurre el rebalanceo, la aplicación tiene que seguir disponible.
• El rebalanceo tiene que mover la menor cantidad de datos posible, para minimizar el uso de red y disco.

Estrategias de Rebalanceo

Hash mod N

Nunca usarla porque genera movimiento de datos ante cambios en la cantidad de nodos del cluster.

Tamaño fijo

Se crean N particiones siendo N mucho mas grande que el tamaño del cluster, por lo cual a cada nodo le quedan varias particiones, si se suma un nuevo nodo simplemente se le sacan algunas a cada nodo. Si un nodo se cae, sus particiones se reparten entre el resto. No hay movimientos de claves a otras particiones, sino solo la asignación de particiones a nodos.

Esto permite incluso suplir diferencias de hardware, donde los nodos más capaces pueden tomar mas particiones. Empieza a ser importante seleccionar el numero correcto de particiones, pocas hayan lento los rebalanceos por su tamaño, muchas generaran sobrecarga.

#### Dinamico

Si se usan particiones por rango de claves, tener un numero fijo de particiones con rangos fijos seria un peligro, puesto que podriamos terminar con todos los datos en una partición. Por eso en varias bases de datos que utilizan este sistema las particiones se crean dinámicamente, por ejemplo en HBase, cuando una partición supera cierto tamaño, se parte en 2 particiones cada una con la mitad de los datos. Lo mismo sucede al eliminar datos, si una partición ocupa menos de cierto margen, se fusiona con las particiones adyacentes.

Al dividirse una partición, su resultado puede moverse a otros nodos. Esto permite que se vaya adaptando al volumen de datos, con pocos datos hay pocas particiones, y estas van aumentando a medida que es necesario. Al inicio se puede iniciar con un mínimo de particiones, para evitar comenzar con un solo nodo. MongoDB desde la 2.4 utiliza esta modalidad (cuando dejo el rango de claves para particionar).

Proporcional a los nodos

En vez del modo dinámico, donde ajustamos las particiones según el volumen de datos, en esta modalidad la cantidad de particiones es fija por nodo. El tamaño de cada partición aumenta con el volumen de datos, mientras que el cluster se mantenga sin cambios. Apenas aparezca un nodo nuevo, las particiones se achican. Dado que un volumen de datos mayor, requiere un cluster mayor para guardarlo, esto genera cierta estabilidad en el tamaño de las particiones.

Cuando un nodo se suma al cluster, elige aleatoriamente una cantidad fija de particiones existentes, las divide a la mitad y se hace dueño de 1 de esas mitades y deja la otra donde estaba. Esta selección aleatoria puede generar divisiones que no sean las mejores, pero se promedia al tomar una gran cantidad de particiones (256 por nodo en Cassandra).

Rebalanceo Automatico o Manual

Tenemos las opciones totalmente automáticas y las manuales, con algunas opciones intermedias como Couchbase que sugiere una asignación de particiones a nodos pero no lo aplica sin confirmación manual.

Totalmente automatico es la manera mas conveniente pero es impredecible, se mueven grandes volúmenes de datos de un nodo al otro y se necesita redirigir el trafico a otro nodo mientras tanto. Esto puede generar problemas de red o de performance en los nodos.

Si ademas tenemos detección de fallas sobre los nodos, el cluster puede entender que los problemas de red de un nodo son razones para quitarle trafico y generar una ola de movimiento de de particiones.

Elasticsearch soporta ambos extremos, asignación automática o movimiento de particiones manualmente a cada nodo.

Ruteo de solicitudes

¿Como sabemos a que ip y puerto conectarnos para buscar una clave dada? No es trivial esto luego de que particionamos la data, los indices y adémas lo replicamos.

Esto es un ejemplo de service discovery, algunas maneras de resolver esto son:

• Permitir que cualquier nodo reciba solicitudes (por detrás de un balanceador de carga), si ese nodo tiene los datos responde, sino le envia la solicitud al nodo indicado, recibe la respuesta y se la devuelve al cliente.
• Enviar todas las solicitudes de clientes a una capa de ruteo primero, que determina a que nodo invocar y le pasa la solicitud, funciona como un balanceador de carga que conoce las particiones.
• Requerir al cliente que conozca las particiones y su asignación a nodos, esto implica que el cliente puede conectarse directamente sin intermediarios a un nodo.

En cualquiera de ellas, tanto el nodo que recibe, la capa de ruteo o el cliente, necesitan detectar cambios en las asignaciones de las particiones.

La solución más común es delegar esta tarea a un servicio de coordinación separado, por ejemplo un ZooKeeper, que lleve un status de la metadata del cluster. Cada nodo se registra en ZooKeeper, y es este el que mantiene un mapeo de las particiones a los nodos. Luego la capa de ruteo o los clientes se subscriben a esta información del ZooKeeper. Este es el encargado de notificar a todos los subscriptores ante cualquier cambio en el cluster.

• HBase y Kafka utilizan esta modalidad para llevar trackeo de las particiones.
• MongoDB hace algo parecido pero utilizando un servidor de configuración y deamons para la capa de ruteo.
• Cassandra usa otro enfoque, llevando el mapeo de particiones en los mensajes de gossip, esto genera una complejidad en los nodos pero independiza de sistemas externos.
• Couchbase no tiene rebalanceo automático por lo cual simplifica este sistema, utiliza una capa de tuteo bastante estática.