PySpark 4.2、Python UDFがデフォルトでArrowに乗る — 型強制変換225マスのうち132マスが変わる話

• はじめに — 二日前に静かに反転したデフォルト値
• 何が変わったのか — 設定3つ
• どれくらい速くなるのか — 誰も数字を出していません
• 本当の話は型強制変換だ
• 225マスのうち132マス
• 最も危険な18マス — 何も失敗しない
• あなたのUDFはArrowに乗っていないかもしれない — importの順序の罠
• 一緒についてくるもの
• だからどうアップグレードするか
• いつこのデフォルト値をオフにしたまま生きるか
• おわりに
• 参考資料

はじめに — 二日前に静かに反転したデフォルト値

Apache Spark 4.2.0が2026年7月14日にリリースされました。4.xラインの3回目のリリースで、リリースノートの言葉を借りれば1,700件を超えるJiraチケットと250人以上のコントリビューターが入ったバージョンです。見出しは華やかです — GEOMETRY/GEOGRAPHY地理空間型、SQLのCHANGES句によるCDC、NEAREST BYのtop-K結合、Data Source V2トランザクション。

ところが、PySparkでバッチを回している人にとって今夜本当に眠れなくなる項目は、そのリストの中でずっと目立たない一行です。

[SPARK-54555] Enable Arrow-optimized Python UDFs and Arrow-based PySpark IPC by default

デフォルト値をひとつ反転させただけです。新しいAPIも新しい構文もありません。ですがこの一行が、あなたが3年間動かしてきたPython UDFの出力データを変えることがあります。エラーもなく、警告もなくです。

本稿は、そのデフォルト値が正確に何を変えるのかを、Sparkリポジトリに実際にチェックインされている資料だけを頼りに追いかけます。結論から言えば、面白いのは速度ではなく型強制変換(type coercion)です。

何が変わったのか — 設定3つ

まず事実確認から始めます。SPARK-54555はPR #53264として入り、コミットea0a35e065d3が2025年12月2日にマージされました。Jiraチケットのfixバージョンは4.2.0で、このコミットはv4.2.0タグには含まれていますがv4.1.0にはありません。

変わったのはSQLConf.scalaのデフォルト値です。branch-4.1とbranch-4.2の同じファイルを直接比較するとこうなります。

設定                                             4.1     4.2
------------------------------------------------------------
spark.sql.execution.pythonUDF.arrow.enabled     false -> true
spark.sql.execution.pythonUDTF.arrow.enabled    false -> true
spark.sql.execution.arrow.enabled               false -> true

3つ目の項目が少しややこしいのですが、ドキュメントやマイグレーションガイドが言及するspark.sql.execution.arrow.pyspark.enabledは自身のデフォルト値を持たず、fallbackConfとしてspark.sql.execution.arrow.enabledを指しています。つまり実際のソースで反転したのは後者です。効果は同じです — DataFrame.toPandasとpandas/NumPy入力を受け取るSparkSession.createDataFrameが、これでデフォルトでArrowカラム転送を使うようになります。

つまり今回のリリースでは、Python側は3方向で同時にArrowに乗ります。通常のPython UDF、Python UDTF、そしてJVMとPythonの間のカラムデータ交換(IPC)です。

どれくらい速くなるのか — 誰も数字を出していません

ここは正直に書きます。

PR #53264の「Why are the changes needed?」の項目はこう述べています — ArrowのカラムIPCがJVMとPythonの間のスループットを"significantly improves"し、シリアライズ/デシリアライズのオーバーヘッドを減らすと。しかしそのPRにはベンチマーク数値が一つもありません。DatabricksのSpark 4.2紹介ブログも「既存のUDFがコード書き換えなしでより速いカラム経路を使える」とだけ述べ、数値は示していません。

Sparkリポジトリにはsql/core/benchmarks/UDFBenchmark-results.txtというチェックイン済みのベンチマーク結果がありますが、開いてみるとScala UDFのwhole-stage codegenのon/off比較です。ArrowとpickleのPython UDFを比較するものではありません。python/pyspark/sql/tests/pandas/bench_arrow_columnar_udf.pyのようなベンチマークハーネスはリポジトリにありますが、結果ファイルはチェックインされていません。

ですからこの記事では「N倍速くなる」とは言いません。そのような数値は今回のリリースと一緒に公開されたことがありません。方向性としては十分納得できます — 行単位のpickleシリアライズの代わりにカラムバッチを渡すことがなぜ有利なのかは、CatalystとTungsten編で扱ったロジックと同じですし、Polarsがpandasに勝った理由とも同じ系統の話です。ただし、あなたのUDFが何パーセント速くなるかはあなた自身が測るしかありません。それが今回のリリースで確認できるすべてです。

そして結局のところ、このデフォルト値変更で本当に重要なのは速度ではありません。

本当の話は型強制変換だ

Python UDFを使うとき、私たちは戻り値の型を宣言します。

@udf("int")
def f(x):
    return some_value

問題は、some_valueが宣言したintと異なる型だったときに何が起こるかです。pickle方式とArrow方式はこの点でルールが異なります。これは知られた事実で、PySparkのソースにコメントとして書き込まれてさえいました。python/pyspark/sql/udf.pyの_create_py_udf関数の上にあるコメントです。

# Arrow and Pickle have different type coercion rules, so a UDF might have a
# different result with/without Arrow optimization. That's the main reason the
# Arrow optimization for Python UDFs is disabled by default.

面白いのは、この注釈がbranch-4.2にもそのまま残っているという点です。「だからデフォルトでは無効にしてある」と書かれているのに、肝心の4.2では有効になっています。コメントがリリースに追いついていないわけです。しかもこのコメントは強制変換の表がpython/pyspark/sql/tests/udf_type_testsにあると案内していますが、そのパスはbranch-4.2には存在しません。

表は移動していました。実際の場所はpython/pyspark/sql/tests/coercion/で、ここにはCIが検証するゴールデンファイルが入っています。

golden_python_udf_return_type_coercion_vanilla.csv           <- pickle (4.1デフォルト)
golden_python_udf_return_type_coercion_with_arrow.csv        <- Arrow  (4.2デフォルト)
golden_python_udf_return_type_coercion_with_arrow_and_pandas.csv

3つ目のファイルは4.1より前のArrowの挙動(途中でpandas変換をもう一度経由していた時代)です。spark.sql.legacy.execution.pythonUDF.pandas.conversion.enabledが4.2でfalseがデフォルトなので、4.2で生きている表は2つ目のファイルです。ちなみにPySparkの公式ドキュメントは、いまだに強制変換の違いを2023年に開かれたPR #41706という「DO NOT MERGE」なドキュメント用PRに案内していますが、リポジトリ内のゴールデンファイルの方が新しく、CIが実際に守っている資料です。

そこで、この2つのゴールデンファイルを実際にdiffしてみました。

225マスのうち132マス

表は宣言SQL型15種 × Python戻り値15種 = 225マスです。2つのファイルをマス単位で比較すると132マスが異なります。半分を超えています。Xはエラーを意味します。

カテゴリ別に分けるとこうなります。

分類                                     マス数
---------------------------------------------
1. None  -> エラー (静かだったNULLが今は失敗)     99
2. None  -> 値    (静かだったNULLが今はデータ)     18
3. 値    -> エラー (通っていたのが今は失敗)         2
4. エラー -> 値    (失敗していたのが今は値)         7
5. 値    -> 別の値 (出力文字列が変わる)             6
---------------------------------------------
合計                                      132

1番(99マス)が圧倒的に大きいです。pickle Python UDFの設計は「宣言型と合わなければ静かにNoneにする」でした。これがなくなりました。今ではほとんどの不一致が例外になります。これは方向性としては明らかな改善です — 型のミスをNULLに飲み込ませるのは、データパイプラインで最も厄介な種類のバグだからです。しかしうまく動いていたジョブが4.2で突然落ちるなら、十中八九この99マスのどれかです。そして落ちるのが正しいのです。ただ、それをリリースノート一行で知ることになるのが問題なのです。

実際のマスをいくつか選んでみます。値はゴールデンファイルに書かれているとおりです。

宣言型            返したPython値             pickle(4.1)                        Arrow(4.2)
----------------------------------------------------------------------------------------------
int             1.0        (float)       None                               1
int             1970-01-01 (date)        None                               0
bigint          1970-01-01 00:00:00      None                               0
boolean         1          (int)         None                               True
double          1          (Decimal)     None                               1.0
int             'a'        (str)         None                               X  (エラー)
tinyint         True       (bool)        None                               X  (エラー)
binary          'a'        (str)         b'a'                               X  (エラー)
struct<_1:int>  [1]        (list)        Row(_1=1)                          X  (エラー)
date            1          (int)         X  (エラー)                          datetime.date(1970, 1, 2)
string          1970-01-01 (date)        'java.util.GregorianCalendar[...'  '1970-01-01'
string          (1,)       (tuple)       '[Ljava.lang.Object;@<hash>'       '(1,)'
string          bytearray(b'ABC')        '[B@<hash>'                        "bytearray(b'ABC')"

最後の3行は純粋なバグ修正です。pickle経路でdateをstringとして宣言すると、java.util.GregorianCalendar[time=?,...というJVM内部オブジェクトのtoStringがそのままデータに流れ込んでいました。bytearrayは[B@<hash> — Javaバイト配列のidentity hashです。こういうものが本番テーブルに入っていたなら、4.2はそれを'1970-01-01'に直してくれます。ありがたいことですが、そのカラムを読むダウンストリームがあれば、その値も一緒に変わります。

4番(7マス)も注目に値します。dateを宣言して整数1を返すと、pickleではエラーでしたが、Arrowでは1970-01-02になります。epoch + 1日です。論理的には一貫していますが、「エラーで弾かれていたミスが、今はもっともらしい日付になって通過する」ということでもあります。

最も危険な18マス — 何も失敗しない

99マスはやかましく失敗します。やかましいものは管理できます。デプロイして、爆発して、ロールバックするか直せばいいだけです。

問題はカテゴリ2、18マスです。ここでは何も起こりません。例外もなく、警告もなく、ログもありません。ただ昨日NULLだったマスに、今日は値が入っているだけです。

全部並べるとこうなります。

宣言        戻り値                   pickle -> Arrow
--------------------------------------------------
boolean    1        (int)          None -> True
boolean    1.0      (float)        None -> True
tinyint    1.0      (float)        None -> 1
tinyint    1        (Decimal)      None -> 1
smallint   1.0      (float)        None -> 1
smallint   1        (Decimal)      None -> 1
int        1970-01-01 (date)       None -> 0
int        1.0      (float)        None -> 1
int        1        (Decimal)      None -> 1
bigint     1970-01-01 00:00:00     None -> 0
bigint     1.0      (float)        None -> 1
bigint     1        (Decimal)      None -> 1
float      True     (bool)         None -> 1.0
float      1        (int)          None -> 1.0
float      1        (Decimal)      None -> 1.0
double     True     (bool)         None -> 1.0
double     1        (int)          None -> 1.0
double     1        (Decimal)      None -> 1.0

このリストを見るとパターンが見えます。ほとんどが「数値型を宣言したのに別の数値型を返している場合」です。そしてこれは実務でいちばん間違えやすい組み合わせです。@udf("int")を付けておいて中で割り算をすると、Pythonはfloatを返します。@udf("double")を付けてDecimalを返すのもよくあることです。

4.1まではこうしたUDFはすべてNULL列を吐いていました。誰も気づいていなければ、そのNULLはすでにダウンストリームに溶け込んでいます。COUNT(*)、WHERE x IS NULLフィルタ、AVG集計、結合キー、ダッシュボードの数字 — 全部です。4.2に上げた瞬間、それらのマスは実際の値で埋まり、そのすべてが静かに動きます。

これは回帰ではありません。4.2側が正しいのです。intを宣言して1.0を返したなら1が出るのが正しく、NULLが正しいはずはありません。しかし、「直ること」と「出力が変わること」は運用上は同じ出来事です。後者として扱わなければ痛い目を見ます。特に再処理(backfill)を回すとき、過去のバッチと新しいバッチで結果が変わる形でです。

あなたのUDFはArrowに乗っていないかもしれない — importの順序の罠

ここで本当に皮肉な部分が出てきます。このデフォルト値は均一には適用されません。

python/pyspark/sql/udf.pyの_create_py_udfを見ると、こう判定しています。

is_arrow_enabled = False
if useArrow is None:
    from pyspark.sql import SparkSession

    session = SparkSession._instantiatedSession
    is_arrow_enabled = (
        False
        if session is None
        else session.conf.get("spark.sql.execution.pythonUDF.arrow.enabled") == "true"
    )
else:
    is_arrow_enabled = useArrow

SparkSession._instantiatedSessionがNoneなら、設定を読みにも行かずにFalseです。そしてこの判定はUDFを呼び出すときではなく、UDFオブジェクトを作るときに一度だけ行われ、eval typeとして固定されます。

つまり、モジュールの先頭にこう書いたUDFは、

# udfs.py — import時点でデコレータが評価される
from pyspark.sql.functions import udf

@udf("int")
def parse_score(s):
    return float(s) / 2

SparkSession.builder.getOrCreate()より先にimportされると、まだセッションが存在しないためpickle UDFとして固定されます。4.2でもです。同じコードをセッション生成後にimportすればArrow UDFになります。

つまり、一つのジョブの中でもUDFごとに強制変換ルールが異なりえます。import順序によってです。上の例はintを宣言してfloatを返しているので、まさにあの18マスのうちの一つ(int + 1.0(float))に該当します — import順序によってNULLになるか値になるかが変わります。

このロジック自体は新しいものではありません。branch-4.1のudf.pyにも、一字一句同じコードがあります。ただ4.1では設定のデフォルトがそもそもfalseだったので、session is None分岐が同じ答えを出していただけです。4.2でデフォルト値が反転したことで、この分岐は今や新しいデフォルト値を静かに無効化する経路になりました。

似たような静かな逸脱がもう一つあります。同じ関数で、PyArrowやpandasがない場合はこう処理します。

except ImportError:
    is_arrow_enabled = False
    warnings.warn(
        "Arrow optimization failed to enable because PyArrow or Pandas is not installed. "
        "Falling back to a non-Arrow-optimized UDF.",
        RuntimeWarning,
    )

RuntimeWarningを一つ出してpickleに戻ります。ドライバイメージにはPyArrowがあり、他の環境にはないとしたら、同じコードが環境によって異なる強制変換ルールで動くことになります。ちなみに4.2はPyArrowの最小バージョンを15.0.0から18.0.0に引き上げました。pandasの最小は2.2.0のままです。

もう一つ触れておくと、このフォールバックはUDFにはありません。spark.sql.execution.arrow.pyspark.fallback.enabled(デフォルトtrue)という設定があり、「Arrowがエラーになったら非最適化実装に自動フォールバックする」を行いますが、設定のドキュメントを読むと、これはspark.sql.execution.arrow.pyspark.enabledが有効にする最適化、つまりtoPandasとcreateDataFrameの経路にしか適用されません。UDF実行中に発生した強制変換エラーはフォールバックされず、そのままジョブを殺します。

一緒についてくるもの

4.2のマイグレーションガイドには、Python側の変更がもう少しあります。Arrowのデフォルト値とは別ですが、同じアップグレードにまとめて付いてきます。

• pandas UDFのnullable整数列。 バッチにnullが混ざったnullable整数列が、今ではfloat64ではなくpandasのnullable整数拡張dtype(Int8/Int16/Int32/Int64)で渡されます。float64入力を前提に書いたUDFコードは手直しが必要です。
• PyPyサポート終了。 4.2からPyPyは公式サポート対象外です。CPythonを使うよう案内されています。
• NumPy ndarrayからのcreateDataFrame。 今ではpandasではなくPyArrowを要求し、pandasを経由せずArrowへ直接変換します。Arrowを切ったままNumPy dtypeベースのスキーマ推論に頼っていた場合、推論されたスキーマを再確認するようガイドが明示的に警告しています。
• Python Data Sourceのスキーマ不一致。 返したArrowデータのカラム型が宣言スキーマと異なると、DATA_SOURCE_RETURN_SCHEMA_MISMATCHで失敗します。以前はカラム数と名前の不一致だけがこのエラーを出していました。

Arrowバッチサイズはspark.sql.execution.arrow.maxRecordsPerBatchで調整でき、デフォルト値は10000です。UDFの中で行あたりのメモリを多く使うなら、この値は今やexecutorのメモリプロファイルに影響します — 以前は行単位で流れていたものが、今はバッチにまとまって来るからです。

だからどうアップグレードするか

まとめると、こうするのが合理的です。

1. まずオフのままアップグレードしてください。 4.2へ上がることとArrow UDFへ乗り換えることは別々の変更です。一度にやらないでください。

spark.sql.execution.pythonUDF.arrow.enabled   false
spark.sql.execution.pythonUDTF.arrow.enabled  false
spark.sql.execution.arrow.pyspark.enabled     false

これで4.2の残り(地理空間型、CDC、DSv2トランザクション、Java 25)を4.1と同じPythonセマンティクスのまま受け取れます。

2. 次にUDFの棚卸しをしてください。 実際に危険なのは「宣言型と戻り値の型が異なるUDF」だけです。両者が一致していれば、132マスはあなたには無関係です。ゴールデンテーブルの対角線(宣言型 = 戻り値型)は2つの表で同じです。つまり問いは一つだけです — 自分のUDFは宣言どおりに値を返しているか?

これは目で読んでわかることではありません。数値型UDFの中の割り算一つ、Decimal一つで済んでしまうからです。小さなサンプルに対して両方の設定で同じジョブを回し、出力を比較するのが唯一信頼できる方法です。

3. 比較するときはNULLの個数を見てください。 18マスのカテゴリは静かなので、エラーログを見ても絶対に捕まりません。カラムごとのNULLカウントが減っていれば、それがまさにその18マスです。減ることが歓迎すべき状況かもしれませんが、それはあなたが判断することであって、リリースが代わりに判断することではありません。

4. import順序を固定してください。 先の罠のせいで、「設定をオンにした」というだけではArrowが有効になっているとは確信できません。確実にするにはudf(..., useArrow=True)と明示すればいいです。useArrowが明示されると、セッションの有無や設定を両方無視してその値が使われます。逆に確実にpickleのまま残したいUDFはuseArrow=Falseで固定してください。デフォルト任せの曖昧さがいちばん悪いのです。

いつこのデフォルト値をオフにしたまま生きるか

正直なところ、オフのままにし続けるのが正しい場合があります。

• UDFが少数で、すべて軽い場合。 Arrowの利益はJVMとPythonの間を行き来するデータが多いときに出ます。1日数万行を処理するジョブなら、再検証のコストが利益を上回ります。
• 宣言型と戻り値の型がずれているUDFが多いが、今それを全部監査する余力がない場合。 132マスがあなたの話です。オフにして、スケジュールを立てて、後でやりましょう。
• 出力が契約で縛られている場合。 ダウンストリームがスキーマだけでなく値の分布まで期待しているなら(規制レポートや財務締めバッチなど)、静かなNULLの減少それ自体が事故です。先に合意してからオンにしましょう。

逆に、今すぐオンにする価値がある場合はこうです。

• UDFが重く、データ量が多い。それならカラム経路が価値を発揮する可能性が高いです — ただし前述のとおり、数字は自分で測る必要があります。
• 宣言型と戻り値の型が一致することをテストで保証できる。それなら132マスは無関係で、ただの無料の恩恵です。
• そもそもUDFを使わないのが最善という場合も忘れないでください。組み込み関数やSQLで表現できるロジックをPython UDFで書いていたなら、Arrowで速くなることよりUDFをなくすことの方がはるかに大きな利益です。4.2はtime_bucketを含め、組み込み関数をさらに増やしました。

おわりに

Spark 4.2.0のPythonの話は、リリースノート一行では伝わりません。「Arrow-optimized Python UDFs enabled by default」はパフォーマンス項目のように読めますが、リポジトリにチェックインされたゴールデンテーブルを実際に開くと、225マスのうち132マスの意味論が変わります。そのうち99マスは静かだったNULLが今はエラーになり、18マスは何の合図もなくNULLが値に変わります。

Apacheのマイグレーションガイドはこれを正直に書いています — 設定名と元に戻す方法まで。一方、ベンダーブログは「コード書き換えなしでより速いカラム経路」とだけ紹介し、強制変換の話は出しません。どちらも嘘ではありませんが、アップグレード計画を立てるときに読むべきは前者です。

そして、この変更の根底にある判断そのものは正しいと思います。「型が合わなければ静かにNULL」は、そもそも擁護できないデフォルト値でした。ただ、その誤ったデフォルト値の上で3年ほど動いてきたパイプラインがあり、そうしたパイプラインにとって、この修正は修正としてではなく出力の変更として届きます。オフにしてアップグレードし、測ってください。それからオンにしてください。順序がすべてです。

参考資料

• Spark Release 4.2.0 — 公式リリースノート (2026-07-14)
• Apache Spark News — リリース日の確認
• SPARK-54555 — Enable Arrow-optimized Python UDFs by default (Jira)
• PR #53264 — SPARK-54555の実装PR
• Upgrading PySpark 4.1 → 4.2 — 公式マイグレーションガイド (branch-4.2)
• 強制変換ゴールデンファイル — python/pyspark/sql/tests/coercion (branch-4.2)
• udf.py — _create_py_udfのセッション/Arrow判定ロジック (branch-4.2)
• SQLConf.scala — 設定デフォルト値の原本 (branch-4.2)
• Introducing Apache Spark 4.2 — Databricks (ベンダー発表)
• Apache Arrow Python UDFs — PySpark公式ドキュメント