Apache Spark® 4.3 のご紹介 - Databricks Runtime 19 で利用可能になりました

はじめに

Apache Spark 4.3では、リアルタイムモードにステートフル処理が導入され、代表的な3つのステートフルベンチマークにおいて、P99エンドツーエンドの遅延が150ミリ秒未満を達成しました。ストリーミング集約、重複排除、およびカスタムステートフル処理がStructured Streamingの低遅延実行モードで実行できるようになり、イベント間でコンテキストを保持する必要があるアプリケーションへと用途が広がります。

その影響は、パイプラインの高速化にとどまりません。サポートされているワークロードでは、ミリ秒単位の遅延を実現するために専用のストリーミングエンジンを別途用意する必要がなくなります。不正検知から、継続的に更新されるユーザーアクティビティ、デバイスの監視に至るまで、開発者はバッチ処理、アナリティクス、AIで使用するおなじみのオープンエンジン上で、慣れ親しんだStructured Streaming APIを使ってステートフルアプリケーションを構築できます。

ステートフルなリアルタイムストリーミングはSpark 4.3の目玉機能ですが、機能改善はエンジン全体におよんでいます。このリリースでは、主に5つのメリットを提供します。

• ステートフルなリアルタイムアプリケーションの構築: リアルタイムモードでストリーミング集約、重複排除、transformWithStateを使用し、イベントの到着に合わせてコンテキストを維持します。
• SCD Type 2 Auto CDCによる履歴の保持: Python、SQL、Spark Connectを使用して、カスタムの変更処理ロジックを減らしながら、レコードの経時的な変化を追跡します。
• 小規模データでの素早いイテレーション: Project Featherは不要な実行オーバーヘッドを削減し、Arrowネイティブキャッシングによってより効率的なデータの再利用をサポートします。
• Sparkへのアプリケーション接続の拡大: Spark Connectの可観測性、JDBC統合、Pythonの機能改善により、リモート実行の開発と運用がより簡単になります。
• SQLでの高度な分析表現: VARIANT変換、JSON_TABLE、テンポラル結合、およびナノ秒タイムスタンプのプレビューにより、ネストされたデータや時系列データのネイティブ処理が拡張されます。

これらの変更により、開発者は1つのオープンエンジンを使用して、データの探索、変化するコンテキストの維持、および分析やAIアプリケーションの駆動を効率的に行うことができます。

また、Spark 4.3からは四半期ごとの機能リリースサイクルが導入され、これまでの半年に1回から3か月に1回に変更されます。この短いサイクルにより、完成した改善をより早くユーザーの皆様にお届けできるようになり、コミュニティがフィードバックを提供して次の開発を形作る機会が増えます。

リアルタイムストリーミング：ミリ秒単位の遅延を実現するステートフルアプリケーション

イベント処理からコンテキストの維持へ

不正検知のシグナルは、最新のトランザクションだけでなく、その前に行われたアクティビティにも依存します。パーソナライズ機能では、ユーザーの最近の行動を反映させる必要があります。デバイス監視では、新しい読み取り値が届くたびにカウンターや集約値を更新する必要があります。これらのアプリケーションには、低遅延と状態（ステート）の両方が求められます。

Spark 4.3では、こうしたステートフル処理パターンにリアルタイムモードが拡張され、完全なストリーミングエンジンを実現しています。ストリーミング集約、dropDuplicates、およびScala/JavaのtransformWithState（SPARK-58635、SPARK-57228）のサポートが追加されました。開発者は、単に遅延を短縮する目的だけで別のエンジンを導入することなく、Structured Streamingの既存のプログラミングモデルを使用して、キー付きステートの維持や結果の更新を行うことができます。

3つのステートフルベンチマークで150ミリ秒未満のP99遅延を達成

SPARK-58884で報告されているベンチマークでは、代表的な3つのステートフルワークロードすべてにおいて、P99エンドツーエンドの遅延が150ミリ秒未満を達成しました。

これらの結果は、重複排除、ウィンドウ集約、およびカスタムステートフル処理における低遅延実行を示しています。これらはテストされた3つのワークロードの測定値であり、すべてのパイプラインでの遅延を保証するものではありません。

使い慣れたAPI、低遅延な実行

リアルタイムモードは、別のプログラミングモデルを新たに導入するのではなく、Structured Streamingを基盤として構築されています。ステートフルオペレーターは、耐久性とリカバリのために、Sparkの既存のステートストアとチェックポイントメカニズムを引き続き使用します。

例：各デバイスの実行中のカウントを維持します。KafkaをバックエンドとするストリーミングDataFrame eventsがあり、device_id列があるとします。

デバイス A、A、B のイベントが到着した場合、Sparkはステートを維持し、カウントを A=2 および B=1 に更新します。さらに A のイベントが到着すると、そのステートは A=3 に更新されます。

継続的なデータフローのために構築された実行エンジン

ステートフルなリアルタイムモードをサポートするには、Sparkがステージをスケジューリングおよび接続する方法を根本的に変更する必要がありました。従来のブロッキングシャッフルでは、ダウンストリームのステージが開始する前に、プロデュース側のステージが完了している必要がありました。そのモデルでは、長時間実行されるストリーミングタスク間の継続的なデータフローをサポートできません。

Spark 4.3では、ネイティブな同時ステージスケジューリングとパイプライン化されたシャッフルが導入され、プロデューサーとコンシューマーのステージを同時に実行できるようになりました（SPARK-54699、SPARK-58076、SPARK-56664）。ストリーミングシャッフルはこの機能をリアルタイムモードで実現し、プロデュース側のステージ全体の完了を待つ代わりに、各キーを担当するダウンストリームタスクへレコードをインクリメンタルに転送します。

概念的には、実行フローが次のように変化します。

プロデューサーのステージ → シャッフルのマテリアライズ → コンシューマーのステージ

から、次のように変わります。

プロデューサーのステージ ⇄ ストリーミングシャッフル ⇄ コンシューマーのステージ

Spark 4.3では、プロデューサーの実行中であってもコンシューマーが読み取りを開始できるようにする新しい種類のシャッフル依存関係である PipelinedShuffleDependency が追加されました（SPARK-58185）。スケジューラーは、これらの依存関係によって接続されたステージをパイプライン化されたグループ（協調的なスケジューリング単位）として扱います。そのステージは一緒にスケジュールされ、1つのステージで障害が発生するとグループ全体が失敗します（SPARK-58263、SPARK-58398）。つまり、一緒に実行され、一緒に失敗します。

パイプライン化されたグループ内のすべてのステージが同時に実行されるため、Sparkは、すべてのタスクに十分なタスクスロットが利用可能な場合にのみ、そのグループを受け入れます。そうでない場合、ハングアップするのではなく、早期に処理を失敗させます。

Sparkは、依存関係のタイプに基づいて各シャッフルをシャッフルマネージャーにルーティングします。

• 通常の ShuffleDependency はブロッキングシャッフルマネージャーを使用します。これにより、ダウンストリームのステージが読み取る前に、すべてのシャッフル出力がマテリアライズされます。SortShuffleManager は組み込みの実装です。
• PipelinedShuffleDependency は、パイプライン化されたシャッフルマネージャーを使用し、生成された出力を即座に使用できるようにします。StreamingShuffleManager は、組み込みの実装です (SPARK-57141)。

依存関係のタイプは実行契約を定義し、シャッフルマネージャーはデータ転送メカニズムを提供します。スケジューラーは、どのワークロードがそれらを作成したかを把握する必要なく、パイプライン化されたステージを調整します。

この基盤はリアルタイムモードを超えて拡張されます。同時ステージのスケジューリング、パイプライン化されたデータ交換、および調整された実行は、単発のストリーミングインフラストラクチャではなく、汎用的な Spark エンジンの機能です。Spark 4.3 では、低レイテンシーのステートフル処理が可能になります。長期的には、従来のステージ単位の実行モデルを超える必要のある他のワークロードの基盤を提供します。

リアルタイムモードのデザインとユースケースの背景については、リアルタイムモードの紹介をご覧ください。その記事では Databricks の実装について説明しており、ここで議論されている Spark 4.3 の機能スコープとベンチマーク結果は上記にリストされているものです。

Project Feather: 小規模データでのより高速なイテレーション

LLM を活用した開発では、通常、変換の生成、サンプルでの実行、結果の検査、コードの洗練という短いフィードバックループが続きます。各イテレーションでは小さなファイルや数行のみが処理される場合がありますが、タスクのスケジューリングやシャッフルには、計算そのものよりもコストがかかることがあります。

Project Feather は、この固定された実行オーバーヘッドをターゲットにしており、小規模なデータセットでの SQL および DataFrame コードの探索と検証において、Spark の応答性を高めます。開発者は、コードを自分で書く場合でも AI アシスタントと連携する場合でも、サンプルから大規模なワークロードへ移行する際に同じ API を使用できます。

Spark 4.3 では、適格な小規模の単一パーティションクエリフラグメントに対して、オプトイン方式のシャッフルなし単一タスク実行が導入されています (SPARK-57851)。適格なスキャンがその出力がすでに 1 つのパーティション内にあることを報告すると、サポートされている下流の操作は不要なシャッフルを回避できます。

例: 小規模な集計を実行して検査します。単一のファイルとして保存され、カテゴリ列を持つ小さな Parquet データセットを使用します。

6 つのケース:

category, Exchange, SPARK_LOCAL_CONNECT_REUSE=1, zip, counter_diff and jaro_winkler_similarity.

結果はカテゴリごとのカウントです。適格なクエリの場合、物理プランでは、それ以外の場合にシャッフルを導入する `Exchange` を省略できます。これにより、変換が開発者によって書かれたものか、コーディングアシスタントによって提案されたものかに関係なく、同じ実行と検査のワークロードがサポートされます。

デフォルトの適格性制限では、適格なファイルスキャンは単一のパーティション内に収まる 1 つの空でない小さなファイルを読み取り、適格なローカルインメモリリレーションは最大 1,000 行を含みます。サポートされるクエリの形状には、選択されたスキャン、集計、distinct 操作、ウィンドウ、ソート、および制限が含まれます。

また、Feather はオプションの Arrow ネイティブのデータセットキャッシュを追加し (SPARK-57268)、圧縮と列統計を備えた Arrow 形式でキャッシュされたバッチを保存します。Arrow ベースの入力は不要な変換を回避でき、同じデータを繰り返し探索するワークロードをサポートします。単一タスクの最適化と Arrow キャッシュの両方がオプトインです。有効にするには、新しい Spark アプリケーションの起動時に静的設定 spark.sql.cache.serializer=org.apache.spark.sql.execution.columnar.ArrowCachedBatchSerializer を設定します。より広範なプロセス内ローカル再パーティショニングは、次のリリースに向けた後続の作業です (SPARK-57399)。

これらの改善は始まりにすぎません。Spark コミュニティは、将来のリリースでも小規模データのパフォーマンスへの投資を続け、実行オーバーヘッドを削減し、対話型探索や LLM 支援開発のフィードバックループを短縮していきます。

オープンソースの Auto CDC における SCD Type 2: データ変更時の履歴の保持

現在状態のテーブルからは顧客が現在どこに住んでいるかはわかりますが、注文が行われたときにどの地域に属していたかはわかりません。レポート、監査、および履歴分析では、多くの場合その両方が必要になります。Spark 4.3 では、Auto CDC にエンドツーエンドの Slowly Changing Dimension (SCD) Type 2 のサポートが追加され (SPARK-56249)、パイプラインはすべての変更で最新の値を上書きするのではなく、すべてのレコードの履歴を追跡できるようになります。この機能は、Python、SQL、および Spark Connect を通じて利用でき (SPARK-58321、SPARK-58247、SPARK-58261)、ユーザーは手動のマージ操作の代わりに変更を自動的に適用できます。

例: 顧客のステータスと地域を時間経過とともに追跡します。顧客 ID、イベント時刻、ステータス、および地域を持つ customer_changes パイプラインソースが与えられた場合:

あるいは、これを SQL で表現することもできます:

SCD-2 出力テーブルでは、すべての固有キーが一連のレコードで表され、各レコードは 2 つの追加メタデータ列 (__START_AT と __END_AT) を使用して有効性の開始と終了を示します。たとえば、変更レコードが ID 17 の顧客が地域 West から East に移動したことを示している場合、パイプラインは前の地域 West を持つレコードを保持し、その終了を変更レコードのイベント時刻としてマークします。また、地域 East とレコードのイベント時刻を開始時刻とする新しいレコードを挿入します。その後、分析ではイベント時刻シーケンス内の関連する時点で適用された地域を使用できます。

内部的には、Auto CDC は宣言型フロー定義をターゲットスキーマの導出と専用の SCD Type 2 ストリーム処理に接続します (SPARK-58319、SPARK-57395、SPARK-58321)。キーによってエンティティが特定され、シーケンスによって変更の順序が決定され、履歴追跡列によってどの属性変更が履歴バージョンを作成するかを指定します。また、エンジンは予約済みの列、シーケンスタイプ、および履歴追跡設定を検証し (SPARK-57251、SPARK-58313、SPARK-58391)、より狭いマイクロバッチを進化したターゲットスキーマと調整します (SPARK-58418)。

Auto CDC は MERGE を使用して各マイクロバッチをターゲットに適用するため、ターゲットテーブルは、コネクタが Data Source V2 の行レベルの操作を実装しているカタログ内にある必要があります。Parquet を含む Spark の組み込みファイル形式は対象外です。そのため、一般的な選択肢は Iceberg です。

Spark Connect: 開発、監視、および運用の簡素化

アプリケーションや AI エージェントは、独自のプロセスから Spark を呼び出し、セッションを管理し、失敗したリクエストをサーバーで実行されている作業に接続する必要があります。Spark 4.3 では、セッションと操作の可視性、リクエストの属性付け、およびより幅広いクライアントサポートにより、これらのワークロード向けの Spark Connect が強化されています。

Spark Connect は、DataFrame 操作を論理プランとして gRPC 経由で送信し、サーバー側の Spark エンジンで実行し、結果を Arrow バッチでストリーミングします。Spark 4.3 では、Connect のセッションや操作を一覧表示および検査するための REST エンドポイントが追加されたほか、クライアントとサーバー間の実行を追跡するための操作 ID が追加されました (SPARK-57941、SPARK-58547)。POSIX システムでのローカルな PySpark 開発向けに、実験的な `SPARK_LOCAL_CONNECT_REUSE=1` オプションを使用すると、実行をまたいで常時稼働する Connect サーバーを再利用できるため、繰り返し発生する起動オーバーヘッドを回避できます (SPARK-57787)。また、PySpark の SQLContext 互換性ラッパーも Spark Connect で動作します (SPARK-57021)。

Spark Connect JDBC ドライバーには、ARRAY、MAP、STRUCT の結果サポートとより高度な結果セット処理 (SPARK-57380、SPARK-57274)、SQL 型のメタデータとより互換性の高いスキーマイントロスペクション (SPARK-57627)、および例外処理の改善 (SPARK-57933、SPARK-58619) が追加されています。API カバレッジも拡張され、DataFrame.zip、SCD Type 2 Auto CDC、およびナノ秒タイムスタンプ型とリテラルがサポートされます (SPARK-57247、SPARK-58247、SPARK-57160、SPARK-57161)。

これらとは別に、コミュニティによる取り組みとして、セッション認識ルーティングとマルチテナント操作のための Spark Connect Gateway (SPARK-58455)、別のクライアント実装パスとしての Rust および Rust 製の Python クライアント (SPARK-58960)、およびクエリを送信する前にサポートされているサーバー操作を特定するための機能検出提案 (SPARK-56398) が進められています。これらの開発タイムラインはそれぞれ異なり、バージョン 4.3 の機能には含まれません。

PySpark: 充実した API と優れた Python 実行環境

Python のワークフローでは、派生カラムの作成を別々のステップで行ってから結果を結合し、pandas、Arrow、Spark の間でデータを移動することがよくあります。Spark 4.3 では、基盤となる実行パスを改善しながら、DataFrame API を通じてこれらの処理をより多く利用できるようになりました。DataFrame.zip を使用すると、Spark Connect を使用する場合も含め、RDD API に移行することなく、互換性のあるカラムの射影を結合する方法が提供されます (SPARK-55886、SPARK-57247)。

例: 識別子と派生値を結合します。同じ入力から 2 つの射影を作成し、それらのカラムをまとめます:

結果は、id=1 と doubled=20、および id=2 と doubled=40 をペアにします。`zip` は、同じ基底プランに対する射影のみの変換を結合します。行の順序を維持するためには、それらの射影の下にあるフィルタリング、ソート、結合、または集約処理が両側で完全に一致している必要があります。

広範な実行の最適化は、データ転送、バッチ処理、およびネイティブ計算を対象としています。Pickle ベースの UDF 入力バッチを行数だけでなくバイト数でも制限できるようになり、行のサイズが変動する場合に制御しやすくなりました (SPARK-57593)。pandas-on-Spark のバッチ射影や追加のネイティブ NumPy 実装により無駄な処理が削減され (SPARK-58230、SPARK-58234)、pandas から Spark への変換では PyArrow ChunkedArray カラムのサポートが追加されました (SPARK-46776)。

このリリースではさらに、既存のベクトル関数の Python および Scala バインディング (SPARK-57284)、グループ化された共分散と記述統計 (SPARK-40336、SPARK-40337)、ローリングおよび拡大する中央値と標準誤差の操作 (SPARK-40341、SPARK-40340)、および show_versions 診断機能 (SPARK-40373) が追加されています。トレースバックのローカル変数には機密データが含まれる可能性があるため、オプトイン方式になっています (SPARK-57271)。

Spark SQL: アナリティクスと AI のためのより良いコンテキストの準備

イベントのペイロードがネストされたドキュメントとして到着する場合や、間隔の短いイベントでマイクロ秒よりも高い精度が求められる場合があります。こうした入力の準備に、すべての変換で個別のカスタム処理レイヤーが必要になってはなりません。Spark 4.3 では、ドキュメント操作、一時的マッチング、およびオプトインのナノ秒タイムスタンププレビューを含め、半構造化データと時間ベースのアナリティクス向けのネイティブ SQL が拡張されています (SPARK-56822)。

例: 同じマイクロ秒内にある 2 つのイベントを区別します。プレビューを有効にして (機能はまだ安定しておらず、互換性のない変更が行われる可能性があります)、1 ナノ秒離れたタイムスタンプを比較します (SPARK-56969、SPARK-56965、SPARK-57211):

比較結果は true に評価されます。末尾の桁は破棄されずに保持されます。このプレビューでは、精度 7 から 9 の TIMESTAMP_NTZ(p) および TIMESTAMP_LTZ(p) がサポートされます。

これには、新しい SQL 型名の受け入れにとどまらない対応が必要でした。ナノ秒のサポートは、パース、内部値表現、キャスト、式、Spark Connect、キャッシング、データソース統合にまで及びます (SPARK-56822)。新しい timestamp_nanos および unix_nanos 関数が数値変換を処理します (SPARK-57526、SPARK-57527)。ストレージとコネクタのサポートは引き続き各データパス固有のものとなるため、プレビューはエンドツーエンドで評価する必要があります。

変化する JSON ペイロードを用途のあるデータに変換する

アプリケーションイベント、ドキュメントエクスポート、モデルの応答などは、ネストされた JSON として届くことがよくあります。Apache Spark 4.3 の JSON_TABLE (SPARK-58366) は、配列の展開と型付きフィールドの抽出を組み合わせる一方、VARIANT 変換により、カスタムのドキュメント処理 UDF を使用せずにパイプラインでフィールドの更新や削除を行えるようになります (SPARK-57804、SPARK-57417)。これらを組み合わせることで、アナリティクス、エンリッチメント、およびリトリーバルパイプラインに向けたコンテンツの準備がシンプルになります。

例: 関連するドキュメントセクションを抽出する。

これにより、パイプラインが後続の処理の前にソースメタデータでエンリッチできる型付き行が生成されます。JSON_TABLE は、各行アイテムの共有トラバーサルを通じて列パスを解決するため、列ごとに個別のパース処理を行う必要がなくなります。また、Apache Spark は、集計、ソート、結合を行うサポート対象のクエリに対して、VARIANT フィールド抽出のプッシュダウンを拡張しています (SPARK-58089)。

SQL/JSON に関する今後の作業 (SPARK-58686) では、この方向性をさらに推し進め、スカラー抽出、存在チェック、フラグメント取得のために JSON_VALUE、JSON_EXISTS、JSON_QUERY が追加されます (SPARK-58685、SPARK-58687、SPARK-58723)。明示的な欠損値およびエラー処理により、アプリケーションは変更されるスキーマをより予測可能に処理できるようになります。これらは今後の開発であり、4.3 の追加機能ではありません。

イベントを当時の情報にマッチさせる

履歴分析やモデル評価では、後から追加された情報ではなく、イベントが発生した時点で利用可能だったコンテキストが必要になります。Apache Spark 4.3 の SQL ASOF JOIN では (プレビュー中の機能であり、互換性のない変更が行われる可能性があります)、手動で記述したレンジ結合とその後のランキング処理に代えて、この時間的なマッチングを直接表現できます (SPARK-58092)。

例: 予測に利用できる特徴量を再構築する。ここでは、available_at が特徴量バージョンが利用可能になった日時を記録しています (SPARK-58122、SPARK-58123):

10:03 の予測は、10:05 ではなく、10:00 の時点で利用可能だった特徴量バージョンにマッチします。専用のソートマージ実行パスにより、このマッチングはアプリケーション固有のクエリロジックではなく、エンジン処理として実行されます。可用性のタイムスタンプが当時わかっていたことを正確に表している場合、トレーニングデータ、評価、インシデント分析に向けたポイントインタイムエンリッチメントがよりシンプルになります。

主な改善点と謝辞

Apache Spark 4.3 では、日常的なデータ処理に向けた新しい API とパフォーマンスの改善も導入されています。SQL には、配列を行に展開するための ANSI UNNEST、累積カウンターをデルタに変換するための `counter_diff` ウィンドウ関数、文字列の類似度を比較するための `jaro_winkler_similarity` が追加されます。サポートされているファイルソースには、tar、zip、7z の各アーカイブの読み込みに対するオプトインサポートが追加され、個別の展開ステップが不要になります。エンジンは集計の実行を改善し、対象となる UNION 操作に対して、デフォルトでホールステージコード生成を有効にします。オプトイン最適化により、対象となるアウター結合やアンチ結合において、null の結合キーがシャッフルパーティション全体に分散されます。ネストされたデータの処理では、struct 配列内での未使用フィールドのプーニングが強化され、より多くの配列高階関数に対してコード生成が行われます。ベクトル化された Parquet の読み込みにより、型変換と辞書デコードが改善されます。

Apache Spark 4.3 は、Apache Spark コミュニティの強さを反映しており、200 人を超えるコントリビューターの協力により 1,100 件を超える Jira チケットが対応されました。このリリースを実現するために、コード、レビュー、テスト、ドキュメント、フィードバックに貢献してくださったすべての方々に感謝いたします。

Spark 4.3 を始める

リリースのダウンロードやドキュメントについては Apache Spark ダウンロードページ をご覧ください。アップグレード前に移行ガイドをご確認ください。Databricks での可用性については、Databricks Runtime リリースノート を参照してください。

(このブログ記事はAI翻訳ツールを使用して翻訳されています) 原文記事