分散結合戦略（broadcast・shuffle・sort-merge）

なぜ分散環境では「結合戦略」が問題になるのか

1台のDBなら、結合の主な選択肢はネステッドループ・ハッシュ結合・ソートマージ結合で、勝負はインデックスとメモリの範囲に収まります。ところがデータが数百ノードに分割された分析基盤では、話が一段変わります。同じ結合キーを持つ行が、別々のノードに散らばっているからです。ノードN1にある注文行の相手（顧客行）が、ノードN7に載っているかもしれない。この状態で結合を成立させるには、まず照合すべき行どうしを同じノードへ集める必要があり、そのデータ移動こそが分散結合のコストの本体です。

つまり分散結合の設計は、結合アルゴリズムそのものより、**「どちらの表を、どうやってノード間で動かすか」**の選択です。この移動戦略が主に三つ——ブロードキャスト・シャッフル・ソートマージ——であり、どれを選ぶかで実行時間が桁で変わります。移動を伴う再分配は、/data-engineering/mapreduce-shuffle/ で見た all-to-all のシャッフルと同じ性質を持ち、分散処理で最も高コストな操作です。

ブロードキャストハッシュ結合：小さい方を配って片側のシャッフルを消す

一方の表が十分に小さい（マスタ表・ディメンション表など）なら、最も速いのはブロードキャストハッシュ結合（broadcast hash join）です。手順はこうです。まず小さい表を全ノードへまるごと複製して配り、各ノードのメモリ上にその表のハッシュ表を作る。次に大きい表は動かさず、各ノードが自分の持ち分の行を1件ずつハッシュ表に問い合わせて照合する。

小表S（数MB）を全ノードへ複製、各ノードでハッシュ表を構築
大表L はそのまま（再分配なし）
  ノードN1: L の担当行 → メモリ上の S ハッシュ表を lookup → 出力
  ノードN2: L の担当行 → 同じ S ハッシュ表を lookup → 出力
  …
→ 大表 L のネットワーク移動はゼロ

肝は大表を一切シャッフルしない点です。全対全の再分配が消え、代わりに小表を配る片方向のブロードキャスト（合計で「小表サイズ×ノード数」程度）だけで済みます。大表が巨大でも小表が数MBなら、これが圧倒的に有利になります。制約は明確で、複製した小表のハッシュ表が各ノードのメモリに載りきる必要があること。だからエンジンは「小さい方のサイズが閾値以下か」を選択の第一基準にします（Sparkの spark.sql.autoBroadcastJoinThreshold は既定10MB）。

シャッフルハッシュ結合：両表を結合キーで再分配する

両方の表が大きく、どちらもブロードキャストできないときの基本形がシャッフルハッシュ結合（shuffle hash join）、Hive系ではリパーティション結合とも呼ばれます。考え方は単純で、両表を結合キーのハッシュで同じ規則に従って再分配し、同じキーの行が必ず同じノードへ集まるようにしてから、各ノード内でローカルにハッシュ結合します。

両表とも partition = hash(結合キー) mod N で再分配（N はノード/パーティション数）
  key=a の行 → L からも R からも同じノードNx へ
  key=b の行 → L からも R からも同じノードNy へ
各ノード: 自分に集まった L 断片から一方のハッシュ表を作り、
          もう一方の断片を lookup して結合

このとき決定的なのは、両表に同一のハッシュ関数と同一の分割数を適用すること。そうして初めて「hash(a) が指すノード」が両表で一致し、キー a の行が漏れなく同じノードに揃います。ローカル結合フェーズでは、集まった断片のうち小さい側でハッシュ表を作り、大きい側でプローブするのが定石です。コストの本体は、/data-engineering/spark-rdd-dag/ でいうワイド依存＝シャッフル、つまり両表ぶんの全対全転送にあります。ブロードキャストが片側だけを動かすのに対し、シャッフルは両側を動かすぶん高くつきます。

ソートマージ結合：ソートしてから2本のストリームを突き合わせる

シャッフルハッシュ結合には弱点があります。ローカル結合フェーズで片方の断片からハッシュ表を作るため、その断片がノードのメモリに載らないと破綻する（スピルしても効率が落ちる）点です。結合キーが偏って一部ノードに巨大な断片が来たり、そもそも両表とも巨大だったりすると、これが起きます。そこで大規模結合の既定として広く使われるのが**ソートマージ結合（sort-merge join）**です。

手順は三段です。(1) 両表を結合キーでシャッフル再分配して同じキーを同じノードへ集める（ここまではシャッフルハッシュと同じ）。(2) 各ノードで、集まった両表の断片をそれぞれ結合キーでソートする。(3) ソート済みの2本のストリームを、2つのポインタで同時に前進させながら突き合わせる（マージ）。

ソート後（両表とも結合キー昇順）:
  L: a a b d e …      ポインタ pL, pR を進めながら比較
  R: a b b c e …
   pL=a, pR=a → 一致（キーa の全組を出力）
   一致キーを出し切ったら、小さい方のポインタを前進
   pL の値 < pR の値 なら pL を進め、逆なら pR を進める

ソートマージの利点は、照合が両ストリームを1回ずつ順に舐めるだけで完結し、巨大なハッシュ表をメモリに常駐させずに済むこと。ソート自体はメモリを超えれば外部マージソートでディスクへスピルでき、大規模データやスキューに対して安定して動きます。さらに、結合キーでソートされた出力もソート済みになるため、後続の集約やウィンドウ関数、次の結合がそのソート順を再利用でき、追加のシャッフルを省けることがあります。Spark SQLがハッシュ結合よりソートマージを大表どうしの既定に据えているのは、この安定性とスピル耐性のためです。

結合キーの偏り（skew）：分散結合を殺す最大要因

戦略を正しく選んでも、**結合キーの偏り（data skew）**が全体を台無しにします。シャッフル系はどちらも hash(結合キー) mod N で行き先を決めるため、特定キーに行が極端に集中する（ホットキー）と、そのキーを担当する1ノードに両表の該当行が山と集まります。他ノードが早々に終わっても、その1ノードだけが巨大な結合を1台で背負い、ジョブ全体がそれを待つ——/data-engineering/mapreduce-shuffle/ のホットキー問題が、結合では両表ぶん効いてさらに悪化します。

どれが選ばれるか：コストベースの判断基準

現代のエンジン（Spark SQL、各種MPP、Presto/Trino）は、これらを**コストベースオプティマイザ（CBO）**が統計を見て選びます。判断の骨子は次の順です。

1. 片方がブロードキャスト閾値以下か。小さい表があり、その推定サイズが閾値（例10MB）以下ならブロードキャストハッシュ結合。全対全を回避できるので最優先。
2. 両表が結合キーで事前配置済みか。両表が同じキー・同じ数のバケット（/data-engineering/partitioning-bucketing/）や、MPPの同じ分散キー（/data-engineering/mpp-data-warehouse/ のコロケーテッド結合）に配置されていれば、シャッフル自体を省いてノード内で直接結合できる。設計で稼ぐ最大の勝ち筋。
3. 上のどちらでもなければシャッフル系。両表を再分配したうえで、ハッシュ表が載るならシャッフルハッシュ、超大規模・スキュー耐性・出力ソート順の再利用を狙うならソートマージ。多くのエンジンが安定性からソートマージを大表どうしの既定にする。

まとめ

• 分散結合の本質は照合すべき行を同じノードへ集める再分配で、コストの大半はこのデータ移動にある。アルゴリズムは三つの再分配戦略に集約される。
• ブロードキャストハッシュ結合は小表を全ノードへ複製して大表を動かさないため、片側のブロードキャストだけで済む。前提は複製した小表のハッシュ表が各ノードのメモリに載ること（閾値判定）。
• シャッフルハッシュ結合は両表を hash(結合キー) mod N で再分配し、同じキーを同じノードへ集めてローカルにハッシュ結合する。両表ぶんの全対全転送が本体コスト。
• ソートマージ結合は再分配後に両表をキーでソートし、2本のソート済みストリームをマージ走査する。常駐ハッシュ表が要らずスピルに強く、出力もソート済みなので、大表どうしの既定として広く使われる。
• 戦略はコストベースで選ばれる。閾値以下ならブロードキャスト、同一バケット／同一分散キーに事前配置していればシャッフルを省略、他はシャッフル系（安定性からソートマージが既定になりやすい）。
• 最大の敵は結合キーの偏り。ホットキー・NULLキー・直積爆発が単一ノードを潰し、hash mod N の均等分割は一様分布の仮定に立つため現実データで崩れる。ノード増ではなくソルトやNULL除外など偏り前提の設計で解く。