AWS Glue ではじめる PySpark 超入門｜分散処理は「いつ」使うのか

「ビッグデータ＝Spark」と身構える前に、まず知ってほしいことがあります。小さいデータなら Lambda や pandas で十分で、Spark はむしろ過剰です。本記事は AWS Glue というサーバーレス ETL を入口に、PySpark で分散処理を触り、「いつ使うか」の判断軸まで持ち帰れる超入門です。

01なぜ分散処理か — その前に「使わない判断」

分散処理とは、ひとことで言えば「1台のマシンに載りきらない／時間がかかりすぎるデータを、複数のマシンに分けて処理する」仕組みです。その代表格が Apache Spark、そして Spark を Python から扱うのが PySpark です。

ただ、入門記事でまず最初に伝えたいのは逆のことです。データが小さいなら、分散処理は要りません。1台のメモリに載って数分で終わる程度の量なら、pandas や AWS Lambda、あるいは Athena（S3 のデータに SQL で直接集計）で十分です。Spark は起動が重く、小さいデータではむしろ遅く・高くつきます。

ざっくりの目安（厳密ではありません）はこうです。

• 分散処理が効く例：数十GB〜TB級のログを毎日 Parquet に変換、大量ファイルの一括加工、1台では時間内に終わらない集計。
• 要らない例：数千〜数百万行の CSV 集計、日次の軽い加工、1ファイルで完結する処理 → Python×Lambda の実践ユースケース や Athena で十分。

02用語の地図 — 最初に押さえる4つ＋α

PySpark を読み書きするうえで、これだけ分かっていれば迷いにくい、という言葉を先に地図にしておきます。細部は後の章のコードで自然に出てきます。

もっと手前の TCP/IP・Linux・クラウドの基礎から積みたい方は、クラウドエンジニアの教科書（基礎トレーニング）もあわせてどうぞ。

03AWS での触り方 — Glue と EMR の違い

AWS で Spark を動かす代表的な選択肢が AWS Glue と Amazon EMR です。ざっくり言うと、Glue はサーバーレスの ETL サービス（クラスタを自分で持たない）、EMR は自分で構築・管理する Spark クラスタです。

Glue のジョブは PySpark で書きます。しかも Glue 独自の DynamicFrame と、素の Spark DataFrame の両方が使えます。カタログ連携やスキーマ揺れに強いのが DynamicFrame、普通の集計・結合が書きやすいのが DataFrame、という住み分けです（後述）。

2026年時点のバージョンの位置づけは次の通りです。バージョンで Spark と Python が変わるので、ジョブ設定の「Glue バージョン」は必ず意識します。

04UC：S3 の CSV/JSON を Parquet に変換する

もっとも基本のユースケースです。生の CSV/JSON を、集計に向いた列指向フォーマット Parquet に変換し、日付でパーティション分割して書き戻します。Glue ジョブの PySpark はこの「読む→変換→書く」が土台になります。

まず Glue ジョブ内で Spark を使うための定型（ボイラープレート）です。glueContext.spark_session から、素の Spark DataFrame API がそのまま使えます。

import sys
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from pyspark.sql import functions as F

sc = SparkContext.getOrCreate()
glueContext = GlueContext(sc)
spark = glueContext.spark_session   # ← 素の Spark DataFrame はこれ経由で使う

# 1) 読む：S3 の CSV（ヘッダあり・型は自動推論）
df = (spark.read
        .option("header", "true")
        .option("inferSchema", "true")
        .csv("s3://YOUR-BUCKET/raw/access/"))

# 2) 変換：列を絞る・型を直す・日付列(dt)を足す
out = (df.select("user_id", "path", "status", "ts")
         .withColumn("status", F.col("status").cast("int"))
         .withColumn("dt", F.to_date("ts")))

# 3) 書く：Parquet で出力。dt でパーティション分割
(out.write
    .mode("overwrite")
    .partitionBy("dt")
    .parquet("s3://YOUR-BUCKET/curated/access/"))

JSON を読むなら spark.read.json("s3://.../")、区切りが違う CSV は .option("sep", "\t") のように指定します。partitionBy("dt") で S3 上が dt=2026-07-15/ のように分割され、後述の Athena で「特定日だけ読む」ことができて速く・安くなります。

05UC：集計と結合（groupBy / agg / join）

次に使う頻度が高いのが、絞り込み・列の追加・集計・結合です。pandas に近い感覚で書けますが、裏では分散実行されます。

from pyspark.sql import functions as F

# filter → withColumn → groupBy().agg()（ユーザー×日次のアクセス数）
daily = (df.filter(F.col("status") < 400)
           .withColumn("dt", F.to_date("ts"))
           .groupBy("user_id", "dt")
           .agg(F.count("*").alias("hits")))

# join：ユーザー属性マスタと左結合（user_id で紐づけ）
users = spark.read.parquet("s3://YOUR-BUCKET/master/users/")   # user_id, plan など
report = daily.join(users, on="user_id", how="left")

# 確認：先頭5行だけドライバに集める（全件 collect はしない）
report.show(5)
report.printSchema()

• filter()：行の絞り込み。where() でも同じ。
• withColumn("新列", 式)：列の追加・変換。
• groupBy(...).agg(...)：集計。count / sum / avg / countDistinct など。
• join(other, on, how)：結合。how は inner / left / outer など。

06UC：カタログ連携（Crawler → Data Catalog → Athena）

Parquet に変換しただけでは「S3 にファイルがある」状態です。これを SQL で扱える「テーブル」に見せるのが Glue Data Catalog です。カタログは、どの S3 パスに・どんな列とパーティションの表があるか、というメタデータの置き場所です。

カタログにテーブルを登録する簡単な方法が Crawler です。Crawler は指定した S3 パスを自動でスキャンし、列やパーティションを推定してカタログにテーブル定義を作成・更新します。作成後は CLI からも起動できます。

# Crawler を起動（テーブル定義を最新化）
aws glue start-crawler --name access-curated-crawler

CLI の基本操作は AWS CLI 実践リファレンスにまとめています。Crawler がテーブル（例：mydb.access）を作ったら、Athena から標準 SQL で中身を確認できます。Athena はメタデータの参照先として Glue Data Catalog を使うため、Glue で作った表がそのまま見えます。

-- Athena：Glue Data Catalog のテーブルを SQL で確認
SELECT dt, count(*) AS rows
FROM   mydb.access
WHERE  dt >= DATE '2026-07-01'
GROUP  BY dt
ORDER  BY dt;

カタログは Glue のジョブからも直接使えます。ここで DynamicFrame の出番です。スキーマが揺れる生データの取り込みに強く、素の DataFrame と相互変換できます。

from awsglue.dynamicframe import DynamicFrame

# カタログのテーブルから DynamicFrame で読む（スキーマ揺れに強い）
dyf = glueContext.create_dynamic_frame.from_catalog(
        database="mydb", table_name="access")

df = dyf.toDF()                                    # → 素の Spark DataFrame に変換して集計
# ...（04・05 のように変換）...
dyf2 = DynamicFrame.fromDF(df, glueContext, "dyf2")  # → 書き戻すとき DynamicFrame に戻す

07つまづき所 — 事故らないための勘どころ

• 小さすぎるデータに Spark は過剰：起動オーバーヘッドがあるため、数百MB程度までなら pandas / Lambda / Athena の方が速くて安いことが多い。「まず Spark を使わない」を検討。
• シャッフルを甘く見ない：join / groupBy / distinct は全体の再配置が起き、遅く・高くなりがち。結合前に行と列を減らす、同じ集計を繰り返さない。
• パーティションは粒度が命：細かすぎると小さなファイルが大量にでき、粗すぎると読み過ぎになる。書き出しの partitionBy は「よく絞る列（日付など）」を選ぶ。
• collect() の罠：全件をドライバに集めるとメモリ不足で落ちる。確認は show(n) や limit() で少量だけ。
• コストは実行時間の従量：ワーカー数×実行時間で決まる。無駄な再実行・巨大な全件スキャンに注意（具体的な単価は必ず公式の料金ページで確認）。
• ローカルは pandas で試作：いきなり Glue で回すと遅く・高い。まずは手元で pandas やサンプルデータで動きを固め、それから Glue へ載せる。
• 秘密情報を直書きしない：接続情報・アクセスキーはコードに書かず、Secrets Manager やジョブパラメータ、IAM ロールで渡す。YOUR_KEY のようなプレースホルダで管理し、鍵をコミットしない。

08早見表 — 1画面チートシート

PySpark DataFrame ミニ早見（F は from pyspark.sql import functions as F）。

Glue / EMR / Athena の使い分け。

09よくある質問（FAQ）