PySpark入門ガイド｜Databricks試験対策の基本構文

PySparkはDatabricks試験で最も出題頻度の高い技術の一つです。 Data Engineer Associate・Spark Developer Associate・ML Associate いずれの試験でも、PySparkのコード読解問題が出題されます。 この記事では、試験対策に必要なPySpark基本構文をDataFrame API・フィルタリング・集計・結合・UDF・Structured Streamingまで体系的に解説します。

PySparkとは

PySparkは、Apache SparkをPythonから操作するためのAPIです。Sparkは大規模データの分散処理エンジンであり、PySparkを使うことでPythonプログラマーがSparkの強力な分散処理能力を活用できます。

Databricks環境ではPySparkがデフォルトで利用可能で、 ノートブック上でインタラクティブにデータ処理を行えます。 試験ではPySparkの基本構文を読んで理解できる力が求められます。 コードを書く必要はありませんが、各APIの動作を正確に理解しておく必要があります。

• SparkSession: PySparkの起点となるオブジェクト。Databricksではデフォルトでspark変数として利用可能
• DataFrame: Sparkの主要なデータ構造。行と列で構成され、SQLテーブルに似た操作が可能
• Transformation: 遅延評価される変換処理（select, filter, groupBy等）
• Action: 実際に計算を実行する操作（show, collect, count, write等）

DataFrame API の基本操作

DataFrameはPySparkの中核となるデータ構造です。Databricks試験では、DataFrameの作成・表示・列選択の基本操作が頻出します。

DataFrameの作成

DataFrameを作成する方法は複数あります。試験で最も出題されるのはspark.readを使ったファイル読み込みです。

# CSVファイルから読み込み
df = spark.read.csv("/path/to/file.csv", header=True, inferSchema=True)

# JSONファイルから読み込み
df = spark.read.json("/path/to/file.json")

# Deltaテーブルから読み込み
df = spark.read.format("delta").load("/path/to/delta")
# または
df = spark.table("my_catalog.my_schema.my_table")

# Pythonリストから作成
data = [("Alice", 30), ("Bob", 25)]
df = spark.createDataFrame(data, ["name", "age"])

DataFrameの表示（show）

show()メソッドでDataFrameの内容を表示します。デフォルトで先頭20行が表示されます。

# 先頭20行を表示
df.show()

# 先頭5行を表示
df.show(5)

# 列を切り詰めずに表示
df.show(truncate=False)

# スキーマ（列名と型）を表示
df.printSchema()

# 行数を取得
df.count()

列の選択（select）

select()で必要な列だけを取得します。列の参照方法は複数ありますが、試験ではcol()関数を使う方法が多く出題されます。

from pyspark.sql.functions import col, upper, lit

# 文字列で列を指定
df.select("name", "age")

# col()関数で列を指定
df.select(col("name"), col("age"))

# 列に変換を適用
df.select(col("name"), (col("age") + 1).alias("age_next_year"))

# withColumnで新しい列を追加
df.withColumn("name_upper", upper(col("name")))

# 定数列を追加
df.withColumn("country", lit("Japan"))

フィルタリング（where / filter）

where()とfilter()は完全に同じ動作をします。条件に合致する行だけを抽出します。試験では両方の書き方が出題されます。

from pyspark.sql.functions import col

# 基本的なフィルタリング
df.filter(col("age") > 25)
df.where(col("age") > 25)  # filterと同じ

# 文字列条件（SQL式）
df.filter("age > 25 AND city = 'Tokyo'")

# 複数条件（AND）
df.filter((col("age") > 25) & (col("city") == "Tokyo"))

# 複数条件（OR）
df.filter((col("age") > 30) | (col("city") == "Osaka"))

# NOT条件
df.filter(~(col("status") == "inactive"))

# NULL判定
df.filter(col("email").isNotNull())
df.filter(col("email").isNull())

# IN条件
df.filter(col("city").isin("Tokyo", "Osaka", "Nagoya"))

# LIKE条件
df.filter(col("name").like("A%"))
df.filter(col("name").contains("田"))

# BETWEEN条件
df.filter(col("age").between(20, 30))

試験のコード問題では、&（AND）や|（OR）を使う際に 各条件を括弧で囲む必要がある点が問われることがあります。 括弧がないとPythonの演算子優先順位の問題でエラーになります。

集計（groupBy / agg）

groupBy()とagg()を組み合わせることで、グループごとの集計処理を行います。SQLのGROUP BY句と同等の操作です。

from pyspark.sql.functions import count, sum, avg, max, min, round

# 基本的なgroupBy
df.groupBy("department").count()

# 複数の集計関数を同時に適用
df.groupBy("department").agg(
    count("*").alias("employee_count"),
    avg("salary").alias("avg_salary"),
    max("salary").alias("max_salary"),
    min("salary").alias("min_salary"),
    sum("salary").alias("total_salary")
)

# 複数列でグループ化
df.groupBy("department", "job_title").agg(
    count("*").alias("count"),
    round(avg("salary"), 2).alias("avg_salary")
)

# ソート（orderBy）
df.groupBy("department").count().orderBy(col("count").desc())

# 上位N件を取得
df.groupBy("department").count().orderBy(col("count").desc()).limit(5)

試験ではalias()で集計列に名前を付ける方法と、orderBy()のasc/descの指定方法がよく問われます。sort()はorderBy()のエイリアスで、同じ動作をします。

結合（join）

複数のDataFrameを結合するにはjoin()メソッドを使います。SQLのJOIN句と同等で、結合タイプ（inner, left, right, full, cross, semi, anti）を指定できます。

# inner join（デフォルト）
df1.join(df2, df1["id"] == df2["id"], "inner")

# left outer join
df1.join(df2, df1["id"] == df2["id"], "left")

# right outer join
df1.join(df2, df1["id"] == df2["id"], "right")

# full outer join
df1.join(df2, df1["id"] == df2["id"], "full")

# cross join（直積）
df1.crossJoin(df2)

# left semi join（df2に存在するdf1の行のみ）
df1.join(df2, df1["id"] == df2["id"], "left_semi")

# left anti join（df2に存在しないdf1の行のみ）
df1.join(df2, df1["id"] == df2["id"], "left_anti")

# 同名列での結合（列名が同じ場合のショートカット）
df1.join(df2, "id")         # inner join
df1.join(df2, ["id", "name"])  # 複数列での結合

試験では特にleft semi joinとleft anti joinの違いが問われます。 semi joinは「結合キーが一致する左テーブルの行だけを返す」、 anti joinは「結合キーが一致しない左テーブルの行だけを返す」という動作です。 いずれも右テーブルの列は結果に含まれません。

ユーザー定義関数（UDF）

組み込み関数で対応できないカスタム処理が必要な場合、UDF（User Defined Function）を定義できます。 ただし、UDFはSpark最適化が効かないため、可能な限り組み込み関数を使うべきです。

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType, IntegerType

# 方法1: udf()関数で登録
def classify_age(age):
    if age < 20:
        return "young"
    elif age < 60:
        return "adult"
    else:
        return "senior"

classify_age_udf = udf(classify_age, StringType())
df.withColumn("age_group", classify_age_udf(col("age")))

# 方法2: デコレータで登録
@udf(returnType=StringType())
def format_name(first, last):
    return f"{last} {first}"

df.withColumn("full_name", format_name(col("first_name"), col("last_name")))

# 方法3: Pandas UDF（高速・推奨）
from pyspark.sql.functions import pandas_udf
import pandas as pd

@pandas_udf(IntegerType())
def double_value(s: pd.Series) -> pd.Series:
    return s * 2

df.withColumn("doubled", double_value(col("value")))

試験では「UDFとPandas UDFの違い」がよく問われます。Pandas UDFはApache ArrowベースでベクトルI/Oを行うため、 通常のPython UDFよりも大幅に高速です。新規開発ではPandas UDFの使用が推奨されます。

Structured Streaming

Structured Streamingは、Sparkのバッチ処理APIを使ってストリーミングデータを処理するエンジンです。Databricks試験ではAuto Loader（cloudFiles）と合わせた出題が非常に多い分野です。

# Auto Loader でクラウドストレージからストリーム読み込み
df_stream = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", "/schema/path")
    .load("/data/input/"))

# ストリームへの変換処理
df_transformed = (df_stream
    .select("id", "name", "timestamp")
    .filter(col("status") == "active"))

# Deltaテーブルへのストリーム書き込み
(df_transformed.writeStream
    .format("delta")
    .option("checkpointLocation", "/checkpoint/path")
    .outputMode("append")
    .trigger(availableNow=True)
    .toTable("my_catalog.my_schema.output_table"))

# トリガーモードの種類
# .trigger(processingTime="10 seconds")  # マイクロバッチ（10秒ごと）
# .trigger(once=True)                     # 1回だけ実行（非推奨）
# .trigger(availableNow=True)             # 利用可能な全データを処理（推奨）

checkpointLocationは必須オプションです。チェックポイントにはストリームの進行状況が記録され、障害からの回復に使われます。試験では「checkpointLocationを指定しないとどうなるか」という形式でも出題されます。

outputModeには3種類あります。

• append: 新しい行のみ出力。集計なしの処理に使用（最も一般的）
• complete: 全結果を毎回出力。集計処理に使用
• update: 更新された行のみ出力。集計処理で変更分のみ必要な場合に使用

PySpark問題にチャレンジ

ここまでの内容を踏まえて、実際の試験に近い問題を解いてみましょう。

よくある質問（FAQ）