pandas API on Spark

pandas API on Spark（pyspark.pandas）は、 使い慣れたPandasの構文をそのままApache Sparkの分散処理エンジン上で実行できるAPIです。 シングルノードのPandasでは扱えない大規模データセットを、 コードの書き換えを最小限にして処理できます。

この記事では、Koalasからの統合経緯、Pandasとの比較、DataFrame相互変換のコード例、分散環境での制約と非対応API、Pandas UDFとの違い、Databricks試験での出題ポイントまでを網羅的に解説します。

pandas API on Sparkとは

pandas API on Sparkは、Apache Spark 3.2で正式にSpark本体に統合されたAPIで、pyspark.pandasモジュールとして提供されています。 前身はDatabricksが開発したオープンソースプロジェクト「Koalas」で、 PandasユーザーがSparkの分散処理を透過的に利用できるよう設計されました。

最大の特徴は、PandasのDataFrameやSeriesと同じAPIを提供しつつ、内部的にはSpark DataFrameとして分散処理される点です。これにより数百GB〜数TB規模のデータを、Pandasの直感的なコードで処理可能です。

Koalasからの統合経緯

Koalasは2019年にDatabricksがOSSとしてリリースしました。 「PandasコードをそのままSparkで動かしたい」というニーズに応え、 PandasのAPI互換レイヤーをSpark上に実装したものです。 Spark 3.2（2022年1月リリース）でApache Spark本体に統合され、pyspark.pandasとして標準モジュールになりました。 現在はpip install koalasは不要で、 PySpark 3.2以降であれば追加インストールなしで利用可能です。

インポート方法

pandas API on Sparkのインポートは以下の1行で完了します。 慣例としてpsというエイリアスが使われます。

import pyspark.pandas as ps

# CSVファイルの読み込み（Pandasと同じ構文）
df = ps.read_csv("/mnt/data/sales.csv")

# Parquetファイルの読み込み
df = ps.read_parquet("/mnt/data/sales.parquet")

# Deltaテーブルの読み込み（Databricks固有）
df = ps.read_delta("/mnt/data/delta_table")

# 基本操作
df.head(10)
df.describe()
df.shape
df.dtypes

Pandas vs pandas API on Spark 比較表

ネイティブPandasとpandas API on Sparkの主な違いを表にまとめます。試験ではこの違いが正確に理解できているかが問われます。

DataFrame相互変換のコード例

Pandas DataFrame・Spark DataFrame・pyspark.pandas DataFrameの3種類のDataFrameを相互に変換する方法は試験頻出トピックです。各変換メソッドの違いを正確に覚えましょう。

import pandas as pd
import pyspark.pandas as ps
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# ---- 1. Pandas → pyspark.pandas ----
pandas_df = pd.DataFrame({"name": ["Alice", "Bob"], "age": [30, 25]})
ps_df = ps.from_pandas(pandas_df)

# ---- 2. pyspark.pandas → Pandas ----
# 注意: データ全体がドライバに集約される
pandas_df2 = ps_df.to_pandas()

# ---- 3. Spark DataFrame → pyspark.pandas ----
spark_df = spark.createDataFrame(
    [("Alice", 30), ("Bob", 25)], ["name", "age"]
)
ps_df2 = spark_df.to_pandas_on_spark()
# 別の方法
ps_df2 = ps.DataFrame(spark_df)

# ---- 4. pyspark.pandas → Spark DataFrame ----
spark_df2 = ps_df.to_spark()

# ---- 5. Spark DataFrame → Pandas（参考）----
# 注意: データ全体がドライバに集約される
pandas_df3 = spark_df.toPandas()

# ---- 6. Pandas → Spark DataFrame（参考）----
spark_df3 = spark.createDataFrame(pandas_df)

重要：toPandas()とto_pandas()は データ全体をドライバノードのメモリに集約します。 大規模データセットではOutOfMemoryErrorのリスクがあるため、to_spark()でSpark DataFrameに変換して処理するのが安全です。

制約と非対応API一覧

pandas API on Sparkは約80〜85%のPandas APIをカバーしていますが、分散環境の特性上、以下のAPIや動作は制限されています。

非対応・制限のある主要API

ベストプラクティス

• 大規模データの前処理にpyspark.pandasを使い、 可視化やレポート作成の最終段階のみto_pandas()でローカルに集約する
• apply()よりも組み込み集計関数（sum・mean・count）を優先する
• JOIN・ウィンドウ関数などはSpark DataFrameのAPIが高速なため、to_spark()で変換してから処理する
• sort_values()の結果は後続処理で順序が崩れる可能性があるため、 最終出力の直前でソートする

Pandas UDFとの違い

「Pandas API on Spark」と「Pandas UDF」は名前が似ているため混同されやすいですが、目的と使い方が異なります。

from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import DoubleType
import pandas as pd

# Scalar Pandas UDF: カラム単位でPandas関数を適用
@pandas_udf(DoubleType())
def multiply_by_two(s: pd.Series) -> pd.Series:
    return s * 2

df = spark.table("sales")
result = df.withColumn("doubled", multiply_by_two(df["amount"]))

# Grouped Map: グループ単位でPandas DataFrameを受け取り処理
def normalize(pdf: pd.DataFrame) -> pd.DataFrame:
    pdf["normalized"] = (
        (pdf["value"] - pdf["value"].mean()) / pdf["value"].std()
    )
    return pdf

df.groupby("department").applyInPandas(normalize, schema=output_schema)

試験出題ポイント

pandas API on SparkはSpark Developer AssociateとData Engineer Associate（DEA）の両試験で出題されます。 特に以下の3点が重要です。

1. インポート文の正確な記述

import pyspark.pandas as psが正しいインポート文です。import koalas as ks（旧Koalas）やfrom pyspark import pandasは試験の誤答選択肢として出題されます。

2. DataFrame変換メソッドの使い分け

to_pandas_on_spark()（Spark→pyspark.pandas）とtoPandas()（Spark→ネイティブPandas）の違いは頻出です。 前者は分散状態を維持し、後者はドライバにデータを集約します。

3. 分散環境での制約の理解

「行の順序が保証されない」「ilocが非効率」「一部のPandas関数が未サポート」といった制約が正誤問題で問われます。特にネイティブPandasとの動作の違いについて正確に理解しておく必要があります。

問題で確認

よくある質問（FAQ）