SparkのRDDとDAG実行

RDDとは何か：分散された不変のコレクション

Sparkの計算モデルの中核は RDD（Resilient Distributed Dataset：耐障害性のある分散データセット） です。名前が示す通り、これは3つの性質を持つデータ構造です。分散（クラスタの複数ノードにパーティション分割されて置かれる）、不変（一度作ったら書き換えない。変換すると新しいRDDができる）、そして耐障害性（一部が失われても再計算で復元できる）。

利用者から見るとRDDは「1つの巨大なコレクション」ですが、内部では複数の パーティション（partition） に分かれています。パーティションこそが並列処理と障害復旧の単位で、Sparkは1パーティションにつき1タスクを実行します。HDFS上のファイルを読めば既定でブロック（既定128MB）ごとに1パーティションができ、パーティション数がそのステージの並列度の上限を決めます。

ここで決定的に重要なのが、RDDがデータそのものを保持するとは限らない点です。RDDが本当に持っているのは「どの親RDDから、どの変換で、どう自分を計算するか」という系譜（lineage：リネージ） の情報です。実データは、必要になったときにこの系譜をたどって計算されます。

遅延評価：変換は記録され、アクションで初めて走る

RDDの操作は2種類に厳密に分かれます。この区別が遅延評価の土台です。

変換はすべて遅延（lazy） です。map や filter を呼んでも、その場では1バイトも処理されません。Sparkは「この親RDDにこの変換を適用する」という事実を系譜に書き足すだけで、新しいRDDを表すオブジェクトを返します。実際の計算が始まるのは、collect（結果をドライバへ集める）や count（件数を数える）といったアクションが呼ばれた瞬間です。

rdd  = sc.textFile("logs.txt")        # 変換: まだ何も読まない。ファイルへのポインタ
errs = rdd.filter(_ contains "ERROR") # 変換: filter を系譜に足すだけ
upts = errs.map(parseTimestamp)       # 変換: map を系譜に足すだけ
upts.count()                          # アクション: ここで初めて全体が計算される

なぜ遅延にするのか。アクションが呼ばれるまで全体像が確定しないから、そこでまとめて最適化できるからです。上の例なら、filter と map は連続するナロー依存（後述）なので、Sparkは1行を読むごとに「フィルタ→パース→カウント」を1パスでパイプライン実行でき、中間RDDを丸ごとメモリに実体化せずに済みます。filter の時点で即座に全件を実体化していたら、この融合はできません。

ナロー依存とワイド依存：DAGを切る境界

系譜は、RDDどうしの依存関係のグラフとして表せます。各RDDは1つ以上の親を持ち、この親子関係の向きは循環しないので、全体は DAG（有向非巡回グラフ） になります。この依存には2種類あり、その違いが実行計画のすべてを決めます。

• ナロー依存（narrow dependency）：親RDDの各パーティションが、子RDDの高々1つのパーティションにしか使われない。map、filter、union などが該当。データはパーティションをまたいで動かず、各パーティションは独立に計算できる。
• ワイド依存（wide dependency）：子RDDの1つのパーティションが、親RDDの複数（多くは全部）のパーティションに依存する。groupByKey、reduceByKey、join、repartition などが該当。同じキーを1か所へ集めるため、ノード間でデータを再配置する必要がある。

ナロー依存（map/filter）           ワイド依存（reduceByKey）
  親P0 ── 子P0                       親P0 ─┐
  親P1 ── 子P1                       親P1 ─┼─→ 子P0（key=a を集約）
  親P2 ── 子P2                       親P2 ─┘
  ※各パーティションが独立           ※全親から該当キーを掻き集める

ワイド依存を実現する全ノード間のデータ再配置が シャッフル（shuffle） です。シャッフルは、各パーティションを読んでキーごとに振り分け直し、ネットワーク越しに送り、受け手側で該当キーを集める all-to-all の操作で、Spark全体で最も高コストな処理です。ディスク書き出しとネットワーク転送を伴うため、「どこでシャッフルが起きるか」を見切ることが性能理解の核心になります。

ステージとタスク：DAGスケジューラの分割

アクションが呼ばれると、DAGスケジューラがRDDの系譜（DAG）を実行計画へ変換します。手順は明快です。アクションの対象RDDから親へさかのぼり、ワイド依存（シャッフル）に出会うたびにそこで切る。こうしてDAGは複数の ステージ（stage） に分割されます。

• ステージの内部はナロー依存だけで構成される。だからステージ内の変換の連鎖は、パーティションごとに1本のパイプラインとして融合実行できる（中間RDDを実体化しない）。
• ステージの境界は必ずシャッフル。前段ステージが全パーティションのシャッフル出力を書き終えないと、後段ステージは始められない。ここがステージ間の同期点になる。

ステージにはシャッフル出力を生む中間ステージ（ShuffleMapStage）と、結果をアクションへ返す最終ステージ（ResultStage）の区別があります。各ステージは、その担当パーティションの数だけの タスク（task） に展開されます。

系譜:  textFile → map → filter →[シャッフル]→ reduceByKey → collect
                └─────── Stage 1 ───────┘      └──── Stage 2 ────┘
                (ShuffleMapStage: 出力を書く)   (ResultStage: 結果を返す)

Stage 1 が 3 パーティション → タスク3個（各パーティションを独立処理）
Stage 2 が 2 パーティション → タスク2個（Stage 1 のシャッフル出力を読む）

タスクこそが実際にexecutor（ワーカー上のJVMプロセス）へ送られて動く最小単位です。DAGスケジューラがステージを切り、下位の タスクスケジューラ が各タスクをクラスタの空きスロットへ割り当てます。1タスク＝1パーティションの処理なので、あるステージの並列度はそのパーティション数で頭打ちになり、コア数より多くのタスクは順番待ちになります。

メモリ計算の利点：なぜMapReduceより速いのか

Sparkの速さの本質は、しばしば「インメモリだから」と一言で語られますが、正確には中間結果の扱い方にあります。古典的なMapReduceは、Map→Reduceの各段の間で必ず中間結果をディスク（HDFS）へ書き、次のジョブがそれを読み直します。反復アルゴリズム（機械学習の学習ループ、グラフ処理など）では、この「毎段ディスクへ書いて読む」往復が支配的なコストになります。

Sparkは2つの点でこれを避けます。第一に、ナロー依存の連鎖をメモリ上でパイプライン融合し、ステージ内では中間RDDをディスクにもメモリにも実体化しません。ディスクI/Oが発生するのは原理的にシャッフル境界だけです。第二に、繰り返し使うRDDを明示的にキャッシュできます。

rdd = load(...).map(...).filter(...)   # 変換（まだ計算されない）
rdd.cache()                            # 「実体化したらメモリに残せ」と指示（これも遅延）
rdd.count()                            # 1回目: 実際に計算し、結果をメモリに保持
for i in 1..100:                       # 反復:
    model = train(rdd, model)          #   毎回ディスクを読まず、メモリ上のrddを再利用

cache()（＝ persist(MEMORY_ONLY)）は「このRDDが次に実体化されたとき、各ノードは自分の担当パーティションをメモリに保持し、以降のアクションで再利用せよ」という指示です。注意点として、cache() 自体も遅延で、最初のアクションが走って初めてメモリに載ります。反復のたびに系譜を最初から再計算する無駄が消えるため、「10倍以上速くなることも珍しくない」と公式が言うのはこの再利用が効くケースです。

なお、以上はRDDという低レベルAPIの実行モデルです。実務で多いDataFrame/Dataset APIは、この上にCatalystオプティマイザとTungsten実行エンジンを重ね、宣言的なクエリを解析して同種のステージ／タスクへ落とし込みます。列指向のメモリ表現や述語プッシュダウンなどの最適化が加わりますが、「遅延評価→DAG→シャッフル境界でステージ分割→パーティション単位のタスク」という骨格は共通です。したがってRDDとDAG実行を理解することが、Sparkのあらゆるレイヤを読む土台になります。分散処理一般の考え方は /devops/、機械学習ワークロードとの接続は /ai/ も参照してください。

まとめ

• RDD は分割された不変の分散コレクションで、実データではなく系譜（lineage） を持つ。系譜が再計算による耐障害性の源になる。
• 操作は変換（遅延） とアクション（即時起動） に分かれる。変換は系譜へ記録されるだけで、アクションが呼ばれて初めて全体が計算される（遅延評価）。
• 依存にはナロー依存（親パーティションが子の高々1つに使われる／独立計算可）とワイド依存（子が複数の親に依存／シャッフルが必要）があり、後者が最もコストの高い all-to-all のデータ再配置を生む。
• DAGスケジューラは系譜をワイド依存＝シャッフルで切ってステージに分割する。ステージ内はナロー依存のパイプライン、ステージ境界はシャッフルという同期点。各ステージは1パーティション＝1タスクに展開される。
• 速さの本質は「インメモリ」よりも、中間結果をディスクに書かずナロー依存を融合実行し、再利用するRDDだけを明示キャッシュする点。ディスクI/Oは原理的にシャッフル境界に集中する。
• DataFrame/DatasetはこのモデルにCatalyst／Tungstenの最適化を重ねるが、遅延→DAG→ステージ→タスクの骨格は共通。