MongoDB シャードクラスタの書き込み経路シリーズの第1/6回である。
この記事が答える問い: シャーディングされたコレクションに bulk 書き込み1,000件を ORDERED で送ると、mongos と shard の間で往復は何回起きるのか？ なぜ shard が1つ遅くなるだけで bulk 全体が遅くなるのか？ UNORDERED に変えると何を得て、何を代償として払うのか？
基準バージョン: MongoDB 4.2（ソースr4.2.12タグ、4.2 公式ドキュメント）。4.2 は 2023-04-30 にサポートが終了した（EOL）。4.4 以降のバージョンの mongos の動作は確認していない（2026-11 時点）。
ORDERED bulk を受け取った mongos は、同じ shard へ向かう連続区間だけを1つのバッチとして送り、その応答を受け取ってからようやく次のバッチを作る。hashed シャードキーは bulk 内で隣り合う文書を複数の shard に散らすので、バッチがほぼ文書ごとに区切られる。文書 n件、shard S個のとき、期待往復回数は 1 + (n − 1)(1 − 1/S) である。1,000件を shard 5個に送ると約800回になる。順序に依存しない bulk は BulkMode.UNORDERED（ドライバーオプション ordered(false)）に変えれば、mongos が shard ごとのバッチを一度に同時に送る。
bulk 書き込みは、insert・update・replace・delete の複数件を1つのリクエストに載せて送る方法である。シェルでは db.collection.bulkWrite() を使う。Spring Data MongoDB では BulkOperations を使う。クライアントはリクエストを1回送り、応答を1回受け取る。
bulk には2つのモードがある。4.2 公式ドキュメントの "Ordered vs Unordered Operations" 節の説明は次のとおりである。
| モード | 実行方式 | 途中でエラーが起きると |
|---|---|---|
| ordered（デフォルト） | "MongoDB executes the operations serially" | 残りの操作を処理せずに返す |
| unordered | 並列に実行できる（保証はしない） | 残りの操作を処理し続ける |
シャードクラスタでは、クライアントは mongos（ルーター）に接続する。mongos は bulk に含まれる書き込み1件ごとにシャードキーを見て、どの shard へ送るかを決める（targeting）。そのうえで同じ shard へ向かう書き込みを集め、shard ごとに子バッチ（child batch）を作って送る。
クライアントから見ればリクエストは1回である。しかし mongos と shard の間では、子バッチの数だけ往復が発生する。この記事でいう往復は、この mongos ↔ shard の往復である。
hashed シャードキーは、フィールド値をハッシュした値でチャンク（シャードキー値の範囲で分けたデータ単位）を分ける。4.2 公式ドキュメント "Hashed Sharding" はこう書いている。
Post-hash, documents with "close" shard key values are unlikely to be on the same chunk or shard
たとえば _id: { productId, storeId } を hashed シャードキーとして使う store_inventory コレクションを考える。1つの商品を店舗の数だけ文書に展開したコレクションである。同じ商品の店舗1番の文書と店舗2番の文書はキーが似ているが、ハッシュ後には別々の shard にある可能性が高い。hashed シャードキーがデータを均等に分散させる代わりに払うコストである。
mongos は src/mongo/s/write_ops/batch_write_op.cpp の BatchWriteOp::targetBatch で子バッチを作る。263-292行のコメントは 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
1件の書き込みが1つの shard にだけ向かう通常のケースでは A が適用される。ordered モードでは、一度に1つの shard にだけバッチを送る。次の書き込みが現在のバッチにない shard へ向かうと、isNewBatchRequiredOrdered（96-106行）が true を返す。すると mongos はバッチを区切る。
実行ループは batch_write_exec.cpp にある。mongos は while (!batchOp.isFinished() && !abortBatch)（120行）の中でバッチを作って送る。そのあと ars.next() で応答を待つ。255行のコメントは "Block until a response is available." である。応答をすべて受け取ってから次のバッチを作る。この記事では、このループがバッチを送って応答をすべて受け取るまでの1回分を「周回」と呼ぶ。
そのため ordered モードでは、同じ shard へ向かう連続区間1つが往復1回になる。
unordered モードは、残りの書き込みをすべて targeting して shard ごとに1つずつまとめる。1周回で最大 S個（S は shard 数）の子バッチを同時に送る。バッチを区切る条件は3つある。
kMaxWriteBatchSize（100,000件、write_ops.h:62）に達したら区切るBSONObjMaxUserSize）を超えたら区切るisNewBatchRequiredUnordered、112-125行）1,000件の bulk は通常これらの条件に当たらないので、1周回で終わる。
shard が A・B・C の3つで、文書8件が下のように割り当てられたとする。割り当ては例示値である。
| 順序 | 文書 | 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 |
2つのモードが作るバッチは次のとおりである。
| 周回 | ORDERED | UNORDERED |
|---|---|---|
| 1 | A: [d1] | A: [d1, d5], B: [d2, d3, d8], C: [d4, d6, d7]（同時に送る） |
| 2 | B: [d2, d3] | なし |
| 3 | C: [d4] | なし |
| 4 | A: [d5] | なし |
| 5 | C: [d6, d7] | なし |
| 6 | B: [d8] | なし |
ORDERED は6周回を順に回る。各周回は前の周回の応答を待つ。UNORDERED は1周回で子バッチ3つを同時に送る。
ORDERED の往復回数は、バッチが区切られた回数に1を足した値である。バッチは隣り合う2つの文書の shard が異なるときに区切られる。上の例では、隣り合うペア7か所のうち5か所で shard が変わったので、1 + 5 = 6周回である。
hashed シャードキーによって文書が S個の shard に均等に、互いに独立して割り当てられると仮定する。すると隣り合う2つの文書が同じ shard にある確率は 1/S で、異なる確率は 1 − 1/S である。1,000件なら文書間の境界は999か所ある。shard が5個なら、そのうち 4/5 にあたる約799か所でバッチが区切られる。これに最初のバッチ1回を足すと約800回になる。
一般式は次のとおりである。
기대 왕복 수 = 1 + (n − 1) × (1 − 1/S)
n = 1,000, S = 5 → 1 + 999 × 0.8 = 800.2
shard 数ごとの計算値は次のとおりである（n = 1,000）。
| shard 数 S | ORDERED 期待往復回数 | UNORDERED 周回数（同時リクエスト数） |
|---|---|---|
| 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) |
shard を増やすほど、ORDERED の往復回数は n に近づく。書き込みスループットを上げるために shard を追加すると、ORDERED bulk 1つはかえって細かく分割される。
write ticket は、同時にストレージエンジンに入れる書き込みの数を制限する入場券である。shard の mongod は書き込みを実行するときにグローバルロックを IX（intent exclusive）モードで取り、このときに write ticket を受け取る。この動作は lock_state.cpp にある。グローバルロックを取る関数 _lockGlobalBegin が先に _acquireTicket を呼ぶ。子バッチはその shard で実行されている間だけ ticket を持つ。mongos が応答を受け取って次のバッチを作る間は持たない。
そのため ORDERED bulk が ticket を長く持つことはない。その代わり bulk 1つの寿命がすべての shard にまたがる。shard が1つ遅くなると、その bulk 全体が止まる。shard 5個なら bulk 1つが各 shard を平均160回（800 ÷ 5）経由する。そのうち1つの shard の応答が10倍遅くなると、bulk はその遅い応答を160回順番に待つ。
リトルの法則によれば、システム内に同時に存在する作業の数は、1秒あたりに到着する作業の数に、作業1つが滞在する時間を掛けた値に等しい。下の数字はすべて例示値である。
| 状況 | bulk 1つの時間 | 同時に存在する bulk 数 |
|---|---|---|
| ORDERED、正常 | 800 × 1ms + 1,000 × 0.2ms = 1.0秒 | 20 |
| ORDERED、shard 1つが遅い | 約2.8秒 | 約56 |
| UNORDERED、正常 | 1ms + 200 × 0.2ms = 41ms | 約0.8 |
| UNORDERED、shard 1つが遅い | 10ms + 200 × 2ms = 410ms | 約8 |
ORDERED・遅い行の計算はこうである。遅い shard へ向かう160周回と文書200件が 1.6秒 + 0.4秒 = 2.0秒を使う。残りの640周回と文書800件が 0.64秒 + 0.16秒 = 0.8秒を使う。
UNORDERED も最も遅い shard を待つ。それでもその待ちを1回だけで済ませる。
同時に存在する bulk が多いほど、クライアントのスレッドとコネクションもその数だけ拘束される。bulk を送る writer サービスが Kafka コンシューマーなら、コンシューマースレッドはその間、次のメッセージを処理できない。
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)
}
}
各 replaceOne が互いに異なる文書を操作するなら、実行順序は結果を変えない。そのため2つのコードは同じ最終状態を作る。順序依存は、1つの bulk に同じ文書を2回入れたときに生じる。UNORDERED のコードは bulk を作る前に _id ごとに最後の値だけを残して、このケースをなくしている。
UNORDERED に変える代償は部分失敗の処理である。ORDERED は最初のエラーで止まるので、どこまで処理されたかが1つの位置で決まる。UNORDERED はエラーが起きても残りを処理し続ける。そのため失敗した書き込みが bulk のあちこちに散らばりうる。BulkOperationException.errors の各項目には、その書き込みが bulk の何番目かを表す index がある。失敗した位置の文書だけを選んで再処理する。
公式 Java ドライバーを直接使う場合は、オプション1行で変えられる。
collection.bulkWrite(models, BulkWriteOptions().ordered(false))
_id が重複しうるなら、bulk を作る前にまとめる。index だけを再試行するか、別キューに送る。db.collection.getShardDistribution() と sh.status() で見る。特定のキーが向かう shard を見るには、そのキーで検索する find() に explain() を付ける。結果の winningPlan.shards に出てくる shard 名がそのキーの shard である。4.0 からは convertShardKeyToHashed() でハッシュ値も見られる。落とし穴
- Spring Data MongoDB
bulkOpsとシェルbulkWrite()のデフォルトモードは ordered である。モードを指定していなければ ORDERED で動いている。- UNORDERED も最も遅い shard を待つ。遅い shard の影響を1回に減らすだけで、なくすわけではない。
- UNORDERED で1つの bulk に同じ文書が2回入ると、どちらの書き込みが最後に反映されるかは期待できない。
- 往復は mongos と shard の間で起きる。そのためクライアント側のリクエスト数だけを見ていると見えない。
実行で検証していない手順である。下の結果はソースに基づく予想である。
mlaunch init --replicaset --nodes 1 --sharded 5 を実行する。docker compose で構成してもよい。sh.enableSharding("demo") の次に sh.shardCollection("demo.store_inventory", { _id: "hashed" }) を実行する。空のコレクションを hashed でシャーディングすると、チャンクが shard ごとにあらかじめ分割される。db.serverStatus().metrics.commands.update.total の値を記録する。_id が互いに異なる replace（upsert）1,000件を db.store_inventory.bulkWrite(ops, { ordered: true }) で送る。{ ordered: false } で送り、3〜5の手順を繰り返す。予想（ソースに基づく）結果はこうである。ORDERED の増分の合計は、上の計算表の値の近くに収まる。UNORDERED の増分の合計は shard 数と同じになる。子バッチ1つが update コマンド1つとして shard に届くと仮定した予想である。
- 実際のチャンク分布: 上の式は、文書がすべての shard に均等に、独立して割り当てられると仮定している。チャンクが1つの shard に偏ると、同じ shard が続く確率が大きくなり、往復回数は減る。分布がどの程度偏るとどれだけ減るかは、別途計算していない。
- 4.4 以降のバージョンの mongos: 根拠は 4.2 のソース（
r4.2.12）と 4.2 のドキュメントである。4.4 以降の mongos のバッチ規則が同じかは確認していない。- 再現手順のカウンター:
metrics.commands.update.totalが子バッチ1つごとに正確に1ずつ増えるか、プロファイラーが子バッチ単位で記録されるか書き込み1件（statement）単位で記録されるかは、実行で確認していない。- コード例: Kotlin の例は Spring Data MongoDB
BulkOperationsAPI で書いたが、実行していない。