シャーディング戦略

シャーディングは異なるデータサブセットを異なるリソースに割り当て、レプリケーションは同じサブセットの複数のコピーを保持します。前者は容量と処理能力を拡張し、後者は冗長性と読み取りサービスを提供します。実際のシステムでは通常、両方が同時に使用されます。

シャーディングキーを選択する前に、最も一般的なクエリ条件、一緒に更新する必要があるデータ、データの増加分布、アクセスの偏りをリストアップしてください。優れたシャーディングキーは、一般的なリクエストを直接位置指定でき、頻繁に共同操作されるデータを一緒に配置しますが、通常、すべてのクエリを同時に最適化することはできません。

キー、論理パーティション、マシンの区別から始める

ビジネスキーはシャーディングルールによって論理パーティションにマッピングされ、ルーティングメタデータが論理パーティションをレプリカグループまたはマシンにマッピングします。1台のマシンが複数のパーティションをホストでき、1つのパーティションに複数のレプリカが存在する可能性があります。

したがって、「パーティションが100個ある」ことは「マシンが100台ある」こととは等しくありません。論理パーティションの数は移行の粒度とメタデータの規模を決定し、マシンの数と仕様は実際のリソースを決定します。この2つの層を分離することで、拡張時に一部のパーティションのみを移行でき、各ビジネスキーの意味を再定義する必要がなくなります。

範囲シャーディング：キーの順序を保持する

範囲シャーディングは、順序付けられたキーの境界に基づいて切り分けを行います。例えば [0,1000)、[1000,2000) などです。id BETWEEN 950 AND 1050 のようなクエリは2つのパーティションにアクセスします。より狭い範囲では1つのパーティションのみがアクセスされ、より広い範囲ではすべてがカバーされる可能性があります。利点は、境界に基づいて関連のないパーティションを切り捨てられることです（すべての範囲クエリが1つのパーティションのみをアクセスすることを保証するわけではありません）。

レコードが増加する時間順に書き込まれる場合、最新のデータは末尾のパーティションに集中し続ける可能性があります。分割はデータ範囲を狭めますが、新しい書き込みを分散させるとは限りません。後続のリクエストはすべて、新しく生成された右端のパーティションに集中する可能性があります。書き込みの並行性、タイムバケット、および他のキー次元と組み合わせて設計する必要があります。

論理的に隣接しているからといって、物理的な順序でのI/Oが行われるわけではありません。ページ構成、LSMのフラッシュ、圧縮、ストレージメディアが実際のI/Oを決定します。シャーディングは、どのリソースがどの範囲を担当するかを決定するだけです。

ハッシュシャーディング：キーの隣接関係を分散させる

ハッシュシャーディングは、まずシャーディングキーに対して決定論的なハッシュを実行し、バケットまたはトークン範囲を選択します。異なるキーは通常、より均一に分散されやすくなりますが、ハッシュは元の順序を壊します。ルーティングが hash(order_id) のみに依存している場合、注文時間に基づくクエリでは、通常、どのパーティションにアクセスすべきかを直接知ることはできません。

キーの数が均一であっても、負荷が均一であるとは限りません。1万件の一般的なキーでも、1つのホット商品キーのリクエスト数より少なくなる可能性があります。ハッシュは、単一のホットキーを自動的に分散させるわけではありません。DynamoDBの公式ガイドラインでも、キーの値の数だけでなく、実際のアクセスアクティビティに基づいてパーティションキーを設計することが強調されています。パーティションキー設計ガイドライン

複合キー：パーティション内で意味のある順序を保持する

注文システムを想定し、user_id をパーティションキーとし、(created_at, order_id) をパーティション内のソートキーとします。これにより、あるユーザーの直近20件の注文を照会する場合、そのユーザーが属するパーティションを特定し、有界範囲スキャンを実行できます。同じタイムスタンプ内では order_id によってレコードを区別します。

これはアクセスパターンの設計であり、特定のデータベースの構文を制限するものではありません。その境界は、大規模テナントまたは大規模ユーザーです。あるテナントのデータが単一の論理パーティションに存在する必要がある場合、そのパーティションの容量またはスループットが先に上限に達する可能性があります。月ごとのバケットやハッシュサフィックスを追加することで対応できますが、その代償として、複数のバケットをクエリして結果をマージする必要があります。

シャーディングを跨いで直近K件を取得：まずローカル候補を取得し、その後グローバルマージ

ユーザーデータが複数のバケットに分割されている場合、時間を降順で並べ替えて直近K件を取得するには、各バケットから条件を満たす上位K件を取得し、その後グローバルな上位K件を選択する必要があります。ローカルでのK+1件目のレコードは、すでに同じバケット内にK件のレコードが存在するため、グローバルな上位K件に入ることはできません。この推論は、各バケットが同じフィルタ条件と同一の全順序を使用していることを前提としています。

以下のタプルは (created_at, order_id) を表し、2つの列とも降順で、order_id はグローバルに一意です。すべてのデータは固定されており、各シャード内のレコードは事前にソートされています。

from heapq import merge
from itertools import islice

shards = [
    [(105, 9), (100, 8), (99, 7)],
    [(104, 6), (100, 5), (98, 4)],
    [(103, 3), (102, 2), (100, 1)],
    [],
]

def page(shards, k, before=None):
    if k <= 0:
        raise ValueError("ページサイズは正でなければなりません")
    candidates = [list(islice((row for row in shard
                               if before is None or row < before), k))
                  for shard in shards]
    return list(islice(merge(*candidates, reverse=True), k))

assert page(shards, 3) == [(105, 9), (104, 6), (103, 3)]
collected, cursor = [], None
while batch := page(shards, 2, cursor):
    collected.extend(batch)
    cursor = batch[-1]
assert collected == sorted([row for shard in shards for row in shard], reverse=True)
assert len(collected) == len(set(collected)) == 9

# 同じタイムスタンプでページを跨ぐ場合、時間を保存するだけでは他の注文が漏れます。
ties = [[(100, 8), (100, 5), (100, 1)]]
first = page(ties, 2)
assert first == [(100, 8), (100, 5)]
assert page(ties, 2, first[-1]) == [(100, 1)]
assert [row for row in ties[0] if row[0] < first[-1][0]] == []

モデルは最大で P×K 件の候補（Pはシャード数）を収集し、その後マージして K 件を返します。実際のクエリでは、タイムアウト、部分的な結果、および各バケットからの応答の非同期性に対するルールを策定する必要があります。Pythonのフィルタリングはメモリ内スキャンですが、データベースの実装では、ソートインデックスを利用してカーソル位置から位置指定し、この関数を効率的なデータベースクエリプランとして使用すべきではありません。

order_id はページ付け順序の最後の区別次元です。created_at のみでソートするか、または時間のみをカーソルに格納する場合、同一ミリ秒の複数の注文がページを跨いで漏れる可能性があります。SQLの LIMIT でも、一意の順序を決定できる ORDER BY が必要です。PostgreSQL LIMIT/OFFSET ドキュメント を参照してください。

複合カーソルでも、ページを跨ぐスナップショットが自動的に提供されるわけではありません。最初のページ以降に挿入され、カーソルの前に並ぶレコードは、後のページには表示されません。元のレコードのソートキーが変更された場合、読み飛ばしや再読み取りが発生する可能性があります。プロダクトで「同一時刻の完全な結果をトラバースする」ことを要求する場合、利用可能なスナップショットまたは読み取りバージョンを固定し、スナップショット識別子、フィルタ条件、およびソールルールをカーソルにバインドする必要があります。最後のIDのみを保存するだけでは不十分です。

各シャードから何件取得すれば、グローバル上位K件が得られるか？

ローカル候補と最終的な返却値を分離することで、クエリファンアウトの実際のコストが見えます。3つのシャードがそれぞれK件を返す場合、合計で3K件の候補が収集されますが、最終的に必要なのはK件だけです。金色の結果は、同じ (time,id) の全順序に基づいて選択されなければなりません。時間のみを比較して同点の項目を任意に捨てることはできません。図中のレコードは固定されており、ページ付け中に新規作成および変更されたレコードについては、本文のスナップショット規約に従う必要があります。

ホットスポットは原因に応じて処理する

ソルティング（Salting）は、論理キーにバケット次元を追加することです。例えば (product_id, bucket_id) のようになります。バケットの選択方法、旧記録の検索方法、およびクエリがすべてのバケットをトラバえるかどうかを定義する必要があります。原子性のある更新が必要な同じレコードを異なる位置にランダムに書き込むと、ビジネスセマンティクスが変更されるため、一般的な拡張テクニックとしては使用できません。

例えば (日付, hash(order_id) % B) とすることで、日付とorder_idを知ることで重複チェックのバケットを計算できます。バケットが書き込みのたびにランダムに選択される場合、位置を別途保存するか、バケットをトラバースする必要があります。いずれの場合も、ある日のすべての注文をクエリするには、通常、すべてのB個のバケットにアクセスする必要があります。AWSの書き込みシャーディングの説明では、ランダムなサフィックスと計算可能なサフィックスのクエリコストを比較しています。Write sharding

クライアントがターゲットパーティションに直接接続することでプロキシオーバーヘッドを削減できますが、ターゲットパーティション自体のストレージまたは計算の上限を高めるわけではありません。

セカンダリインデックス：クエリ条件が変われば、維持する組織形式も増える

主テーブルが user_id でシャーディングされているが、クエリ条件が email であると仮定します。各シャードに独自のローカル email インデックスしかない場合、単一シャードでの検索は高速になりますが、呼び出し元は email がどのシャードにあるかを知ることができないため、依然としてファンアウトが必要になる可能性があります。

別の方法は、email で整理されたグローバルインデックスを維持し、email をレコードの主キーにマッピングすることです。この場合、まずインデックスを検索し、その後主表に戻して取得します。その代償として、1回のビジネス書き込みで2つの組織形式を維持する必要があります。これらがパーティションを跨ぐ場合、クロスパーティショントランザクション、非同期伝播、または他の一貫性プロトコルのいずれを採用するかを明確にする必要があります。

「ローカル」「グローバル」の具体的な定義は製品によって異なり、名前だけでインデックスの更新が原子性であるか、強一貫性読み取りをサポートするかどうかを推測することはできません。非同期インデックスは、一時的に新しいレコードを見逃すか、すでに削除されたレコードを指す可能性があります。アプリケーションは、必要なセマンティクスに従って処理する必要があります。インデックスを通じて「このemailは存在しない」ことを確認してから挿入しても、自動的にグローバル一意制約にはなりません。

「インデックスに存在しない」ことを一意性ロックとして使用できない理由

2つの登録リクエストが同じ email を使用しているが、user_id によって異なるシャードにルーティングされています。各主シャードでの単一挿入が原子性であっても、以下のインターリーブは合法です。

2つの主レコードはすでに同時に存在しており、後続のインデックスが最終的に1つまたは2つのエントリを保持しても、この事実を取り消すことはできません。問題はインデックスの伝播が遅いだけではありません。2回のクエリがともに強一貫性であっても、「存在しないことを確認する」と「作成権限を取得する」が2つの独立したステップに分かれている限り、2回の確認がどちらも最初の作成の前に発生する可能性があります。

1つの方向性は、正規化された email で位置指定された一意の場所で、作成権限を原子性で確保することです。または、カバー制約と主トランザクションを使用します。前者では、プレースホルダーの成功、主レコードの未完成時にどのように回復するかを定義する必要があり、永続的なプレースホルダーが残らないようにする必要があります。後者では、対応するクロスパーティション調整のコストを負担する必要があります。グローバル一意制約所谓のコストは、この共同判断を必要とするキーにあります。すべてのシャードでローカル一意インデックスを個別に構築して組み合わせることで解決できるものではありません。

シャーディングキーレビューのクイックチェック

1. 主要なクエリにシャーディングキーが含まれているか？含まれていない場合、どの程度の数のパーティションにアクセスするか？
2. 単一パーティションの容量、書き込み、ホットスポットが先に枯渇する可能性があるか？
3. 頻繁に一緒に更新されるレコードを共置できるか？パーティションを跨ぐ場合、どのようなトランザクションセマンティクスが必要か？
4. シャーディングキーは変更されるか？変更される場合、それはインプレース更新か、レコードの移動を意味するか？
5. セカンダリインデックス、ページ付け、グローバルソートの追加コストは許容できるか？

ファンアウトは、テールレイテンシと障害処理コストを増幅します。並列クエリの応答は、通常、必要な結果の中で最も遅いもの待ちになります。部分的な結果を許可する場合、その完全性を明確にする必要があります。完全な結果を要求する場合、再試行、タイムアウト、および重複マージを考慮する必要があります。

続きを読む：一貫性ハッシュ は、メンバーが変更されたときにマッピングの再計算をどのように減らすかを説明します。動的再バランス は、パーティションがどのように移行するかを説明します。

参考文献

• DynamoDB：パーティションの動作。
• DynamoDB：パーティションキーの設計。

Keywords: sharding, partition key, range partitioning, hash partitioning, hot key, salting, secondary index