Este es el artículo 1/6 de la serie sobre la ruta de escritura en clústeres fragmentados de MongoDB.
Preguntas que responde este artículo: si se envían 1,000 escrituras bulk en modo ORDERED a una colección fragmentada, ¿cuántos viajes de ida y vuelta ocurren entre mongos y los shards? ¿Por qué basta con que un solo shard se vuelva lento para que todo el bulk se ralentice? Si se cambia a UNORDERED, ¿qué se gana y qué se paga a cambio?
Versión de referencia: MongoDB 4.2 (código fuente en la etiquetar4.2.12, documentación oficial de 4.2). El soporte de 4.2 terminó el 2023-04-30 (EOL). No se ha verificado el comportamiento de mongos en 4.4 y versiones posteriores (a 2026-11).
Cuando mongos recibe un bulk ORDERED, envía como un solo lote únicamente el tramo consecutivo de escrituras que van al mismo shard, y solo después de recibir su respuesta arma el siguiente lote. Una shard key hashed reparte entre varios shards documentos que están contiguos dentro del bulk, así que el lote se corta casi en cada documento. Con n documentos y S shards, el número esperado de viajes de ida y vuelta es 1 + (n − 1)(1 − 1/S). Con 1,000 documentos y 5 shards, son unos 800. Si el bulk no depende del orden, al cambiarlo a BulkMode.UNORDERED (opción del driver ordered(false)) mongos envía a la vez, en una sola pasada, los lotes de cada shard.
Una escritura bulk envía varias operaciones insert·update·replace·delete dentro de una sola petición. En el shell se usa db.collection.bulkWrite(). En Spring Data MongoDB se usa BulkOperations. El cliente envía la petición una vez y recibe la respuesta una vez.
El bulk tiene dos modos. La sección "Ordered vs Unordered Operations" de la documentación oficial de 4.2 los describe así.
| Modo | Forma de ejecución | Si ocurre un error a mitad |
|---|---|---|
| ordered (valor por defecto) | "MongoDB executes the operations serially" | Devuelve el control sin procesar las operaciones restantes |
| unordered | Puede ejecutarse en paralelo (no está garantizado) | Sigue procesando las operaciones restantes |
En un clúster fragmentado, el cliente se conecta a mongos (el router). mongos mira la shard key de cada escritura del bulk y decide a qué shard va (targeting). Después agrupa las escrituras que van al mismo shard y envía a cada shard un sublote (child batch).
Desde el punto de vista del cliente hay una sola petición. Pero entre mongos y los shards hay tantos viajes de ida y vuelta como sublotes. En este artículo, "viaje de ida y vuelta" se refiere a este viaje mongos ↔ shard.
Una shard key hashed divide los chunks (unidades de datos separadas por rangos de valores de la shard key) según el hash del valor del campo. La página "Hashed Sharding" de la documentación oficial de 4.2 lo dice así.
Post-hash, documents with "close" shard key values are unlikely to be on the same chunk or shard
Por ejemplo, pensemos en una colección store_inventory que usa _id: { productId, storeId } como shard key hashed. Es una colección que despliega un producto en tantos documentos como tiendas haya. El documento de la tienda 1 y el de la tienda 2 del mismo producto tienen claves parecidas, pero después del hash es muy probable que estén en shards distintos. Es el precio que paga una shard key hashed a cambio de repartir los datos de forma uniforme.
mongos arma los sublotes en BatchWriteOp::targetBatch de src/mongo/s/write_ops/batch_write_op.cpp. El comentario de las líneas 263-292 describe así la regla de ordered.
// Targeting of ordered batches is a bit more complex - to respect the ordering of the
// batch, we can only send:
// A) a single targeted batch to one shard endpoint
// B) multiple targeted batches, but only containing targeted writes for a single write op
En el caso normal, en el que cada escritura va a un solo shard, se aplica A. En modo ordered solo se envía un lote a un shard cada vez. Si la siguiente escritura va a un shard que no está en el lote actual, isNewBatchRequiredOrdered (líneas 96-106) devuelve true. Entonces mongos corta el lote.
El bucle de ejecución está en batch_write_exec.cpp. mongos arma y envía los lotes dentro de while (!batchOp.isFinished() && !abortBatch) (línea 120). Después espera la respuesta con ars.next(). El comentario de la línea 255 es "Block until a response is available.". Solo arma el siguiente lote cuando ha recibido todas las respuestas. En este artículo llamamos vuelta a cada pasada de este bucle, desde que envía los lotes hasta que recibe todas las respuestas.
Por eso, en modo ordered, cada tramo consecutivo que va al mismo shard es un viaje de ida y vuelta.
El modo unordered hace targeting de todas las escrituras restantes y forma un lote por shard. En una vuelta envía a la vez hasta S sublotes (S es el número de shards). Hay tres condiciones que cortan un lote.
kMaxWriteBatchSize (100,000 escrituras, write_ops.h:62)BSONObjMaxUserSize)isNewBatchRequiredUnordered, líneas 112-125)Un bulk de 1,000 escrituras normalmente no cumple ninguna de estas condiciones, así que termina en una vuelta.
Supongamos que hay tres shards, A·B·C, y que 8 documentos se asignan como se ve abajo. La asignación es un valor de ejemplo.
| Orden | Documento | shard |
|---|---|---|
| 1 | d1 | A |
| 2 | d2 | B |
| 3 | d3 | B |
| 4 | d4 | C |
| 5 | d5 | A |
| 6 | d6 | C |
| 7 | d7 | C |
| 8 | d8 | B |
Los lotes que arma cada modo son los siguientes.
| Vuelta | ORDERED | UNORDERED |
|---|---|---|
| 1 | A: [d1] | A: [d1, d5], B: [d2, d3, d8], C: [d4, d6, d7] (enviados a la vez) |
| 2 | B: [d2, d3] | Ninguno |
| 3 | C: [d4] | Ninguno |
| 4 | A: [d5] | Ninguno |
| 5 | C: [d6, d7] | Ninguno |
| 6 | B: [d8] | Ninguno |
ORDERED da 6 vueltas una tras otra. Cada vuelta espera la respuesta de la anterior. UNORDERED envía 3 sublotes a la vez en 1 vuelta.
El número de viajes de ida y vuelta de ORDERED es el número de cortes del lote más 1. El lote se corta cuando dos documentos vecinos tienen shards distintos. En el ejemplo anterior el shard cambia en 5 de los 7 pares vecinos, así que son 1 + 5 = 6 vueltas.
Suponemos que, con la shard key hashed, los documentos se asignan a los S shards de forma uniforme e independiente. Entonces la probabilidad de que dos documentos vecinos estén en el mismo shard es 1/S, y la de que estén en shards distintos es 1 − 1/S. Con 1,000 documentos hay 999 fronteras entre documentos. Con 5 shards, el lote se corta en 4/5 de ellas, unas 799. Si se suma 1 por el primer lote, salen unos 800.
La fórmula general es la siguiente.
기대 왕복 수 = 1 + (n − 1) × (1 − 1/S)
n = 1,000, S = 5 → 1 + 999 × 0.8 = 800.2
Los valores calculados según el número de shards son los siguientes (n = 1,000).
| Número de shards S | Viajes de ida y vuelta esperados con ORDERED | Vueltas con UNORDERED (peticiones simultáneas) |
|---|---|---|
| 2 | 500.5 | 1 (2) |
| 3 | 667 | 1 (3) |
| 4 | 750.25 | 1 (4) |
| 5 | 800.2 | 1 (5) |
| 6 | 833.5 | 1 (6) |
| 8 | 875.125 | 1 (8) |
Cuantos más shards hay, más se acerca a n el número de viajes de ida y vuelta de ORDERED. Si se añaden shards para aumentar el throughput de escritura, cada bulk ORDERED se parte en trozos todavía más pequeños.
Un write ticket es un pase que limita cuántas escrituras pueden entrar a la vez en el storage engine. Cuando el mongod de un shard ejecuta una escritura, toma el lock global en modo IX (intent exclusive) y en ese momento obtiene un write ticket. Este comportamiento está en lock_state.cpp. La función que toma el lock global, _lockGlobalBegin, llama primero a _acquireTicket. Un sublote solo retiene el ticket mientras se ejecuta en ese shard. No lo retiene en el intervalo en que mongos recibe la respuesta y arma el siguiente lote.
Por eso un bulk ORDERED no retiene el ticket mucho tiempo. A cambio, la vida de un solo bulk abarca todos los shards. Si un shard se vuelve lento, todo ese bulk se detiene. Con 5 shards, un bulk pasa en promedio 160 veces (800 ÷ 5) por cada shard. Si la respuesta de uno de esos shards se vuelve 10 veces más lenta, el bulk espera esa respuesta lenta 160 veces, una tras otra.
Según la ley de Little, el número de trabajos en curso a la vez en un sistema es igual al número de trabajos que llegan por segundo multiplicado por el tiempo que permanece cada trabajo. Todos los números de abajo son valores de ejemplo.
| Situación | Tiempo de un bulk | Bulks en curso a la vez |
|---|---|---|
| ORDERED, normal | 800 × 1ms + 1,000 × 0.2ms = 1.0 s | 20 |
| ORDERED, un shard lento | Unos 2.8 s | Unos 56 |
| UNORDERED, normal | 1ms + 200 × 0.2ms = 41ms | Unos 0.8 |
| UNORDERED, un shard lento | 10ms + 200 × 2ms = 410ms | Unos 8 |
El cálculo de la fila ORDERED·lento es este. Las 160 vueltas y los 200 documentos que van al shard lento consumen 1.6 s + 0.4 s = 2.0 s. Las 640 vueltas y los 800 documentos restantes consumen 0.64 s + 0.16 s = 0.8 s.
UNORDERED también espera al shard más lento. Aun así, paga esa espera una sola vez.
Cuantos más bulks hay en curso a la vez, más hilos y conexiones del cliente quedan ocupados en la misma cantidad. Si el servicio writer que envía los bulks es un consumidor de Kafka, el hilo del consumidor no puede procesar el siguiente mensaje durante ese tiempo.
data class StoreInventory(
@Id val id: InventoryId, // { productId, storeId }, hashed 샤드키
val stock: Int,
val price: Long,
)
fun saveAll(items: List<StoreInventory>) {
val ops = mongoTemplate.bulkOps(BulkOperations.BulkMode.ORDERED, StoreInventory::class.java)
items.forEach { item ->
ops.replaceOne(
Query.query(Criteria.where("_id").`is`(item.id)),
item,
FindAndReplaceOptions.options().upsert(),
)
}
ops.execute()
}
fun saveAll(items: List<StoreInventory>) {
// 같은 _id가 두 번 들어가면 순서가 결과를 바꾼다. 마지막 값만 남긴다.
val deduped = items.associateBy { it.id }.values.toList()
val ops = mongoTemplate.bulkOps(BulkOperations.BulkMode.UNORDERED, StoreInventory::class.java)
deduped.forEach { item ->
ops.replaceOne(
Query.query(Criteria.where("_id").`is`(item.id)),
item,
FindAndReplaceOptions.options().upsert(),
)
}
try {
ops.execute()
} catch (e: BulkOperationException) {
val failed = e.errors.map { deduped[it.index] }
retryOrDeadLetter(failed)
}
}
Si cada replaceOne toca un documento distinto, el orden de ejecución no cambia el resultado. Por eso ambos códigos producen el mismo estado final. La dependencia del orden aparece cuando se mete el mismo documento dos veces en un bulk. El código UNORDERED elimina este caso dejando solo el último valor por _id antes de armar el bulk.
El precio de cambiar a UNORDERED es el manejo de fallos parciales. ORDERED se detiene en el primer error, así que hasta dónde se procesó queda fijado en una sola posición. UNORDERED sigue procesando el resto aunque haya un error. Por eso las escrituras fallidas pueden quedar dispersas por todo el bulk. Cada elemento de BulkOperationException.errors tiene un index que indica qué posición ocupa esa escritura en el bulk. Se seleccionan solo los documentos de las posiciones fallidas y se vuelven a procesar.
Si se usa directamente el driver oficial de Java, se cambia con una línea de opción.
collection.bulkWrite(models, BulkWriteOptions().ordered(false))
_id, fusionarlos antes de armar el bulk.index fallidos o enviarlos a una cola aparte.db.collection.getShardDistribution() y sh.status(). Para ver a qué shard va una clave concreta, se añade explain() a un find() que consulte esa clave. El nombre de shard que aparece en winningPlan.shards del resultado es el shard de esa clave. Desde 4.0 también se puede ver el valor hash con convertShardKeyToHashed().Trampas
- El modo por defecto de
bulkOpsde Spring Data MongoDB y debulkWrite()del shell es ordered. Si no se indicó el modo, se está ejecutando en ORDERED.- UNORDERED también espera al shard más lento. Reduce el impacto del shard lento a una sola vez, pero no lo elimina.
- En UNORDERED, si el mismo documento entra dos veces en un bulk, no se puede prever qué escritura quedará reflejada al final.
- Los viajes de ida y vuelta ocurren entre mongos y los shards. Por eso no se ven si solo se mira el número de peticiones del lado del cliente.
Es un procedimiento que no se ha verificado ejecutándolo. Los resultados de abajo son lo esperado según el código fuente.
mlaunch init --replicaset --nodes 1 --sharded 5 con mlaunch (la herramienta de mtools para lanzar clústeres locales). También se puede montar con docker compose.sh.enableSharding("demo") y luego sh.shardCollection("demo.store_inventory", { _id: "hashed" }). Si se fragmenta con hashed una colección vacía, los chunks se reparten de antemano entre los shards.db.serverStatus().metrics.commands.update.total en el primary de cada shard._id distintos mediante db.store_inventory.bulkWrite(ops, { ordered: true }).{ ordered: false } y repetir los pasos 3~5.El resultado esperado (según el código fuente) es este. La suma de incrementos de ORDERED queda cerca de los valores de la tabla de cálculo de arriba. La suma de incrementos de UNORDERED es igual al número de shards. Es una expectativa que supone que cada sublote llega al shard como un comando update.
- Distribución real de chunks: la fórmula de arriba supone que los documentos se asignan a todos los shards de forma uniforme e independiente. Si los chunks se concentran en un shard, aumenta la probabilidad de que se repita el mismo shard y bajan los viajes de ida y vuelta. No se calculó por separado cuánto bajan según el grado de sesgo de la distribución.
- mongos en 4.4 y posteriores: la base es el código fuente de 4.2 (
r4.2.12) y la documentación de 4.2. No se verificó si las reglas de lotes de mongos en 4.4 y posteriores son las mismas.- El contador del procedimiento de reproducción: no se confirmó ejecutándolo si
metrics.commands.update.totalsube exactamente en 1 por cada sublote, ni si el profiler registra por sublote o por cada escritura (statement).- Ejemplos de código: los ejemplos en Kotlin se escribieron con la API
BulkOperationsde Spring Data MongoDB, pero no se ejecutaron.