2023年にリリースされたApache Spark 3.4と3.5で、私たちはPySparkのパフォーマンス、柔軟性、使いやすさの改善に重点を置きました。 このブログ記事では、主な改善点をご紹介します。
2023年にApache Spark 3.4と3.5で追加された最も重要な機能の概要です：
次のセクションでは、これらの各項目について検証し、さらに注目すべき改善点をいくつかご紹介します。
Spark ConnectはApache Spark 3.4でデビューし、クライアントとサーバーを分離したアーキテクチャを導入することで、任意の場所で実行されているアプリケーションからSparkクラスタへのリモート接続を可能にしました。 このようにクライアントとサーバーを分離することで、最新のデータアプリケーション、IDE、ノートブック、プログラミング言語がインタラクティブにSparkにアクセスできるようになります。 さらに、非結合アーキテクチャにより、安定性、アップグレード性、デバッグ性、可観測性が向上します。
Apache Spark 3.5ではScalaのサポートが完了し、Structured Streaming (SPARK-42938)、ML と PyTorch (SPARK-42471)、Pandas API on Spark (SPARK-42497) などの主要な Spark コンポーネントのサポートも完了しました。
Databricks Connectを使用してDatabricks上のSpark ConnectまたはApache Spark用のSpark Connectを直接開始します。
Arrow-optimized Python UDFs (SPARK-40307)は、Arrowのカラムナーフォーマットを活用することで、パフォーマンスの大幅な最適化を可能にします。 例えば、同じクラスタ内でUDFを連鎖させた場合、32GBのデータセットでArrow最適化Python UDFはpickle Python UDFよりも1.9倍高速に実行されます。
Apache Spark 3.5では、PySparkのUDFサポート をユーザー定義テーブル関数で拡張しました。 一度登録されると、SQLクエリのFROM句に表示することができます。 例えば、以下のUDTF SquareNumbers は、入力とその2乗値を表として出力します：
from pyspark.sql.functions import udtf
@udtf(returnType="num: int, squared: int")
class SquareNumbers:
def eval(self, start: int, end: int):
for num in range(start, end + 1):
yield (num, num * num)from pyspark.sql.functions import lit
SquareNumbers(lit(1), lit(3)).show()
+---+-------+
|num|squared|
+---+-------+
| 1| 1|
| 2| 4|
| 3| 9|
+---+-------+PySparkの大きな利点の1つは、Spark SQLがPySpark DataFramesとシームレスに動作することです。 2023年にSpark SQLは多くの新機能を導入し、PySparkはspark.sqlを介して、GROUP BY ALL および ORDER BY ALL、一般的なテーブル値関数のサポート、INSERT BY NAME、PIVOTおよびMELT、ANSI コンプライアンスなどを直接活用できるようになりました。 以下はGROUP BY ALLと ORDER BY ALLの使用例です：
spark.sql("""
SELECT name, firstname, level, sum(comp) as totalcomp
FROM {table}
GROUP BY ALL
ORDER BY ALL
""", table=df)Structured StreamingにおけるPythonの任意のステートフル操作は、ストリーミングクエリの実行にまたがるステート処理を可能にすることで、PySparkにおける膨大な数のリアルタイム分析や機械学習のユースケースの障害を取り除きます。 次の例は、任意のステートフル処理を示しています：
# Group the data by word, and compute the count of each group
output_schema = "session STRING, count LONG"
state_schema = "count LONG"
sessions = events.groupBy(events["session"]).applyInPandasWithState(
func,
output_schema,
state_schema,
"append",
GroupStateTimeout.ProcessingTimeTimeout,
)
# Start running the query that prints the windowed word counts to the console
query = sessions.writeStream.foreachBatch(
lambda df, _: df.show()).start()TorchDistributorはPySparkでPyTorchのネイティブサポートを提供し、Sparkクラスタ上でのディープラーニングモデルの分散学習を可能にします。 PyTorchのプロセスを開始し、配布の仕組みはPyTorchに任せ、プロセスが調整されていることを確認するためだけに機能します。
TorchDistributorの使い方は簡単で、いくつかの主な設定があります：
from pyspark.ml.torch.distributor import TorchDistributor
model = TorchDistributor(
num_processes=2,
local_mode=True,
use_gpu=True,
).run(<function_or_script>, <args>)pyspark.testing パッケージの新しいテスト API (SPARK-44042) は、PySpark アプリケーションをテストする開発者に大きな機能拡張をもたらします。詳細なエラーメッセージを含む、等式テストのためのユーティリティ関数を提供し、DataFrameスキーマとデータの不一致の特定を容易にします。 下の出力例はそれを示しています：
*** actual ***
Row(name='Amy', languages=['C++', 'Rust'])
! Row(name='Jane', languages=['Scala', 'SQL', 'Java'])
*** expected ***
Row(name='Amy', languages=['C++', 'Rust'])
! Row(name='Jane', languages=['Scala', 'Java'])English SDK for Apache Sparkは、ユーザーが平易な英語で指示を入力し、それをPySparkとSpark SQLのコードに変換することで、Apache Sparkの使用を簡素化します。 これにより、特にDataFrameの変換操作、データの取り込み、UDFに関連するコードでPySparkプログラミングがより身近になり、キャッシュのおかげで生産性がさらに向上します。 English SDKは、開発プロセスを合理化し、コードの複雑さを最小限に抑え、Sparkコミュニティの裾野を広げる大きな可能性を秘めています。 ご自身で試してみてください！
Apache Spark 3.4と3.5で導入されたその他の機能のうち、まだご存知でなければ調べてみると良いでしょう：
2023年、オープンソースコミュニティの活発なイノベーションにより、PySparkとApache Sparkの両方が大幅に強化され、データ専門家が利用できるツールキットの幅が広がり、分析ワークフローが合理化されました。 Apache Spark 4.0の登場を目前に控え、PySparkは新機能と性能の向上を通じてデータ処理にさらなる革命を起こす態勢を整え、データエンジニアリングとデータサイエンスのコミュニティにおけるデータ分析の推進へのコミットメントを再確認しています。
この投稿では、PySparkの使いやすさ、パフォーマンス、柔軟性を強化するために2023年にApache Spark 3.4と3.5で行われた最も重要な改善の概要を簡単に説明しました。 これらの機能はすべてDatabricks Runtime 13および14で利用可能です。
