Replicación, particionamiento y sharding

Dos motivos distintos para repartir los datos

Un único servidor de base de datos acaba quedándose sin algo: disponibilidad (es una sola máquina que puede fallar), capacidad de lectura, capacidad de escritura, almacenamiento o proximidad a usuarios de otros continentes. Hay dos formas complementarias de repartir los datos entre varias máquinas, y es importante no confundirlas. La replicación mantiene una copia de los mismos datos en varios nodos, para lograr disponibilidad, escalado de lecturas y menor latencia cerca de los usuarios. El particionamiento, llamado sharding en muchos productos, divide los datos para que cada nodo guarde un subconjunto, y así lograr throughput de escritura y almacenamiento más allá de una sola máquina. Los sistemas reales suelen hacer ambas cosas: cada partición se replica.

Ambas tienen consecuencias serias para la corrección. La replicación plantea la pregunta de qué ve un lector mientras las copias no coinciden, que es el tema de modelos de consistencia, CAP y PACELC. El particionamiento plantea la pregunta de qué ocurre con las consultas, los índices y las transacciones que abarcan varias particiones. Esta sección explica los mecanismos con la profundidad suficiente para juzgar lo que afirma una base de datos y para diseñar una clave de sharding.

Replicación líder-seguidor

El esquema más común es la replicación líder-seguidor (también llamada primario-réplica; la documentación antigua dice maestro-esclavo). Un nodo, el líder, acepta todas las escrituras. Registra cada cambio en su log (el write-ahead log de PostgreSQL, el binary log de MySQL) y transmite ese log a los seguidores, que aplican los cambios en el mismo orden. Las lecturas pueden ir al líder o a los seguidores. La replicación en streaming de PostgreSQL, la replicación de MySQL, los grupos de disponibilidad de SQL Server, los conjuntos de réplicas de MongoDB y la mayoría de las bases de datos gestionadas en la nube funcionan así.

La decisión de diseño clave es cuándo comunica el líder que un commit ha tenido éxito. Con la replicación síncrona, espera hasta que un seguidor confirma que tiene el cambio, de modo que una escritura confirmada sobrevive a la pérdida del líder; el precio es que cada commit incluye un viaje de ida y vuelta por la red y que, si el seguidor síncrono está caído, las escrituras se detienen. Con la replicación asíncrona, el líder confirma de inmediato y envía el cambio después; los commits son rápidos y los seguidores pueden estar lejos, pero si el líder muere, se pierden los últimos cambios que no habían llegado a ningún seguidor, aunque al cliente se le dijo que estaban confirmados. Un compromiso habitual, llamado semisíncrono en MySQL, espera a un seguidor y deja que el resto sea asíncrono. En el banco, perder un pago confirmado es inaceptable, así que lo normal es tener al menos una copia síncrona (o una base de datos basada en consenso); para la base de datos de la carta de la plataforma de reparto de comida, unos segundos de posible pérdida pueden ser un precio aceptable a cambio de velocidad. Este es, en forma concreta, el compromiso del RPO visto en disponibilidad y fiabilidad.

El retardo de replicación y las anomalías que causa

Los seguidores asíncronos suelen ir milisegundos por detrás, pero con carga, durante una actualización por lotes grande o a través de una red lenta, el retardo de replicación puede crecer hasta segundos o minutos. Leer de los seguidores produce entonces anomalías que los usuarios notan. Douglas Terry y sus colaboradores definieron las dos primeras garantías que siguen, leer tus propias escrituras y lecturas monótonas, como garantías de sesión en 1994; la tercera importa sobre todo cuando los datos también están particionados.

Ninguna de estas anomalías existe si todas las lecturas van al líder, así que la solución más sencilla suele ser enviar al líder las lecturas que deben estar actualizadas y usar los seguidores solo para las lecturas que toleran datos obsoletos: listados de productos, informes, feeds de búsqueda. Diseña esto de forma deliberada para cada consulta, en lugar de dirigir todas las lecturas a las réplicas por defecto.

Failover y split brain

El failover (conmutación por error) sustituye a un líder que ha fallado por un seguidor. Tiene tres pasos: detectar que el líder ha fallado (normalmente mediante un timeout), elegir el nuevo líder (idealmente el seguidor más actualizado, a menudo mediante un protocolo de consenso como Raft entre los nodos o un coordinador externo como etcd o ZooKeeper) y redirigir las escrituras hacia él. Cada paso puede salir mal. Un timeout demasiado corto provoca un failover durante un breve corte de red o un pico de carga, lo que empeora las cosas; uno demasiado largo alarga la interrupción. Con la replicación asíncrona, se pierden las escrituras que el antiguo líder aún no había enviado y, si el antiguo líder vuelve y todavía las conserva, entran en conflicto con las escrituras nuevas.

El fallo más peligroso es el split brain (cerebro dividido): dos nodos creen ser el líder y ambos aceptan escrituras, de modo que los datos divergen de maneras que puede ser imposible fusionar automáticamente. Suele producirse tras una partición de red en la que el antiguo líder queda aislado pero sigue vivo. La protección necesita un quórum (un nodo solo puede liderar mientras una mayoría esté de acuerdo), fencing (la autoridad del nuevo líder se impone mediante un token siempre creciente que el almacenamiento comprueba, o se apaga a la fuerza el antiguo líder, una técnica apodada STONITH: shoot the other node in the head, es decir, dispara al otro nodo en la cabeza) y clientes que sigan de forma fiable al líder actual.

El incidente de GitHub del 21 de octubre de 2018 muestra lo difícil que es esto en la práctica. Un fallo de red cortó la conexión entre su nodo central de la Costa Este de EE. UU. y su centro de datos principal de la Costa Este durante 43 segundos. Sus herramientas de failover, funcionando según lo previsto, promovieron a primarios las bases de datos de la Costa Oeste. Cuando volvió la conectividad, ambas costas tenían escrituras de las que carecía la otra, así que GitHub no pudo simplemente volver atrás; para proteger la integridad de los datos decidió restaurar y conciliar, y el servicio estuvo degradado durante 24 horas y 11 minutos. El failover automático entre regiones había convertido un corte de 43 segundos en una recuperación de un día entero.

Replicación multilíder y sin líder

En la replicación multilíder, varios nodos aceptan escrituras y se replican entre sí. Se usa para varios centros de datos (cada región escribe localmente, así que los usuarios evitan la latencia de cruzar un océano), para clientes que funcionan sin conexión (una app de calendario o de notas es en la práctica un líder en cada dispositivo) y para la edición colaborativa. El problema inevitable son los conflictos de escritura: dos líderes modifican el mismo registro de forma concurrente. Entre las soluciones están last write wins (sencillo, pero descarta datos en silencio y depende de los relojes), fusionar los valores, conservar ambos y preguntar a una persona, o usar tipos de datos diseñados para fusionarse automáticamente (CRDT, tipos de datos replicados libres de conflictos). El enfoque más fiable es evitar los conflictos: asignar a cada registro una región de origen que sea la única que acepta sus escrituras, como el país de residencia de un cliente.

La replicación sin líder, popularizada por el artículo Dynamo de Amazon (DeCandia y colaboradores, 2007) y usada en Apache Cassandra, ScyllaDB y Riak, no tiene ningún líder. El cliente, o un nodo coordinador que actúa en su nombre, envía cada escritura a las N réplicas de una clave y la considera correcta cuando W de ellas la confirman; envía cada lectura a las réplicas y espera R respuestas, quedándose con la versión más reciente. Si R + W > N, cada conjunto de lectura se solapa con cada conjunto de escritura correcta en al menos una réplica, así que una lectura normalmente ve la última escritura confirmada. Las réplicas que se perdieron escrituras se reparan mediante la reparación en lectura (read repair: el coordinador vuelve a escribir la versión más reciente cuando ve una obsoleta) y una antientropía en segundo plano que compara las réplicas. Cassandra expone estas opciones en cada solicitud como niveles de consistencia, como ONE, QUORUM y ALL.

A menudo se confía demasiado en los quórums. R + W > N no hace por sí solo que el sistema sea linealizable: las escrituras concurrentes, las escrituras que fallaron en algunas réplicas pero tuvieron éxito en otras, los quórums laxos (sloppy quorums: escribir en nodos sustitutos durante los fallos) y last-write-wins con relojes desfasados permiten datos obsoletos o perdidos. El Capítulo 6 recorre la aritmética de los quórums y estos casos límite en CAP y PACELC. Los almacenes sin líder son excelentes para lograr una alta disponibilidad de escritura frente a fallos y entre centros de datos; encajan mal con invariantes como no vender nunca dos veces el mismo asiento, salvo que ofrezcan una transacción ligera o una función de consenso para esas operaciones.

Particionamiento: por rango de claves o por hash

El particionamiento asigna cada registro a una partición según su clave de partición (o clave de sharding), y el objetivo es repartir los datos y la carga de forma uniforme. Una partición que recibe una carga desproporcionada es un punto caliente, y un reparto desigual se llama desequilibrio (skew). Hay dos esquemas básicos.

El particionamiento por rango de claves da a cada partición un rango contiguo de claves, como los tomos de una enciclopedia. Los recorridos por rango son eficientes porque las claves adyacentes se guardan juntas (todos los pedidos de octubre en una o dos particiones), y las particiones pueden dividirse cuando crecen, como en Bigtable, HBase, Spanner y el sharding por rangos de MongoDB. El peligro es una clave que aumenta con el tiempo: si la clave es una marca de tiempo o un número de pedido secuencial, cada escritura nueva cae en la última partición, que se convierte en un punto caliente mientras las demás están ociosas. El particionamiento por hash aplica una función hash a la clave y asigna rangos de valores hash a las particiones, de modo que reparte uniformemente incluso las claves secuenciales. El precio es que las consultas por rango sobre la clave tienen que preguntar ahora a todas las particiones. Muchos almacenes combinan ambos con una clave compuesta: Cassandra aplica un hash a la clave de partición para elegir el nodo y ordena las filas dentro de la partición por columnas de agrupamiento (clustering columns), de modo que (customer_id, placed_at) pone juntos los pedidos de un cliente, en orden temporal, en un solo nodo.

Hashing consistente y reequilibrado

La forma obvia de asignar hashes a nodos, hash(key) mod N, tiene un defecto fatal: cuando N cambia, casi todas las claves se mueven. Pasar de 4 a 5 nodos reubica alrededor del 80% de las claves (cualquier clave cuyo resto difiera para 4 y para 5), y pasar de 10 a 11, alrededor del 91%, lo que significa copiar casi todo el conjunto de datos solo para añadir una máquina. El hashing consistente, introducido por David Karger y sus colaboradores en 1997 para las cachés web, coloca tanto los nodos como las claves en un espacio de hash circular; cada clave pertenece al siguiente nodo en el sentido de las agujas del reloj. Un nodo nuevo asume solo las claves situadas entre él y su predecesor, alrededor de 1/N del total, y al retirar uno, este entrega sus claves a su vecino. Con pocos nodos los arcos son desiguales, así que a cada nodo físico se le asignan muchos nodos virtuales (tokens) en distintos puntos del anillo, lo que equilibra la carga y permite que un nodo nuevo tome un poco de cada nodo existente. Dynamo y Cassandra usan este esquema.

Hay otras dos estrategias de reequilibrado habituales. Con un número fijo de particiones, se crean al principio muchas más particiones que nodos (por ejemplo, 1024 para 10 nodos) y se asignan particiones enteras a los nodos; añadir un nodo mueve unas pocas particiones enteras y nunca vuelve a aplicar el hash a las claves. Los shards de Elasticsearch, los vBuckets de Couchbase y muchos esquemas de sharding hechos a mano usan este enfoque, y la decisión principal es elegir un número de particiones lo bastante grande para el crecimiento futuro. Con el particionamiento dinámico, como en HBase y MongoDB, una partición se divide cuando supera un umbral de tamaño y las mitades pueden moverse a nodos distintos. Sea cual sea el esquema, el reequilibrado mueve grandes cantidades de datos por la red mientras el sistema atiende tráfico, así que muchos equipos prefieren que una persona apruebe cada movimiento, y lo limitan para proteger la latencia.

Las solicitudes también deben dirigirse a la partición correcta: o bien cualquier nodo acepta una solicitud y la reenvía (los nodos de Cassandra conocen el anillo mediante gossip), o bien una capa de enrutamiento conoce el mapa de particiones (el mongos de MongoDB), o bien la biblioteca cliente guarda el mapa. El propio mapa suele guardarse en un servicio de coordinación como etcd o ZooKeeper, o en el grupo de consenso de la propia base de datos.

Claves calientes y desequilibrio

El hash reparte uniformemente las claves, no la carga. Si una clave es mucho más activa que el resto, su partición está caliente sea cual sea el esquema: la cuenta de una celebridad en una red social, el restaurante insignia de una cadena nacional durante una promoción o la propia cuenta de liquidación de un banco, que interviene en cada pago. Los almacenes gestionados hacen explícitos los límites; Amazon DynamoDB, por ejemplo, documenta un máximo por partición de 3000 unidades de lectura y 1000 unidades de escritura por segundo. Entre las mitigaciones están almacenar en caché las lecturas calientes, dividir una clave caliente añadiéndole un pequeño sufijo aleatorio (escrituras de restaurant:88#0 a restaurant:88#9, a costa de leer las diez y combinarlas), mover un cliente (tenant) muy grande a particiones dedicadas y agregar contadores en memoria antes de escribirlos. Todas añaden complejidad, así que primero mide qué claves están realmente calientes.

Índices secundarios entre particiones

El particionamiento funciona sin problemas para las búsquedas por la clave de partición. Las consultas por cualquier otro atributo necesitan un índice secundario, y hay dos formas de particionarlo. Un índice local (particionado por documento) vive con cada partición y cubre solo las filas de esa partición. Las escrituras actualizan una sola partición, pero una consulta por el atributo indexado debe enviarse a todas las particiones y fusionar los resultados (dispersión-recolección, scatter-gather), así que su latencia la marca la partición más lenta y su costo crece con el número de particiones. Un índice global (particionado por término) está a su vez particionado por el valor indexado, así que una consulta lee una sola partición del índice; pero una sola escritura puede necesitar ahora actualizar particiones del índice en otros nodos, lo que normalmente se hace de forma asíncrona, así que el índice puede ir brevemente por detrás de los datos. DynamoDB muestra ambos: un índice secundario local comparte la clave de partición de la tabla, mientras que un índice secundario global tiene la suya propia y es eventualmente consistente.

Errores comunes

Tratar las réplicas como copias de seguridad. Una réplica copia fielmente un DELETE erróneo en milisegundos; siguen haciendo falta copias de seguridad y recuperación a un momento dado (Capítulo 0). Leer de las réplicas por defecto sin decidir qué consultas toleran el retardo. Confiar en un failover automático que nunca se ha ensayado, o que puede promover un nodo mientras el antiguo líder sigue aceptando escrituras. Hacer sharding demasiado pronto: una base de datos con sharding complica los joins, las transacciones, los cambios de esquema y las operaciones, y los nodos individuales modernos son muy grandes; agota primero la indexación, la caché, las réplicas de lectura, el archivado y el escalado vertical. Hacer sharding por una clave que crece de forma monótona, lo que concentra cada inserción en una sola partición. Olvidar las rutas de acceso secundarias y descubrir después de la división que una consulta importante debe preguntar ahora a todos los shards. Como muestra el resto de este capítulo, las buenas alternativas suelen implicar datos derivados: modelos de lectura, cachés y almacenes analíticos, cada uno construido para un tipo de consulta.