Bases de datos distribuidas: sharding.
Tags: basesdedatos, mongodb, sharding, nosql
Published: 2022-07-15 12:30:38

En el post anterior hablé de algunos conceptos básicos para entender el modelo de documento. Básicamente, el modelo de documento surgió de una necesidad real por parte de los desarrolladores de tener un sistema más idiomático de representar datos y sus relaciones, que se pudiera mapear a los objetos que típicamente creamos cuando usamos patrones de programación orientados a los mismos, OOP.

> **_Básicamente, el modelo de documento surgió de una necesidad real por parte de los desarrolladores de tener un sistema más idiomático de representar datos y sus relaciones, que se pudiera mapear a los objetos_**

Es preciso recordar que las primeras bases de datos relacionales de modelo tabular, surgieron en los 70. Es decir, cuando el uso que se le daba a los sistemas computacionales era diferente del de ahora (¡no comprendía la web, por ejemplo!), y los mismos se desplegaban y servían de manera centralizada. No existía ni una necesidad, ni una infraestructura, ni una governanza que requiriera que las bases de datos se distribuyeran a través de diversas ubicaciones.

> **_Las primeras bases de datos relacionales de modelo tabular, surgieron en la década de 1970. Es decir, cuando el uso que se le daba a los sistemas computacionales era bastante diferente, y los mismos se desplegaban y servían de manera centralizada_**

Pero los tiempos cambiaron y **la necesidad vino determinada por la globalización producto del acceso masivo a los mismos productos, servicios y aplicaciones, por parte de los usuarios de internet a nivel mundial**. Esta globalización exacerbó problemas ya existentes, pero menos relevantes, como la latencia. **La latencia determina cuánto tiempo pasa desde que se inicia una petición hasta que se recibe una respuesta, en un modelo cliente servidor.**


![Latencia en despliegues](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/yhdte0q57obdc2q0ndvk.jpg)

Antes de que las nubes públicas y sus proveedores entraran en escena, los despliegues se hacían 'en premisas', es decir en las mismas premisas donde las empresas y organizaciones operaban sus negocios o en un centro de datos cercano. (Recordemos que todavía hay un enorme porcentaje de organizaciones que por motivos de seguridad o retraso en la adopción de sistemas en la nube, siguen operando en premisas o como se conoce en inglés 'on-prem(ise)'). Más adelante, los centros de datos que se dedicaban a proveer y mantener infraestructuras y las redes necesarias, fueron creciendo y distribuyéndose. Finalmente, el requerimiento de un mejor rendimiento, menos latencia, y de los controles de privacidad en forma de soberanía de datos por parte de países y regiones (como puede ser GDPR en la Unión Europea), definieron la descentralización y los modernos sistemas de distribución de datos.

Pero muchas veces, aunque estos sistemas cachean datos en diversas regiones del globo (es decir, guardan una copia para servir más eficientemente a puntos distribuidos lo que se conoce como CDN), siguen teniendo su nodo transaccional en un único punto. 

## ¿Qué es el sharding? 

Sharding es simplemente ese proceso de dividir grandes volumenes de datos en partes más pequeñas, que se guardan generalmente en particiones físicas o virtuales diferentes. Cualquier base de datos se puede dividir así, pero es un proceso complicado, que requiere toda una refactorización y reconfiguración de acuerdo a la nueva distribución.

Partir modelos tabulares de modo que podamos desplegar parte de la base de datos en una región de la nube, por ejemplo, y otra parte en otra región, exige un diseño de sistema y una infraestructura realmente complejos. Particularmente por cómo está diseñado el modelo relacional, para desacoplar datos a los que se accede de manera frecuente, pero deben seguir relativos a otro (u otros) grupos de datos para conformar una entidad. 

> **_Partir modelos tabulares de modo que podamos desplegar parte de la misma base de datos en una región de la nube, por ejemplo, y otra parte en otra región, exige un diseño de sistema y una infraestructura, realmente complejos._**

![Distribuir modelos tabulares implica esfuerzos mayores de diseño](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/xn5bkedp8ov5t9l5a4m9.jpg)

##El modelo de documento simplifica la distribución de datos

**Sin embargo, el modelo documento propuso que todos los datos a los que se accede juntos, se guardan juntos,** en un único documento. Y aunque existen muchas excepciones a esa regla, y formas de relacionar documentos en distintas colecciones a través de ciertos patrones de diseño, es muy importante tener esta definición en cuenta, a la hora de diseñar bases de datos de modelo de documento.

Una de las ventajas es que es mucho más simple 'dividir' bases de datos en conjuntos de datos completamente diferentes, siguiendo una estrategia de `particionamiento` si respetamos ese precepto. 

En el caso de MongoDB además, el proceso de sharding es nativo y se puede automatizar, interviniendo en él otros procesos como el enrutado, el auto-balanceo, etc.

## Sharding si, sharding no 

El `sharding` de MongoDB no es en lo absoluto un proceso simple. Es un proceso complejo que necesita una estrategia bien meditada para ser exitoso. Una de las preguntas más importantes que debemos hacernos (como con todos los procesos que implican una re-estructuración de la arquitectura, con un coste elevado) es si realmente lo necesitamos.

**_Es importante tener en cuenta que en Cosmos DB, por ejemplo, el sistema está particionado por defecto, en particiones pequeñas. Hablaremos de esto en el siguiente post._**

Para empezar, es importante saber que el `sharding` en una base de datos MongoDB se recomienda solamente para volúmenes muy grandes de datos. Pensemos en varios TB's de los mismos. Generalmente el volumen recomendado estará ligado a la infraestructura donde corre el software y su capacidad del hardware, que determina el rendimiento del motor a nivel escaneado, compresión y cacheado. Pero eso es tema para otro blog post. Así que prosigamos.

Incluso si nuestro volumen de datos comprimido está por encima de los TB's recomendados en ese momento, la primera pregunta que hay que hacerse antes de habilitar `sharding`, tendrá que ver con la arquitectura de nuestros datos, y las politicas de retención de los mismos. Es decir...¿realmente necesitamos tener en caliente, todos los datos que guardamos?.

### Datos fríos y datos calientes

> **_Una de las definiciones más importantes que tiene que hacer una empresa en el momento en que empieza a brindar un servicio a usuarios a través de una aplicación, es la política de retención de datos._**

A nivel estrategia de negocio siempre será más barato, y por ende más ventajoso para nuestra empresa, el definir claramente las expectativas y acuerdos de acceso a los datos por parte de nuestros usuarios. Es decir, imaginemos que somos una `start-up` que recabamos datos a los que los usuarios acceden a través de nuestra aplicación o plataforma, podemos definir que los usuarios tendrán acceso inmediato a los datos de menos de 12 meses de edad en la base de datos, y que deben esperar un poco más tiempo para obtener datos más antiguos. Por un poco más de tiempo, me refiero a algunos segundos en vez de nano o milisegundos.

Ese tipo de términos de uso son generalmente aceptados como normales por los usuarios, y nos permiten mantener datos en caliente por menos tiempo (es decir, en un cluster de acceso inmediato) mientras que los más antiguos de guardan en almacenamiento frío, por ejemplo un `data lake` (lago de datos? Nunca lo he escuchado en español, pero eso sería literalmente el nombre!). Los `data lakes` son almacenamientos en máquinas menos potentes, y por lo tanto son más baratos y admiten volúmenes enormes de datos.

## Múltiples clusters como alternativa al sharding

Otra alternativa para volúmenes grandes de datos que deben estar en caliente, de más fácil implementación y mantenimiento particularmente cuando se debe observar la soberanía de datos, es la de desplegar bases de datos a través de diferentes clusters, en vez de recurrir al `sharding`.

![Multiples shards distribuidos](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/x2ekg2gi744dqyw4497h.jpg)

## Qué pasa en nuestra base de datos, cuando la particionamos?

Ahora que ya hemos hablado de volúmenes mínimos de datos, y de políticas de datos que nos permitan mover datos a almacenamientos más baratos, y que ya hemos establecido que el `sharding` es un proceso caro a nivel complejidad de la implementación y mantenimiento, hablemos del mecanismo de `sharding` en si.

El `sharding` es un proceso que reconfigura al driver para que la aplicación ignore que se está comunicando con múltiples instancias de la base de datos, y ejecuta un balanceo de cargas que mueve datos automáticamente entre las mismas.

> **_El `sharding` es un proceso que reconfigura al driver para que la aplicación ignore que se está comunicando con múltiples instancias de la base de datos, y ejecuta un balanceo de cargas que mueve datos automáticamente entre las mismas, de acuerdo a una estrategia predefinida._**

Así como enfatizamos en que la replicación es una copia del mismo conjunto de datos que permite la redundancia de los mismos en varios nodos, es importante recordar que cada `shard` o partición, es (si se siguen buenas prácticas) un conjunto de réplicas en si.

> **_Cada `shard` o partición, es un conjunto de réplicas en sí. Técnicamente es posible que cada shard sea un nodo único standalone, pero recordemos que eso no es lo recomendado_**

En un momento hablaremos de las estrategias de `sharding`, que dependen de su `clave de sharding`. La clave de `sharding` o `sharding key` es básicamente el campo que define cómo se separarán los datos, y cómo se indexarán. Antiguamente, una vez se elegía una clave, no se podía cambiar, pero desde la versión 5.0 de MongoDB se puede cambiar con un proceso de `resharding`.

## Componentes de la arquitectura de `sharding`

Una de las razones por las que el `sharding` es más caro de mantener, es que requiere varios nodos adicionales al `standalone` o a las `réplicas` (en el caso de que cada `shard` sea un conjunto de réplicas, como se recomienda). 

Vamos a suponer que tenemos 3 `shards` de arquitectura PSS. Uno de los 3 `shards` va a ser el primario. Además, necesitaremos un nodo adicional que corra el proceso `mongos`. También necesitaremos un servidor de configuración, que también es un conjunto de réplicas. ¡Con esta configuración y usando buenas practicas, tenemos nada menos que 13 nodos corriendo! 


![arquitectura mongodb sharding](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/r9ni9h65brqwljtyllg4.jpg)

## El proceso Mongos

Este proceso es un proceso de enrutado (un router) que mantiene una tabla de contenido de la base de datos que dirige las peticiones del cliente hacia el `shard` correcto, y actúa como intermediario entre la aplicación y la base de datos particionada. La arquitectura acabará siendo como se ilustra en el diagrama anterior.

Cuando habilitamos `sharding` en una base de datos, para una o varias colecciones, debemos elegir la `clave de shard`, que como decíamos, no es más que un campo indexado de la colección en cuestión. Voy a dedicar una entrada de blog a los índices (indexes en la jerga de MongoDB, no sé por qué, pero el plural en este contexto es así 🤷‍♀️). La indexación es uno de los temas más complejos y difíciles de entender, en lo que a MongoDB respecta.

### Shards

Cuando se inicia el proceso de `sharding`, que por cierto es un proceso paulatino y asíncrono que puede tardar días y hasta semanas en completarse (dependiendo del volumen de datos), pero que no interfiere con el funcionamiento de la base de datos (es decir, se puede ejecutar en sistemas en producción, sin problemas), los datos de las colecciones afectadas se irán distribuyendo en forma de chunks del tamaño predeterminado, a través de la cantidad de shards configurada.

**_Un cluster particionado siempre tiene además un `shard primario` que es el que guarda las colecciones que no están sujetas a partición._**

### Chunks

Durante este proceso, el sistema particionará los datos en `chunks` o `cachos`(muy gracioso, pero no encuentro mejor traducción!) y los irá distribuyendo y redistribuyendo de manera balanceada, a través de los `shards`. 

Estos chunks deben de tener, por convención, un tamaño mínimo de 64MB. No tiene sentido tener chunks de menos tamaño. Recordar que el tamaño máximo de un documento MongoDB es 16MB. 

> **_El `sharding` es un proceso paulatino y asíncrono que puede tardar días y hasta semanas en completarse, dependiendo del volumen de datos, y no interfiere con el funcionamiento de la base de datos; se puede ejecutar en sistemas en producción._**

### Consultas (queries)

Las consultas que ejecutamos contra un sistema particionado son de dos tipos: 

`targeted queries`, vendría a ser como `consultas objetivas`, y son las que contienen la clave de `sharding`, y que pueden ser dirigidas por `mongos` al `shard` que contiene los datos relevantes.

Por otro lados las consultas que no contienen la clave, se deben de mandar a todos los `shards` y se conocen como `broadcast` o `scatter-and-gather`, (algo así como `desparramar-reunir`), ya que son consultas que se `lanzan` o `desparraman` entre todos los `shards` para finalmente `reunir` los resultados.

## Habilitando un sistema de `sharding`

Mientras que el `sharding` se habilita a nivel base de datos es importante entender que el mismo se ejecuta a nivel de colección. Más abajo describo algunos métodos.

Podemos utilizar la clase `ShardingTest` del servidor de MongoDB que instalamos con la versión Community, para crear un cluster de prueba. Es una clase que se usa internamente, para hacer pruebas de `sharding`, pero que está expuesta para uso externo. Podemos ver la implementación, [aquí](https://github.com/mongodb/mongo/blob/master/src/mongo/shell/shardingtest.js). 

_Nota: Es importante saber que esto está disponible solo para la versión anterior de la herramienta de linea de comandos `mongo` y no para `mongosh`. Intentaré para el siguiente post encontrar tiempo de cargar los scripts y snippets de compatibilidad para la clase y jstest, pero no prometo nada!)_

Lo primero que hacemos es inicializar la consola de mongo

```
mongo --nodb --norc
```

con los parametros --nodb y --norc para no inicializar una base de datos ni la evaluación de JavaScript.

Luego podemos crear un par de shards de test así

```
testdesharding = ShardingTest({
  name: "TestdeSharding",
  shards: 2,
  chunkSize: 1
});
```
_Nota: Si esta clase nos da un error de boost relativo a la no existencia de la ruta `data/db`, se la podemos pasar al task runner así_

`MongoRunner.datapath = 'mi-ruta/es-tu-ruta/data/path';`
(hay gente que va a entender la referencia y otra que no...jaja)

Hay muchísimas más opciones de configuración que podemos pasar a la hora de crear nuestro cluster de test, que están descritas a partir de la línea 8 del [script](https://github.com/mongodb/mongo/blob/master/src/mongo/shell/shardingtest.js). Entre ellas encontramos:

- name (nombre del cluster)
- shards (número de shards)
- rs (al que podemos pasar un objeto de configuración para cada conjunto de replica, como cantidad de nodos y tamaño del oplog)
- mongos (número de mongos que se correrán)
- chunkSize (el tamaño de cada chunk)
- enableBalancer (si habilitamos el load balancer por defecto)
- enableAutoSplit (si habilitamos la auto partición)
 
etc

Cuando ejecutamos esta clase de test, creamos un cluster y se inicializan los procesos necesarios, incluyendo `mongos` y los conjuntos de réplica configurados, así como el servidor de configuración. 

Una vez creado el cluster, podemos probar los [métodos de sharding](https://www.mongodb.com/docs/manual/reference/method/js-sharding/), que se describen en la documentación oficial del software. También podremos experimentar con las diversas estrategias de `sharding`. Por defecto el proceso `mongos` corre  en el puerto 20009, pero se puede reconfigurar.

Para experimentar, abrimos otra ventana o pestaña de la terminal donde podemos crear una base de datos e importar o crear algunos datos, para poder experimentar.

Esto es solamente para hacer pruebas y aprender. Si fuéramos a habilitar un sistema de `sharding` de MongoDB real, deberíamos seguir el proceso que se describe [aquí](https://www.mongodb.com/docs/manual/core/sharded-cluster-components/).

Básicamente deberíamos inicializar todos los nodos por separados, **primero que nada un conjunto de réplicas para el servidor de configuración**, que se inicializa con el parámetro `--configsvr` (siempre antes que el proceso `mongos`). Otro nodo para `mongos` que se inicializa con el comando `mongos` y el parámetro `--configdb` (al que se pasan los datos de los servidores de configuración) Finalmente conectamos al que será el shard primario, asígnandole el rol a cada miembro con la opción `--shardsvr`, lo que deberemos repetir con cuantos nodos decidamos tener para los conjuntos de réplicas que conformaran cada `shard`.

Una vez tenemos todos los nodos activos, se pueden agregar al cluster de `sharding` manualmente con el comando

`sh.addShard(//réplica set path + puerto);`

al que pasamos cada réplica set individualmente.

### Habilitar el proceso de distribución de datos

Una vez tenemos todo configurado, debemos habilitar manualmente el `sharding` para la base de datos. 

`sh.enableSharding("nombre_base_de_datos");`

y seguidamente, particionamos las colecciones que decidamos

`sh.shardCollection("nombre_base_de_datos.nombre_colección", {"clave_de_partición": 1})`

donde `clave_de_partición` es la clave elegida, o sea un campo **indexado** de la colección.

## Estrategias de `sharding`

Como comentaba anteriormente, la forma más eficiente de distribuir nuestros datos a través de múltiples `shards`, va a determinarse por medio de una estrategia de partición. 

### La estrategia de hash

La estrategia de hashing es la más eficiente a la hora de distribuir datos de manera balanceada. Los datos se distribuyen de manera aleatoria, asegurándose que hay el mismo volumen de datos en cada partición. Es la estrategia ideal cuando la intención es liberar espacio de disco y asegurarnos que los datos se distribuyen de manera equitativa.


![estrategia de sharding: hash](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/1o4ra61dgq0tjbaeefw8.jpg)

### La estrategia de rango

Esta estrategia es la más eficiente para distribuir datos de acuerdo a ciertos rangos. Por ejemplo, si estamos distribuyendo usuarios, los podemos distribuir en el rango de la A a la I, de la J a la Q y de la R a la Z, de acuerdo con la primera letra de su apellido.

Una desventaja de la distribución por rango, es la baja cardinalidad (hablaremos de cardinalidad cuando hablemos de indices) pero básicamente, como habrán muchos apellidos con ciertas letras, y muy pocos con otras, será difícil garantizar el buen balance.

![estrategia sharding: rangos](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/7l5tzylnw9mmko79st6o.jpg)

###  La estrategia de zonas

Finalmente, podemos segmentar los datos ya distribuidos, por zonas. Las zonas pueden venir determinadas por el data _locale_, lo que permitiría distribuir geográficamente, o por otra variable. En el caso de usar la estrategia para distribuir geográficamente,   un `shard` que se encuentre desplegado en cierta región del mundo guardará los datos de los usuarios que pertenecen a esa región. Esta estrategia nos ayuda a cumplir con políticas de protección de datos, por ejemplo, pero al igual que la estrategia de rangos, puede incurrir en un desequilibrio de las particiones.


![estrategia de sharding por zonas](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/wx74kxp5vyairn2fhpod.jpg)

## En el próximo post

La próxima entrada será más corta e intentaré explicar sobre todo,

- Sharding en Producción
- Particionamiento en CosmosDB y como difiere con MongoDB

A prestar atención!