# シャッフルと分散結合 ## 定義 シャッフルとは、入力と出力の両方がシャーディングされた状態でペタバイト規模のデータセットをソートする分散アルゴリズムであり、バッチ処理におけるジョインとグループ化の基盤になる。カードのシャッフルとは異なり、乱雑にする操作ではなく、結果として整列された(ソート済みの)順序を生成する点に注意が必要である。MapReduce、Spark、Flink、Daft、Dataflow、BigQueryはいずれも独自のスケーラブルなシャッフルアルゴリズムを実装する。(Source: [[@2026__OReilly__Designing Data-Intensive Applications 2E - Chapter 11 Batch Processing]] "Shuffling Data") ## MapReduceにおけるシャッフルの機構 入力は複数のシャード(m1, m2, m3...)に分割され、フレームワークはシャードごとに個別のマップタスクを起動する。マッパー出力のキー・バリューペアは、キーのハッシュに基づいて宛先のリデューサが決定され、各マッパーはリデューサごとに個別のローカルファイル(例: マッパー1がリデューサ2向けに書くファイル`m1, r2`)を作成する。マッパーはファイルを書きながらキー・バリューペアをソート(メモリ上のソート構造→セグメントファイル書き出し→段階的マージという、[[外部マージソート]]と同じ技法)する。各リデューサは全マッパーから自分宛のファイルを収集し、ソート順を保ったままマージソート形式で統合する。同一キーのペアはマッパー間をまたいでも隣接するようになり、リデューサ関数がキーごとに1回呼び出される。(Source: [[@2026__OReilly__Designing Data-Intensive Applications 2E - Chapter 11 Batch Processing]] "Shuffling Data") ## ジョインとグループ化 ユーザー活動イベント(ファクトテーブル)とユーザーデータベース(ディメンションテーブル)をuser_idでジョインする例では、user_idをキーとするシャッフルにより、同一ユーザーに関する全レコードが同一リデューサへ集まる。ジョブはレコードの並び順を制御し、ユーザーデータベースのレコード(date_of_birth)を先に、続けて該当ユーザーの活動イベントを時刻順に受け取るよう調整できる(**セカンダリソート**)。この場合リデューサは一度に1ユーザー分のレコードだけをメモリに保持すればよく、ネットワークへのリクエストも不要になる。この手法は**ソートマージジョイン**と呼ばれる。次段のジョブでURLをキーにシャッフルし直せばグループ化・集約(年齢層別ページビュー数のカウント等)も同様に実現できる。(Source: [[@2026__OReilly__Designing Data-Intensive Applications 2E - Chapter 11 Batch Processing]] "Joins and Grouping") 現代のデータフローエンジンやクラウドデータウェアハウスは、MapReduceのようにmap/reduceの間で常にシャッフルするのではなく、シャッフルをメモリ上に保持したり外部ソートサービスへ書き出したりして高速化・耐障害性を高めている(例: BigQueryの外部ソートサービス)。(Source: [[@2026__OReilly__Designing Data-Intensive Applications 2E - Chapter 11 Batch Processing]] "Shuffling Data") ## 横断的知見 - **シャッフルのマッパー側ソート機構は、単一マシン向けに設計された外部マージソートの分散版である**: [[外部マージソート]]が扱うDuckDBの二相マージソート(スレッドローカルソート→T-wayマージ)は単一マシン内のマルチコア並列を対象とするが、DDIA第11章が説明するMapReduceシャッフルのマッパー側処理(「メモリ上のソート済み構造に蓄積→セグメントファイルとして書き出し→段階的にマージ」)は本質的に同じ外部マージソートのアルゴリズムを分散環境(各マッパーがローカルディスク上で)に適用したものである。ハッシュによるリデューサへの事前分配(パーティショニング)を除けば、単一マシンのソートアルゴリズムが分散シャッフルの基礎部品としてそのまま再利用されていることが分かる。(Source: [[@2026__OReilly__Designing Data-Intensive Applications 2E - Chapter 11 Batch Processing]], [[外部マージソート]]) - **ハッシュベースの分散(シャーディング)先決定という同じ発想が、シャッフルによるリデューサ振り分けとDuckDBのハッシュベースグループ集約の両方で使われる**: MapReduceのシャッフルは`hash(key)`でどのリデューサファイルへ書き出すかを決めるのに対し、[[ハッシュベースグループ集約]]が扱うDuckDBの外部グループ集約Phase 1も`h(grp) = k = (salt, p, i)`というハッシュでパーティション番号`p`を決定し、スレッドローカルなハッシュテーブルへの事前集約とパーティション振り分けを同時に行う。前者はマシン間(ネットワーク越し)の分散、後者はスレッド間(単一プロセス内)の分散という規模の違いはあるが、「キーのハッシュで下流の処理単位を事前に決定してから局所的に集約・ソートする」という設計パターンは共通する。(Source: [[@2026__OReilly__Designing Data-Intensive Applications 2E - Chapter 11 Batch Processing]], [[ハッシュベースグループ集約]]) ## 未解決の問い - BigQueryの「外部ソートサービス」はシャッフルデータをどのように複製し、どの程度の耐障害性向上を実現しているか(本章は概要のみで実装詳細は示さない)。 - MapReduceのハッシュベースのリデューサ振り分けは、[[データパーティショニング]]が扱う一貫性ハッシュ法のような動的リバランシング手法とどう関係するか。ジョブ実行中にリデューサ数が変わらない前提であれば静的ハッシュで十分だが、この前提はどこまで一般的か。 - セカンダリソートによるソートマージジョインは、ジョインするテーブルの一方が極端に大きい(スキューがある)場合にどう性能劣化するか。ブロードキャストジョインのような代替アルゴリズムとの使い分け基準は本章に記載がなく、他ソースでの補完が必要。 ## 関連 - ソース: [[@2026__OReilly__Designing Data-Intensive Applications 2E - Chapter 11 Batch Processing]] - 概念: [[MapReduce]](シャッフルを実行するフレームワーク) / [[データフローエンジン]](シャッフルの高速化実装) / [[外部マージソート]](マッパー側ソートの基礎アルゴリズム) / [[ハッシュベースグループ集約]](同型のハッシュ振り分け設計) / [[データパーティショニング]] - 書籍: [[Designing Data-Intensive Applications 2nd Edition]] ## 出典 - [[@2026__OReilly__Designing Data-Intensive Applications 2E - Chapter 11 Batch Processing]]("Shuffling Data"・"Joins and Grouping" 節)