Spark 4.0では何が変わるのか? (後編)

2024年8月12日 08:13

前回の続きの続きです。
前編はこちら
中編はこちら

後編では、以下の内容を紹介します。

Python UDTFの改善

spark 3.5から導入された機能として、python UDTF (User Defined Table Functions) があります。入力として、0個の以上の引数を受け取ります。
出力結果として、テーブルを返すことができます。
UDTFは、pythoh classとして実装する必要がありますが、利用する際には、DataFrame APIから呼びさすこともできますが、SQLの中でFROM句で利用することもできます。

DataFrame APIからの呼び出し

from pyspark.sql.functions import lit, udtf

# Create a UDTF using the class definition and the `udtf` function.
square_num = udtf(SquareNumbers, returnType="num: int, squared: int")

# Invoke the UDTF in PySpark.
square_num(lit(1), lit(3)).show()
# +---+-------+
# |num|squared|
# +---+-------+
# |  1|      1|
# |  2|      4|
# |  3|      9|
# +---+-------+

SQLからの呼び出し

from pyspark.sql.functions import udtf

@udtf(returnType="word: string")
class WordSplitter:
    def eval(self, text: str):
        for word in text.split(" "):
            yield (word.strip(),)

# Register the UDTF for use in Spark SQL.
spark.udtf.register("split_words", WordSplitter)

# Example: Using the UDTF in SQL.
spark.sql("SELECT * FROM split_words('hello world')").show()
# +-----+
# | word|
# +-----+
# |hello|
# |world|
# +-----+

詳しくは、以下にサンプルの記載があります。

上記のUDTFですが、spark 4.0では次の改善がありました。

Polymorphic Analysis

実行時の引数により、出力のスキーマを動的に計算することに対応。analyze static methodをpython UDTF classに実装することで、引数を解析し、出力するデータのスキーマを動的に決めることが可能になりました。
- https://issues.apache.org/jira/browse/SPARK-44380

Input Table Partitioning

UDTFの入力データに、partition句を指定し、入力データをpartitionが毎に分割して評価することができます。以下、抜粋です。

Variable Keyword Arguments

UDTFで、任意の数の位置引数(*args)、任意の数のキーワード引数(**kwags)を受け取れるようになります。
- https://issues.apache.org/jira/browse/SPARK-44749

python UDTFの詳しい使い方については、ユーザガイドが参考になります。

Python UDF with Apache Arrow の性能改善

spark 3.5以降では、Apache Arrowに最適化されたpython UDFがサポートされました。データのシリアライズおよびデシリアライズ速度が向上しました。Spark 4.0以降でデフォルトで有効となりました。

詳しいベンチマークの結果は、以下が参考になりました。

https://www.databricks.com/wp-content/uploads/notebooks/python-udf-benchmark.html

またApache ArrowベースのUDFでは、データ出力の型を強制するため、出力結果が曖昧になる問題に対して対処しています。
※Apache Arrowについては別の機会に纏めたいと思います。

その他

Pandas 2 support

spark 4.0では、PySparkでサポートされるpandasの最小バージョンが1.0.5から2.0.0に引き上げられます。それに伴い、互換性のないAPIに対する変更が数多く行われています。spark 3系で、pandas APIを使用している場合は特に注意が必要です。

https://github.com/apache/spark/blob/master/python/docs/source/migration_guide/pyspark_upgrade.rst#upgrading-from-pyspark-35-to-40

以上です。今後の参考になれば幸いです。Spark 4系を早く本格的に使いたいですね。