🔖 キーワード索引

「Spark」を取り巻く中核キーワード群です。 検索やインデックス作成で参照する際の手がかりにしてください。 各キーワードは関連する概念・手法・道具立てを含み、 文献検索や学習計画の起点になります。

🎨 直感で掴む

1 万冊の蔵書を読みたいとき：

• 1 人で読む ＝ pandas（数十年かかる）
• 100 人で分担 ＝ Spark（数ヶ月で読み終わる）

Spark の役割は「誰がどの本を読むか」 「読んだ結果をどう集めるか」を取り仕切ること。 ユーザーは df.groupBy(...).count() と書くだけで、 裏では 100 台のマシンが並列で動きます。

🎨 概念図で押さえる

Spark の3つの中核概念 (クラスタ構成 / RDD 系統グラフ / Stage 実行) を inline SVG で視覚化する。 これらの図は本文の言葉だけでは把握しづらい「分散処理の流れ」「変換と行動の境界」「DAG スケジューラの仕事」を一枚に集約したものである。

図 1: Spark クラスタ構成 — Driver と Executor の関係

Driver プログラムが SparkContext を作成し、 Cluster Manager (YARN / Kubernetes / Standalone) を介して複数の Worker 上に Executor を立ち上げる。 タスクは Driver → Executor へ送られ、 結果は逆方向に集約される。

Driver は「司令塔」、 Cluster Manager は「リソース調整役」、 Executor は「実働部隊」。 タスクが何百あっても、 各 Executor のスレッド数だけ並列に処理される。

図 2: RDD 系統グラフ (Lineage) — 変換 (transformation) と 行動 (action)

RDD は「親 RDD と変換ルールの記録」。 map / filter / flatMap は遅延評価で、 collect / count / save 等の action が来た時点で初めて DAG 全体が実行される。 これにより最適化と耐障害性 (再計算可能) を両立する。

reduceByKey など widedependency (shuffle) を挟む変換が Stage の境界となる。 系統が分かれば、 Executor が落ちても親 RDD から再構築できる。

図 3: DAG スケジューラ — Job → Stage → Task の分解

action が呼ばれると Job が作られ、 shuffle 境界で Stage に分割、 各 Stage は partition 数だけ Task に分かれて Executor で並列実行される。 この階層が Spark UI の表示構造そのものである。

Stage 1 (例: map+filter) は shuffle なしで完結し、 Stage 2 (例: reduceByKey 後段) は前段の shuffle ファイルを読み込んでから走る。 partition 数 = Task 数なので、 partition を増やせば並列度が上がるが、 タスク起動コストとのバランスが必要。

この3枚を脳内に置けば、 Spark UI の「Jobs / Stages / Tasks」タブを開いたときに何を見ているかが瞬時にわかる。 さらに学習を進めるなら Hadoop・分散処理・クラウドサービス ページへ。

📝 理解度チェック（練習問題）

Apache Spark の中核を本当に掴めたかを 6 問で確認する。 すべて 1 分以内で答えられる粒度。 「自分で」答えを口に出して言える状態を目指したい。

Q1. Spark が Hadoop MapReduce より高速な根本理由は

Q2. Transformation と Action の違い

Q3. shuffle が発生する代表的な操作 3 つ

Q4. RDD と DataFrame、 どちらを使うべきか

Q5. partition 数を増やすと必ず速くなる？

Q6. SSDSE-B-2026 のような 47 行のデータで Spark を使うべきか

これらの問題に詰まった項目があれば、 「DAG 構造」「shuffle の仕組み」「Catalyst Optimizer」の該当セクションへ戻ろう。 Spark は「いつ使うべきか」の判断こそが熟練度のバロメータである。

🎮 触って理解する — 遅延評価とパーティション並列

Spark の心臓部である 遅延評価（lazy evaluation） と パーティション並列 を、 12 個の数値からなる 架空データ（説明用・決定的）で体感します。 map/filter などの transformation を積んでもすぐには計算されず、 計算グラフ（DAG）が伸びるだけ。 count/collect などの action を呼んだ瞬間に一斉実行される — この境界を目で見て掴んでください。 Hadoop のワードカウント（Map→Shuffle→Reduce）とは別の側面に絞っています。

🔁 反復処理：Spark（インメモリ）vs Hadoop（毎回ディスク）

機械学習やグラフ計算のように 同じデータを何度も舐める反復処理こそ Spark の主戦場。 Hadoop MapReduce は反復のたびに中間結果を HDFS（ディスク）へ書き戻すのに対し、 Spark は cache() でメモリ保持し、 読み書きを省く。 反復回数を変えて I/O 差を見てください。

💡 3 つの視点で深掘り

📐 数式または定義

spark の定義や代表的な数式を以下に示す。 数式の各記号の意味は次節で言葉に翻訳する。

spark は文脈に応じて複数の定式化があるが、 教育目的では最も基本的な形を抑えることが重要。 具体的な値での計算例は後続セクションを参照。

$$\text{spark}: f(\mathbf{X}, \boldsymbol{\theta}) \to y$$

記号の対応はこうです。 Spark の実行単位は上位から Job → Stage → Task の 3 層で、 $\text{Stage}$ の切れ目はシャッフル（ノード間のデータ移動）が必要になる箇所に対応します。 $\text{Task}_{1..k}$ の $k$ はパーティション数で、 1 パーティション＝1 タスク＝1 コアで処理される単位です。 スループットの式 $\text{Data Size} / (\text{Cluster Cores} \times \text{Time per partition})$ が示すのは、 コアを増やしても、パーティション数がコア数より少なければ余ったコアは遊ぶということ。 経験則としてパーティション数はコア数の 2〜4 倍にします。 SSDSE の 564 行のような小さなデータでは、 パーティションを分ける費用のほうが計算より高くつきます。

🔬 記号・要素の読み解き

🔬 数式を言葉で読み解く（詳細版）

「Apache Spark」は 大規模分散データ処理の業界標準。インメモリ計算で Hadoop MapReduce より 10〜100 倍高速。 です。 ここでは定義式の各記号、 直感的意味、 SSDSE-B-2026 への当てはめを段階的に解きほぐします。

① Spark のアーキテクチャ

② Spark のスループット概算

③ Spark の 4 つのコアモジュール

④ PySpark の基本構文

SparkSession を取得 → DataFrame 読み込み → transformation (select, filter, groupBy) → action (show, count, write) の 4 ステップが基本。 SQL クエリも spark.sql("SELECT ...") で実行可能。

⑤ Spark の最適化技法

• Catalyst Optimizer：SQL クエリの実行計画を自動最適化。 述語プッシュダウン、 列プルーニング。
• Tungsten：オフヒープメモリで Java オブジェクトオーバーヘッド排除、 CPU キャッシュ効率化。
• Broadcast Join：小テーブルを全 Executor に複製、 shuffle 回避。
• Adaptive Query Execution (AQE)：実行中にプランを動的調整。

SSDSE-B-2026 で「Apache Spark」を体感する

SSDSE-B-2026 を PySpark で都道府県別集計。 入力は 47 都道府県 × 12 年分の SSDSE-B-2026 行、 出力は Spark DataFrame で groupBy・agg・SQL クエリ。 47 都道府県 × 複数年の実データで具体計算を実施します。

🏭 産業界での活用事例（6 件）

⚖️ 関連手法との比較表

💥 失敗例から学ぶ

📝 演習問題（5 問・解答付き）

1. 問題 1：PySpark で SSDSE-B-2026 を読み込み、 都道府県別総人口を表示せよ。
   ▼ 解答

   📋 コピー

   1 2 3 4 5 6 7 8

   from pyspark.sql import SparkSession spark = SparkSession.builder.appName('SSDSE').getOrCreate() import pandas as pd # SSDSE-B-2026.csv は 1 行目が項目コード・2 行目が日本語見出しの 2 段ヘッダ。 # Spark の CSV 読み込みでは 1 行目を飛ばせないので、pandas で読んでから Spark に渡す pdf = pd.read_csv('data/raw/SSDSE-B-2026.csv', encoding='cp932', skiprows=1) df = spark.createDataFrame(pdf) df.filter(df['年度']==2023).select('都道府県','総人口').show(5)

2. 問題 2：都道府県別の出生数合計を Spark SQL で計算せよ。
   ▼ 解答

   df.createOrReplaceTempView('t')\nspark.sql(\"SELECT 都道府県, SUM(出生数) AS s FROM t GROUP BY 都道府県 ORDER BY s DESC\").show(10)

3. 問題 3：Spark DataFrame と pandas DataFrame の違いを 3 つ挙げよ。
   ▼ 解答

   (1) 分散 vs 単一マシン。 (2) lazy vs eager 評価。 (3) 不変 (immutable) vs 可変。 また Spark は Catalyst で実行計画最適化、 pandas は逐次実行。
4. 問題 4：Spark で groupBy の shuffle を避ける方法は？
   ▼ 解答

   (1) groupBy のキーで事前 partitionBy しておく。 (2) 集計関数を reduceByKey 系で局所集約。 (3) salting で偏り解消。 (4) Pre-aggregation でデータ量削減。
5. 問題 5：SSDSE-B-2026 を Spark MLlib で線形回帰し「総人口 → 出生数」を学習せよ。
   ▼ 解答

   📋 コピー

   1 2 3 4 5 6

   from pyspark.ml.feature import VectorAssembler from pyspark.ml.regression import LinearRegression va = VectorAssembler(inputCols=['総人口'], outputCol='features') d = va.transform(df).select('features','出生数') lr = LinearRegression(labelCol='出生数').fit(d) print(lr.coefficients, lr.intercept)

📖 関連用語辞典（10 語）

Apache Spark

Hadoop の後継となった分散インメモリ計算フレームワーク。

PySpark

Spark の Python API。 DataFrame・SQL・MLlib・Streaming すべて Python から呼び出し可。

RDD

Resilient Distributed Dataset。 Spark の低レベル分散コレクション。

DataFrame

RDD の上に SQL ライクな型付きインターフェースを乗せた高レベル API。

Catalyst

Spark SQL の最適化エンジン。 論理プラン→物理プランへ変換。

Executor

Spark のワーカープロセス。 タスクを実行しメモリキャッシュ保持。

Driver

Spark プログラムの中枢。 ジョブを Stage/Task に分解。

Shuffle

ノード間データ再配置。 join/groupBy で発生し、 ネットワーク I/O が支配的。

Broadcast Join

小テーブルを全 Executor に配布する高速 join。

MLlib

Spark の分散機械学習ライブラリ。 線形回帰・分類・クラスタリングを並列実行。

🧮 実値で計算してみる

PySpark で都道府県データを集計：

🧮 数式に値を入れて手で計算する: Spark 並列処理速度

合成 1TB データを N ノードで処理する所要時間を計算する。

Step 1: 単一ノード

Step 2: 並列化

🐍 Python で再現

1
2
3
4
5
6
7

import numpy as np
data_mb = 1_000_000
speed = 100
N = np.array([1, 10, 100])
serial = data_mb / speed
parallel = serial / N
print(f"並列時間: {parallel/60} 分")

📤 実行結果

💬 手計算 (Step 2) と Python 出力が完全一致。

🐍 Python での扱い

最小再現コード。 SSDSE-B のような実データを前提に、 4〜8 行で動く例です：

1
2
3
4
5
6
7
8
9

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName('demo').getOrCreate()
import pandas as pd
# SSDSE-B-2026.csv は 1 行目が項目コード・2 行目が日本語見出しの 2 段ヘッダ。
# Spark の CSV 読み込みでは 1 行目を飛ばせないので、pandas で読んでから Spark に渡す
pdf = pd.read_csv('data/raw/SSDSE-B-2026.csv', encoding='cp932', skiprows=1)
df = spark.createDataFrame(pdf)
df.groupBy('年度').count().orderBy('年度').show()  # action で実行される
spark.stop()

補足：ライブラリのバージョンや前処理状態によって出力は変わります。 自分の環境で動かすときは pip list でバージョンを確認し、 入力 CSV のパス・列名を実態に合わせてください。

🐍 Python 完全コード（4 要素ナレーション付き）

コード 1：SSDSE-B-2026 を PySpark で読み込み・集計

🎯 このコードでやること：47 都道府県 × 12 年分のデータを Spark DataFrame として読み込み、 年度別に総人口を集計する。

📥 入力データ：SSDSE-B-2026.csv (cp932, 564 行)、 年度・都道府県・総人口を含む。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

from pyspark.sql import SparkSession
from pyspark.sql.functions import sum as ssum, avg

spark = (SparkSession.builder
         .appName('SSDSE-Analysis')
         .config('spark.sql.shuffle.partitions', '4')
         .getOrCreate())

import pandas as pd
# SSDSE-B-2026.csv は 1 行目が項目コード・2 行目が日本語見出しの 2 段ヘッダ。
# Spark の CSV 読み込みでは 1 行目を飛ばせないので、pandas で読んでから Spark に渡す
pdf = pd.read_csv('data/raw/SSDSE-B-2026.csv', encoding='cp932', skiprows=1)
df = spark.createDataFrame(pdf)

# 文字列→数値
df = df.withColumn('総人口', df['総人口'].cast('long'))
df = df.withColumn('年度', df['年度'].cast('int'))

print("行数:", df.count())

(df.groupBy('年度')
   .agg(ssum('総人口').alias('全国総人口'),
        avg('総人口').alias('県平均'))
   .orderBy('年度')
   .show())

📤 実行結果（Spark では未実行：Spark 実行環境が無いため、 同じ集計を pandas で実測し、 show() の表の形に整えた値）：

💬 結果の読み方：Spark DataFrame で 564 行を読み込み、 12 年分の全国総人口推移を 1 ジョブで集計。 2012→2023 で約 324 万人減少 (年率 約-29.4 万人)。 同じことを pandas でやっても可能だが、 これが TB 級になると Spark でないと不可能。

コード 2：PySpark + SQL で都道府県別出生率ランキング

🎯 このコードでやること：Spark SQL を使い 2023 年の都道府県別合計特殊出生率を降順表示。

📥 入力データ：df: SSDSE-B-2026 全データ。 「合計特殊出生率」列を含む。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()
import pandas as pd
# SSDSE-B-2026.csv は 1 行目が項目コード・2 行目が日本語見出しの 2 段ヘッダ。
# Spark の CSV 読み込みでは 1 行目を飛ばせないので、pandas で読んでから Spark に渡す
pdf = pd.read_csv('data/raw/SSDSE-B-2026.csv', encoding='cp932', skiprows=1)
df = spark.createDataFrame(pdf)

df.createOrReplaceTempView('ssdse')

result = spark.sql("""
    SELECT 都道府県,
           CAST(合計特殊出生率 AS DOUBLE) AS tfr
    FROM ssdse
    WHERE 年度 = 2023
    ORDER BY tfr DESC
    LIMIT 10
""")
result.show()

📤 実行結果（Spark では未実行：Spark 実行環境が無いため、 同じ集計を pandas で実測し、 show() の表の形に整えた値）：

💬 結果の読み方：Spark SQL で 2023 年の出生率 TOP 10 を抽出。 沖縄が突出して高く 1.60、 西日本・九州が高い傾向。 同じクエリが 47 都道府県でも 47 億行でも書き換え不要 — これが Spark の力。

コード 3：Spark MLlib で線形回帰 (総人口 → 出生数)

🎯 このコードでやること：47 都道府県の 2023 年データで、 総人口から出生数を予測する線形回帰モデルを Spark MLlib で学習。

📥 入力データ：X = 総人口, y = 出生数, 47 行。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

from pyspark.ml.feature import VectorAssembler
from pyspark.ml.regression import LinearRegression
from pyspark.sql.functions import col

d = (df.filter(col('年度')==2023)
       .withColumn('総人口', col('総人口').cast('double'))
       .withColumn('出生数', col('出生数').cast('double'))
       .select('総人口', '出生数'))

va = VectorAssembler(inputCols=['総人口'], outputCol='features')
train = va.transform(d).select('features', '出生数')

lr = LinearRegression(labelCol='出生数', featuresCol='features')
model = lr.fit(train)

print(f"係数: {model.coefficients[0]:.6f}")
print(f"切片: {model.intercept:.2f}")
print(f"R^2 : {model.summary.r2:.4f}")

📤 実行結果（Spark では未実行：Spark 実行環境が無いため、 同じ集計を pandas で実測し、 show() の表の形に整えた値）：

💬 結果の読み方：総人口 1000 万人増 → 出生数 約 6.10 万人増 (出生率約 6.10‰)。 R²=0.991 と極めて高い決定係数。 sklearn と同じ結果が Spark でも得られる。 数十億行の住民データでも同じコードで動く点が Spark の意義。

コード 4：Spark Streaming 風: SSDSE-B 年度ループでバッチ集計

🎯 このコードでやること：SSDSE-B-2026 を年度別バッチとして処理し、 各年度の全国合計・最大・最小を出力 (ストリーミング処理の擬似デモ)。

📥 入力データ：df: 全データ、 年度別にループ。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

from pyspark.sql.functions import sum as ssum, max as smax, min as smin, col

for yr in sorted([r['年度'] for r in df.select('年度').distinct().collect()]):
    sub = df.filter(col('年度')==yr)
    agg = sub.agg(
        ssum('総人口').alias('合計'),
        smax('総人口').alias('最大'),
        smin('総人口').alias('最小')
    ).first()
    print(f"{yr}: 合計={agg['合計']:>10,}, 最大={agg['最大']:>9,}, 最小={agg['最小']:>7,}")

📤 実行結果（Spark では未実行：Spark 実行環境が無いため、 同じ集計を pandas で実測し、 show() の表の形に整えた値）：

💬 結果の読み方：12 年分を 1 ループで集計。 最大は常に東京 (1300万→1400万)、 最小は鳥取 (58万→54万)。 ストリーミングではこのループが Kafka など実時間ストリームになる。 Spark Structured Streaming なら全く同じ DataFrame API で書ける。

❓ FAQ 20 問

📖 Apache Spark の包括ガイド（追補編）

🔍 Apache Spark の多角的解釈

Lazy Evaluation の本質

変換は DAG に蓄積されるだけ。 action (count/show/write) で初めて実行。 不要計算を省くため。

Shuffle のコスト

ネットワーク I/O が最大コスト。 100GB の shuffle ≒ 1 ノード I/O 10 分。 broadcast join で回避。

Catalyst の最適化例

WHERE 句を JOIN の前に押し下げる (predicate pushdown)。 列選択も pushdown。

RDD vs DataFrame の性能差

DataFrame は Catalyst + Tungsten で型最適化＆コード生成。 同じ処理で 2-10 倍速い。

Spark の fault tolerance

Lineage (来歴) を保持。 ノード故障時にパーティションを再計算。 チェックポイントで lineage 短縮。

📓 「Apache Spark」をさらに深掘り — 実データ実践ノート

SSDSE-B-2026 を使った段階的ハンズオン。 初学者→中級→上級と順に深めるシナリオ構成です。

ステップ 1: SparkSession 起動

SparkSession.builder.appName('SSDSE').getOrCreate()。 local モードでも動く。

ステップ 2: CSV 読み込み

spark.read.option('header',True).option('encoding','cp932').csv('data/raw/SSDSE-B-2026.csv')。

ステップ 3: 型キャスト

df.withColumn('総人口', col('総人口').cast('long'))。 CSV は全列 string なので明示。

ステップ 4: フィルタ

df.filter(col('年度')==2023) で 2023 年に絞る。

ステップ 5: 集計

df.groupBy('年度').agg(sum('総人口').alias('全国'))。

ステップ 6: SQL

spark.sql('SELECT * FROM t WHERE 年度=2023 ORDER BY 総人口 DESC LIMIT 5')。

ステップ 7: Window 関数

Window.partitionBy('年度').orderBy(col('総人口').desc()) で年度別ランキング。

ステップ 8: MLlib 回帰

VectorAssembler + LinearRegression で「総人口 → 出生数」を学習。

ステップ 9: Parquet 保存

df.write.partitionBy('年度').parquet('out/')。

ステップ 10: Spark UI 確認

http://localhost:4040 で Job/Stage/Task の所要時間とメモリを可視化。

📋 Apache Spark チートシート

実務で頻用するコード・概念・公式を 1 ページにまとめた早見表。 印刷して机に貼っておくと便利です。

🐍 SSDSE-B-2026 × Apache Spark 追加コード集 (4 要素ナレーション)

本編のコードに加え、 さらに 4 種の発展的コード例を 4 要素 (🎯/📥/📤/💬) 付きで提示。 段階的に「読む→動かす→改造する」を体験できます。

追加コード 1：Spark Window 関数で年度別ランキング

🎯 このコードでやること：各年度内で都道府県の総人口ランキングを Window 関数で計算。

📥 入力データ：SSDSE-B-2026 の関連カラム。

1
2
3
4
5
6
7
8

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, col

w = Window.partitionBy('年度').orderBy(col('総人口').cast('long').desc())
df.withColumn('rank', row_number().over(w)) \
  .filter(col('rank') <= 3) \
  .select('年度','都道府県','総人口','rank') \
  .show(15)

📤 実行結果（Spark では未実行：Spark 実行環境が無いため、 同じ集計を pandas で実測し、 show() の表の形に整えた値）：

💬 結果の読み方：Window 関数で年度別 TOP 3 を抽出。 全年度で東京・神奈川・大阪が不動。 SQL の RANK と同じ機能を PySpark で実現。

追加コード 2：Spark UDF で都道府県カテゴリ分類

🎯 このコードでやること：Pandas UDF (vectorized) で人口規模カテゴリを高速付与。

📥 入力データ：SSDSE-B-2026 の関連カラム。

1
2
3
4
5
6
7
8
9
10

from pyspark.sql.functions import pandas_udf
import pandas as pd
@pandas_udf('string')
def categorize(pop: pd.Series) -> pd.Series:
    # pd.cut はカテゴリ型を返すので、'string' 型の UDF では文字列に直して返す
    return pd.cut(pop.astype(int),
                  bins=[0, 1000000, 3000000, 10000000, 1e10],
                  labels=['小','中','大','超大']).astype(str)
df = df.withColumn('カテゴリ', categorize(col('総人口')))
df.groupBy('カテゴリ').count().show()

📤 実行結果（Spark では未実行：Spark 実行環境が無いため、 同じ集計を pandas で実測し、 show() の表の形に整えた値）：

💬 結果の読み方：合計 115+329+108+12=564 で、 年度で絞っていないので 12 年度分の県×年を数えている。 「超大」の 12 件は東京都の 12 年度分だけで、 1,000 万人を超えるのは東京都しかない。 Pandas UDF は列をまとめて pandas に渡すので 1 行ずつ呼ぶ通常の UDF より速いが、 pd.cut の戻り値はカテゴリ型なので、 型を 'string' と宣言したら文字列に直してから返す。

追加コード 3：Spark MLlib + Pipeline で分類モデル

🎯 このコードでやること：VectorAssembler → StandardScaler → LogisticRegression をパイプライン化。

📥 入力データ：SSDSE-B-2026 の関連カラム。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

from pyspark.ml import Pipeline
from pyspark.ml.feature import VectorAssembler, StandardScaler
from pyspark.ml.classification import LogisticRegression

va = VectorAssembler(inputCols=['総人口','出生数'], outputCol='raw')
sc = StandardScaler(inputCol='raw', outputCol='features')
lr = LogisticRegression(labelCol='label', featuresCol='features')
pipe = Pipeline(stages=[va, sc, lr]).fit(train_df)
preds = pipe.transform(test_df)
preds.select('都道府県','probability','prediction').show(5)

📤 実行結果（train_df・test_df を用意していない擬似コードなので、 表示の形を示す例で実測値ではない）：

💬 結果の読み方：Pipeline で前処理 + 学習 + 推論を一気通貫化。 production デプロイでもこのパイプラインをそのまま使える。 sklearn とほぼ同じ API。

追加コード 4：Spark Structured Streaming で疑似ストリーミング

🎯 このコードでやること：SSDSE-B のファイルを 1 行ずつ追加する疑似ストリーミングを構成。

📥 入力データ：SSDSE-B-2026 の関連カラム。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()
import pandas as pd
# SSDSE-B-2026.csv は 1 行目が項目コード・2 行目が日本語見出しの 2 段ヘッダ。
# Spark の CSV 読み込みでは 1 行目を飛ばせないので、pandas で読んでから Spark に渡す
pdf = pd.read_csv('data/raw/SSDSE-B-2026.csv', encoding='cp932', skiprows=1)
df = spark.createDataFrame(pdf)   # 列名と型（スキーマ）をここから借りる

# Structured Streaming のテンプレート (実環境で動作)
stream = (spark.readStream
          .option('header', True)
          .option('encoding', 'cp932')
          .schema(df.schema)
          .csv('stream_dir/'))

query = (stream.groupBy('年度')
         .count()
         .writeStream
         .outputMode('complete')
         .format('console')
         .start())

📤 実行例（イメージ・未実測：Spark 実行環境が無く、 stream_dir/ にファイルを置いたときに出る表示の形だけを示す）：

💬 結果の読み方：start() は監視を始めるだけで、 その時点では何も表示されない。 stream_dir/ に CSV が置かれるたびに console へ「Batch: 0」「Batch: 1」… と年度別の件数表が出力モード complete で丸ごと出し直される（ここでは実行していないので件数は示さない）。 SSDSE-B-2026.csv は項目コードと日本語見出しの 2 段ヘッダなので、 そのまま置くと 2 行目の見出しがデータ行として読まれる。 置くファイルは見出し 1 行に整え、 スクリプトとして動かすなら最後に query.awaitTermination() を書かないとすぐ終了する。

⚠️ よくある落とし穴

Spark の 5 大事故は 「小さいデータ (数百 MB 以下) で使い性能逆転」「遅延評価で filter().show() しないと何も起きない罠」「collect() で Driver の OOM」「Python UDF が Java/Scala 比 10〜100 倍遅い」「groupBy/join の Shuffle で全ノード間ネットワーク転送」。 特に最後の Shuffle は数 TB 規模で 1 ジョブが数時間刺さる典型原因で、 broadcast join や partitionBy 設計で回避します。

※ 上記は文献調査・現場経験で報告される頻度の高い注意点。 ドメインや手法のバージョンによって追加の落とし穴がある場合があります。

⚠️ さらに踏み込んだ失敗パターン 10 連

初学者を超えた中級者がハマる「2 周目の失敗例」を集めました。 本編の 5 件と合わせて 15 件のチェックリストとして活用してください。

📖 Apache Spark 関連語拡張辞典（20 語）

本編の 10 語に加え、 さらに専門用語 20 個を整理。 文献を読むときの「分からない単語チェッカー」として使えます。

SparkContext

Spark の core エントリポイント。 RDD 操作の起点。

SparkSession

DataFrame/SQL のエントリポイント。 SparkContext を内包。

DataFrame

型付き分散テーブル。 RDD の上の高レベル API。

Dataset

Scala/Java の型安全 DataFrame。 PySpark には無い。

Catalyst

Spark SQL の最適化エンジン。 論理→物理プラン変換。

Tungsten

オフヒープメモリ + コード生成によるパフォーマンス強化。

AQE (Adaptive Query Execution)

実行時にプランを動的調整。 Spark 3.x 標準。

Catalog

テーブル・ビューのメタストア。 Hive と互換。

Delta Lake

Spark 上の ACID トランザクション付きストレージ層。

Structured Streaming

DataFrame ベースのストリーミング API。 micro-batch + continuous。

MLlib

Spark 分散機械学習ライブラリ。 Pipeline API 中心。

Pipeline

Estimator + Transformer をチェーン化するパターン。

Estimator

fit() でモデルを返すコンポーネント。 学習を担う。

Transformer

transform() でデータ変換するコンポーネント。

Broadcast Variable

全 Executor に複製される共有変数。 join 時の小テーブル配布。

Accumulator

Executor 側からカウントアップできる変数。 監視・統計に。

DAG

有向非巡回グラフ。 Spark の実行計画表現。

Stage

shuffle で区切られた処理単位。 並列実行可。

Task

1 パーティションを処理する最小実行単位。

Shuffle

ノード間データ再配置。 join/groupBy で発生。

❓ FAQ 追補 20 問（中〜上級者向け）

🌟 SSDSE-B-2026 で「Apache Spark」の総合演習

47 都道府県 × 12 年分のデータを使い、 Apache Spark を多角的に体験する総合演習。 単発の操作ではなく、 一連の分析フローを通して理解を深めます。

ハンズオン 1: PySpark で SSDSE-B-2026 を全部処理

🎯 このコードでやること：PySpark で SSDSE-B-2026 を全部処理

📥 入力：SSDSE-B-2026.csv の 47 都道府県データ。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

from pyspark.sql import SparkSession
from pyspark.sql.functions import sum as ssum, avg, col

spark = SparkSession.builder.appName('SSDSE').getOrCreate()
import pandas as pd
# SSDSE-B-2026.csv は 1 行目が項目コード・2 行目が日本語見出しの 2 段ヘッダ。
# Spark の CSV 読み込みでは 1 行目を飛ばせないので、pandas で読んでから Spark に渡す
pdf = pd.read_csv('data/raw/SSDSE-B-2026.csv', encoding='cp932', skiprows=1)
df = spark.createDataFrame(pdf)

df = df.withColumn('総人口', col('総人口').cast('long'))
df = df.withColumn('年度', col('年度').cast('int'))

# 年度別全国人口
(df.groupBy('年度')
   .agg(ssum('総人口').alias('全国'))
   .orderBy('年度')
   .show())

📤 実行結果（Spark では未実行：Spark 実行環境が無いため、 同じ集計を pandas で実測し、 show() の表の形に整えた値）：

💬 結果の読み方：564 行を Spark DataFrame で集計。 同じコードが 6 億行でも動く。 これが Spark の意義。

🌐 関連手法・派生

• Spark SQL：DataFrame に SQL でクエリ可能
• MLlib：分散機械学習ライブラリ
• Structured Streaming：ストリームデータ処理
• Databricks：Spark を商用化した分析プラットフォーム
• Delta Lake：Spark + ACID トランザクション

❓ よくある質問

🔗 関連用語 — 前提・並列・発展

「Spark」をより深く理解するには、 周辺の用語を セットで 押さえるのが効率的です。 以下は前提（先に知っておくべき）、 並列（兄弟概念）、 発展（次に学ぶべき）の関係にある用語群：

クリックで各用語の詳細ページへ移動します。 全用語の体系は 用語集トップ または 概念マップ から俯瞰できます。

📚 関連グループ教材

「Spark」は単独で完結する概念ではなく、 より大きな分野の一部です。 上位カテゴリの教材を読むことで、 この用語の 位置づけ が立体的に見えてきます：

• 📚 データ処理 — このカテゴリの体系的解説

📚 参考文献・出典

• Zaharia, M. et al. (2010). Spark: Cluster Computing with Working Sets. HotCloud.
• Zaharia, M. et al. (2012). Resilient Distributed Datasets. NSDI.
• Karau, H. et al. (2017). High Performance Spark. O'Reilly.
• Chambers, B. & Zaharia, M. (2018). Spark: The Definitive Guide. O'Reilly.
• Apache Spark 公式ドキュメント https://spark.apache.org/docs/latest/
• Databricks Spark Tutorials https://docs.databricks.com/
• 独立行政法人統計センター. SSDSE-B-2026 (CC BY 4.0). https://www.nstac.go.jp/use/literacy/ssdse/

🌟 拡張ハンドブック

📜 Spark の歴史

Apache Spark は UC Berkeley AMPLab の Matei Zaharia らが 2009 年に開発を開始、 2010 年公開、 2013 年に Apache プロジェクト入り、 2014 年 1.0 リリース。 Hadoop MapReduce のディスク I/O ボトルネックを「インメモリ計算」で解決し、 反復処理 (機械学習・グラフ) で 10-100 倍の高速化を達成。 2016 年 2.0、 2020 年 3.0 で Catalyst 拡張・Pandas UDF・Adaptive Query Execution を導入。

📐 Spark の DAG (有向非巡回グラフ) 実行モデル

SparkContext がジョブを DAG に変換 → DAGScheduler が Stage に分解 → TaskScheduler が Executor に配布。 各 Stage はパイプライン化されたトランスフォーメーション、 Stage 境界は shuffle。 fault tolerance は lineage (来歴) で実現：パーティション再計算可能。

⚙️ PySpark の動作原理

PySpark は Python プロセスが JVM (Py4J 経由) と通信。 DataFrame 操作はすべて JVM 側で実行されオーバーヘッド極小。 UDF は Python プロセスへシリアライズして実行 → 遅い。 Pandas UDF (Arrow ベース) なら vectorize されて高速。

💼 Spark を実務で使う 5 つの場面

• 大規模 ETL：日次 TB 級ログを集計し DWH に書き込み。
• 機械学習特徴量生成：億行から特徴量テーブルを生成、 学習データに。
• ストリーミング集計：Kafka からリアルタイム取り込み、 ダッシュボード更新。
• SQL データレイク：Delta Lake/Iceberg + Spark SQL で BI クエリ。
• 分散学習推論：MLlib で学習、 または PyTorch/TF を pandas UDF で並列実行。

📊 Spark vs pandas 性能目安

📚 50 連発レシピ集（Apache Spark × Python）

SSDSE-B-2026 を題材に、 Apache Spark の実装を 50 ミニコードで一気に俯瞰します。

1. from pyspark.sql import SparkSession; spark = SparkSession.builder.appName('app').getOrCreate()
2. df = spark.read.csv('data/raw/SSDSE-B-2026.csv', header=True, encoding='cp932')
3. df.printSchema() — 型情報表示
4. df.show(5) — 先頭 5 行
5. df.count() — 行数 (47 × 12年 = 564 行)
6. df.columns — カラム名一覧
7. df.select('都道府県', '総人口').show()
8. df.filter(df['年度']=='2023').show(5)
9. df.filter("年度=2023 AND 総人口>5000000").show()
10. df.groupBy('年度').count().show()
11. df.groupBy('都道府県').sum('総人口').show(5)
12. df.orderBy('総人口', ascending=False).show(5)
13. from pyspark.sql.functions import col, sum as ssum
14. df.agg(ssum('総人口')).show()
15. df.createOrReplaceTempView('t')
16. spark.sql('SELECT 都道府県, SUM(出生数) FROM t GROUP BY 都道府県').show()
17. spark.sql('SELECT * FROM t WHERE 年度=2023 ORDER BY 総人口 DESC LIMIT 5').show()
18. df.write.parquet('out/ssdse.parquet') — Parquet 出力
19. df2 = spark.read.parquet('out/ssdse.parquet')
20. df.repartition(4) — 4 partition に再分割
21. df.coalesce(1) — 1 partition に統合 (collect 用)
22. df.cache() — メモリキャッシュ
23. df.persist() — メモリ + ディスク persist
24. from pyspark.sql.functions import avg, max, min
25. df.groupBy('都道府県').agg(avg('出生数'), max('総人口')).show(5)
26. df.withColumn('密度', col('総人口')/col('面積'))
27. df.drop('地域コード')
28. df.dropDuplicates(['都道府県','年度'])
29. df.fillna(0)
30. df.na.fill({'出生数':0})
31. from pyspark.sql.functions import udf
32. upper_udf = udf(lambda s: s.upper())
33. from pyspark.ml.feature import VectorAssembler
34. va = VectorAssembler(inputCols=['総人口','出生数'], outputCol='features')
35. from pyspark.ml.regression import LinearRegression
36. lr = LinearRegression(featuresCol='features', labelCol='死亡数').fit(va.transform(df))
37. print(lr.coefficients)
38. from pyspark.ml.classification import LogisticRegression
39. from pyspark.ml.clustering import KMeans
40. km = KMeans(k=4).fit(va.transform(df))
41. km.transform(va.transform(df)).select('都道府県','prediction').show()
42. spark.conf.set('spark.sql.shuffle.partitions', '8')
43. spark.conf.set('spark.sql.adaptive.enabled', 'true') — AQE 有効化
44. from pyspark.sql.functions import broadcast
45. df.join(broadcast(small_df), 'key').show()
46. df.toPandas() — Spark → pandas (小データのみ)
47. spark.createDataFrame(pdf) — pandas → Spark
48. df.write.mode('overwrite').parquet('out/')
49. df.write.partitionBy('年度').parquet('out/') — 年度別分割
50. print('🎓 Spark の本質：lazy + 分散 + Catalyst による大規模並列処理')

🗺 概念マップ

関連概念を視覚的に整理した概念マップ。

マップ中心の Apache Spark から 6 軸が広がる。 「Hadoop (前身、 MapReduce ベース)」「Databricks (Spark 商用版)」「Delta Lake (ACID トランザクション層)」「PySpark (Python API)」「MLlib (機械学習ライブラリ)」「Structured Streaming (準リアルタイム処理)」が連結し、 TB 級データに対する分散処理 → 学習 → ストリーミング推論まで一つのフレームワークで完結する。

🔗 隣接手法への橋渡し

「Apache Spark」は クラスタ上で分散処理する大規模データ処理基盤 として、 上流の HDFS/S3 等の分散ストレージと下流の機械学習 (MLlib)・SQL 解析・ストリーミングを繋ぐ。 pandas が単一マシンで詰まる規模（数十 GB 以上）で真価を発揮する選択肢となる。

⬆️ 上流: 分散処理の基礎

• Hadoop — HDFS の上で動く
• MapReduce — Spark の先行概念
• クラスタコンピューティング — 複数ノードで並列

⬌ 並列: 他の分散基盤

• Apache Flink — ストリーミング特化
• Dask — Python ネイティブ
• Ray — ML/RL 寄り

⬇️ 下流: 応用処理

• MLlib — Spark 上の ML ライブラリ
• Structured Streaming — リアルタイム解析
• Delta Lake — ACID テーブル

Spark は大規模データを RDD / DataFrame で分散処理する基盤で、 Hadoop / MapReduce の系譜を継ぎ、 MLlib・Structured Streaming・Delta Lake と連携してバッチ・ML・リアルタイムを一基盤で扱う。

🌳 手法選択フロー

「Spark」を使うかは、 データ量と分散環境の必要性で判断する。

1. データが単一マシンに乗るか (<100GB)? Yes → pandas/polars で十分、 No → Spark 検討
2. SQL 主体か? Yes → SparkSQL または DuckDB/BigQuery で代替可、 No → DataFrame API
3. ストリーミング処理か? Yes → Structured Streaming or Kafka 連携、 No → バッチで十分

SSDSE-B-2026 は 47 行 × 約 100 列で pandas で十分。 Spark はテラバイト級・分散環境が前提なので、 教育用途では概念学習に留め、 実体験は EMR/Databricks で行う。