Zenn
🌏

Apache Spark 4.0 ― “何が進化したか” を俯瞰する

に公開
spark4
sparkconnect
deltalivetables
structuredstreamin
sql拡張機能
tech

はじめに：Spark 4.0 が示す新世代の基準

2025年5月、ApacheSparkは4系列の幕開けとなるSpark4.0.0を正式リリースしました。コミュニティは5100件超のJIRAを解決、390名以上の開発者がコードとレビューを寄せ、従来の3.x系から“互換性を保ったまま標準SQL／クラウドネイティブ／PythonDXを大幅強化”するメジャーアップデートを実現しています。Spark Release 4.0.0

Spark4.0の象徴的なポイントは次の4つですIntroducing Apache Spark 4.0。

  1. ANSI SQLモードを既定でON ― 型変換ミスやゼロ除算を例外で捕捉し、データ品質を守る“厳格なSpark”。
  2. Spark Connectの実用段階突入 ― spark.api.mode=connectの一行でクライアント／クラスタ分離へ移行。わずか1.5MBの軽量PyClientとGo/Swift/Rust連携で、Kubernetesやサーバレスとの親和性が飛躍的に向上。
  3. Python & SQL エコシステム拡充 ― DataFrame 直結のネイティブPlot API、Python DataSource／UDTF、SQL UDF・PIPE・VARIANT型・セッション変数など、“コードレス ETL” を後押しする機能が目白押し。
  4. Structured Streaming 2.0 への進化 ― 新API transformWithStateとState Data Sourceが高度なCEPとデバッグを容易にし、観測性も強化。

これらは単なる機能追加にとどまらず、「コード可搬性・運用信頼性・開発生産性」の三拍子をそろえて次世代データ基盤のデファクトを狙うという、プロジェクトの明確な方向性を示しています。Spark 3.xで構築したワークロードは概ねそのまま動きますが、ANSI 厳格化や Connect への切り替え、VARIANT 型の採用といった“攻めの適応”でこそ真価を享受できます。

本ブログでは、上記の進化を「全体像 → コア & SQL → Connect & Python → Streaming & 運用 → 移行ガイド」の4章構成で深掘りします。Spark 4.0でデータパイプラインを次のステージへ引き上げる準備を始めましょう。

第1章 全体像：4つの進化テーマと注目機能

Apache Spark4.0は「既存ワークロードとの互換性を維持しながら “より厳密・よりクラウドネイティブ・より Python DX” を実現する」という明確な指針を掲げ、大きく4つのテーマに集約して機能強化が行われました。それぞれの狙いと主要アップデートを俯瞰します。

  1. 標準 SQL & データ品質の強化
    • ANSI SQLモードが既定でON ― 型変換誤りや0割による“静かな失敗”を防ぎ、例外で通知。
    • SQL 言語拡張 ― reusable SQL UDF, |> PIPE 構文, セッション変数／パラメータマーカー, SQL スクリプト。コードレス ETL を後押し。
    • 半構造化データ型 VARIANT と Collation でJSONを1列に格納しつつ多言語・大文字小文字感知の比較が可能。
  2. Spark Connect ― モジュラー＆クラウドネイティブな実行モデル
    • spark.api.mode=connectの1行でクライアント／サーバ分離へ。Classicとほぼ同等のAPIカバレージ。
    • 1.5 MBの軽量PyClient、さらにGo・Swift・Rustクライアントも公式提供。CI/CDやサーバレスでのコールドスタートを短縮。
    • MLlib、Plot APIなどもConnect経由で利用可能になり、Kubernetes Operatorとの連携も強化。
  3. Python 開発者体験の飛躍
    • PySpark DataFrame 直結の.plot()（Plotly backend）で可視化がワンライナーに。
    • Python Data Source APIにより100 % Pythonで独自コネクタを実装可能。
    • 多態的 Python UDTF、Arrow 2 対応、統合プロファイリングでUDF性能計測が容易。
  4. Structured Streaming & 観測性のアップグレード
    • 新APItransformWithState（Arbitrary Stateful Processing v2）で任意オブジェクト状態＋タイマー制御の複雑CEPを簡潔実装。
    • State Store Data SourceによりステートをDataFrameとして可視化・デバッグ。
    • 構造化 JSON ログ (spark.log.structuredLogging.enabled=true) と OpenTelemetry Sinkでメトリクス／ログを一元集約。

第2章 コアエンジン & SQL 深掘り

Spark 4.0では “内部基盤を最新化しつつSQLレイヤを本格的にANSI準拠へ — 互換性を保ったまま品質と運用性を底上げする”ことが徹底されました。ここでは ①ランタイム基盤刷新 → ②パフォーマンス／運用強化 → ③SQL 言語拡張の3フェーズで掘り下げます。

2-1. ランタイム基盤の刷新

  • Scala 2.13が唯一の公式バイナリ（Scala 2.12 は廃止）
  • JDK 17が新しい最低ライン（JDK 8/11 をドロップ、JDK 21 まで動作確認）
  • Spark自体のCIはbranch-4.0 ⟨Scala 2.13 × JDK 17 × Hadoop 3⟩ で構築。

JVMのGC/Flight Recorder改善とScalaコレクション強化がそのまま恩恵に。移行時は依存JARの Scalaバージョン差分とJDK 17以上でのリコンパイルを確認しましょう。Apache Spark - A Unified engine for large-scale data analytics

2-2. パフォーマンス & 運用強化ポイント

領域 主なアップデート 期待効果
Shuffle CRC32C チェックサム / 並列 LZF 圧縮 / ShuffleCleanupMode デフォルト有効化 ネットワーク I/O 削減・ジョブ安定性向上 Databricks Runtime 17.0 (Beta)
メモリ管理 TaskInfo.accumulables() の存命期間短縮 Driverヒープ使用量を数十%削減
構造化ログ spark.log.structuredLogging.enabled=trueでJSON Lines出力、OpenTelemetry Sink追加 異常分析と分散トレーシングを統合 Apache Spark 4.0: New Features with Sample Code
セキュリティ AES-GCM暗号化オプション、RPC SSLキーパスワード対応 FIPS準拠クラスタでの暗号強度向上
クラスタ 新Spark Kubernetes Operator、Standalone改良（自動再起動・UI 強化） K8s/Edge-クラスタ運用がシンプルに

2-3. SQL レイヤ徹底強化

  1. ANSI SQLモードが既定でON
    CAST('ABC' AS INT) や0除算は即例外。Hiveライクに戻す場合は

    spark.sql.ansi.enabled=false

    でレガシーモードを一時併用できます。

  2. 言語機能拡張 ― “Pure SQL だけで ETL を完結”
    Spark 4.0のSQLレイヤは「ETL 処理をほぼコードレスに書き切る」ことを目指し、５つの大型拡張を投入しました。従来はDataFrame／UDF／外部言語に逃げていた変換ロジックを、そのまま標準SQLで表現できます。

機能 ねらい & 効果 ミニサンプル
SQL UDF ANSI 準拠で再利用可能な関数をSQL内で宣言。Catalystが最適化できるため Scala/Python UDFより高速に実行可能。 sql<br>CREATE FUNCTION to_celsius(t_f DOUBLE)<br>RETURNS DOUBLE<br>RETURN (t_f-32)/1.8;
PIPE サブクエリやCTEを排して処理フローを>で線形に連結することで、SQLの可読性・保守性・デバッグ容易性を同時に高めつつ、冗長な記述を削減できる仕組みです。 ★★★（下記記載）
VARIANT 型 JSONなど半構造化データを1列で格納しつつ、ネスト検索は列プルーニングで高速。Delta Lakeも同型をサポート。 sql<br>CREATE TABLE logs(id INT, data VARIANT);<br>SELECT data:user.id AS uid FROM logs;Introducing the Open Variant Data Type in Delta Lake and Apache Spark
STRING COLLATE 100+ 言語に対応する照合順序 (大文字小文字／アクセント感知)を列単位で指定。国際化ETLがシンプル。 sql<br>CREATE TABLE user_dim(name STRING COLLATE 'utf8ci_ai');Apache Spark 4.0 is here: Top features revolutionizing data engineering & analytics
SQL スクリプト & セッション変数 DECLARE/SET/IF/WHILEを備えたスクリプトブロックと変数で、日次バッチや差分ロードをフルSQLで記述可能。 sql<br>DECLARE run_dt DATE;<br>SET run_dt = current_date();<br>INSERT INTO sales_part PARTITION (dt = run_dt)<br>SELECT * FROM staging WHERE dt = run_dt;Apache Spark 4.0

何がうれしいのか？

  • 実装工数を圧縮 ― ETLロジックをSQLファイルだけで完結でき、Python/Scala部分のCI/CDから切り離し可能。
  • 最適化メリット ― Catalyst/AQE がフル適用されるため、同じ処理でもスクリプト+UDFより高速になるケース多数。
  • 運用と監査 ― Pipeline（DLT）のLinage UIにSQLステップがそのまま表示され、デバッグや影響分析が容易。

★★★（PIPE）
従来の SQL

SELECT country,
       date(ts)    AS log_day,
       COUNT(*)    AS pv
FROM   raw_events
WHERE  ts >= '2025-06-01'
  AND  country = 'JP'
GROUP  BY country, date(ts)
ORDER  BY log_day;

PIPE 版（Spark 4.0 以降）

FROM raw_events                     -- スタンドアロン FROM が起点
|> WHERE ts >= '2025-06-01'
|> WHERE country = 'JP'
|> EXTEND log_day = date(ts)
|> AGGREGATE COUNT(*) AS pv
     GROUP BY country, log_day
|> ORDER BY log_day;

読み方のポイント

  • FROM句を「入口」として宣言し、そこにパイプを連続追加
  • WHEREを複数行に分ければフィルタ条件の追加・削除が一行作業。
  • EXTENDで派生列を作り、次のAGGREGATEですぐ参照。
  • 途中まで実行すれば中間 DataFrameを即確認できる。

第3章 Spark Connect & Python UX の飛躍

Spark 4.0における最大の進化の一つが、Spark Connectの正式化とPython開発者体験（UX）の強化です。これにより、クライアントとSparkクラスタのアーキテクチャが分離され、Kubernetesやサーバレスとの親和性、ローカル開発・CI/CDの柔軟性が格段に向上しました。

3-1. Spark Connect ― クラウドネイティブを見据えた分離実行モデル

項目 内容
導入方法 spark.api.mode=connectという1行の設定だけで、クライアント⇔クラスタを RPC経由の分離モードに変更可能（Classic モードとも共存可）
特徴 - DataFrame API、MLlib、Plot APIを含むほぼすべての操作がConnect経由で可能
- Catalyst最適化・AQE・パーティショニング戦略はすべてクラスタ側で保持
軽量なgRPC通信によりPythonクライアントは1.5MB程度に圧縮可能（コンテナ／Lambda等で高速起動）
サポート言語 Python（1.5MB クライアント）、Scala、Go、Rust、Swiftなどマルチランゲージ対応が進行中

3-2. Python UX：PySpark を「Pythonic」に変えるアップデート群

Spark 4.0ではPySparkを単なるJVMバインディングではなく、Python 開発者にとって直感的・生産的なAPIに昇華させるための改良が行われました。

項目 概要 例・効果
Plot API df.plot()によるワンライナー可視化（Plotly backend） df.plot(kind="hist", column="price")
Python Data Source API 100%Pythonで独自データソース実装が可能 REST API, MQTT, HDF5 などのカスタム読み込み
Python UDTF（表関数） @udtfデコレータで1行→複数行出力可能なUDTFを定義 センサデータの展開やトークン化処理などに有効
pandas API on Spark 拡張 to_feather() / explode() / json_normalize() などの追加により、既存pandasコード資産がほぼそのままSparkに移植可能 pandas → 分散処理に違和感なく移行できる
プロファイリング統合 DataFrame.explain("formatted")やspark.confによる内蔵プロファイラとArrow 2対応 UDF性能のボトルネックやメモリ使用を簡単に可視化

3-3. Connect × Python の活用シナリオ

シナリオ メリット
サーバレス環境でのバッチ処理 PySparkクライアントが軽量（1.5MB）なので、Lambda/Fargateなどでも起動時間が大幅短縮
ノートブックローカル開発 クラスタと分離しているため、ローカル環境のJupyterや VSCodeからSparkを “リモート実行” 可能
CI/CD パイプラインへの統合 クライアント・サーバ分離により、CI/CD テストステージでもSpark環境と独立したコードチェック・テストが可能
pandas 開発者の移行 pandas-on-Sparkの互換性拡張により、初心者でも違和感なく PySparkへ移行可能

まとめ

Spark Connectにより、Sparkは分離・疎結合アーキテクチャへと移行し、クラウドネイティブな開発・運用が一段と現実的になりました。また、PySparkの強化により、Pythonエンジニアもpandas + 可視化 + UDF開発の文脈でSparkを “自然な拡張” として扱えるようになりました。
Spark 4.0以降の開発では、「まずConnectを有効化してローカルから試す」ことが、PoC を素早く回す上でも最適な第一歩となります。

第4章 Structured Streaming & 観測性アップグレード

Spark 4.0は、リアルタイム処理の中核であるStructured Streamingの柔軟性・可観測性を大きく強化しました。特に注目すべきは、任意ステート管理 API：transformWithStateの追加と、State Storeの可視化／構造化ログ出力による“運用しやすさ”の向上です。

4-1. transformWithState()の登場（Arbitrary Stateful Processing v2）

Spark 4.0では、従来のmapGroupsWithState()やflatMapGroupsWithState()よりも**汎用性・保守性に優れた新API**transformWithState()`が追加されました。

特徴 説明
任意のキー・バリュー・タイムアウト・出力を制御可能 ステート管理・タイマー設定・出力を1関数内でまとめて記述できる
StateStoreAPIによる状態の読み書き 自前で初期化、更新、TTL制御もできる（状態が複数イベントにまたがる場合にも柔軟に対応）
関数の分離とテスト容易性 ステートロジックが明示的な関数になるため、ユニットテスト可能な構造になる

使用例（擬似コード）

from pyspark.sql.streaming.state import GroupStateTimeout, GroupState

def update_state(key, values, state: GroupState):
    count = state.getOption().getOrElse(0) + len(values)
    state.update(count)
    return (key, count)

df.groupByKey(...).transformWithState(update_state, outputMode="update")

従来のmapGroupsWithStateに比べて、タイムアウト処理・状態消去・外部出力の柔軟性が大幅アップしています。

4-2. State Store の可視化機能が追加

これまでブラックボックスだったSparkのState Storeに対して、DataFrameとして直接中身を確認・デバッグできる仕組みが導入されました。

仕組み 説明
State Store を Delta テーブルとして読み取り可能 read.format("stateStore")...でステート内容をクエリ可能（キー・状態・最終更新時刻など）
外部監査・分析との統合 Kafka状態 / 在庫 / トランザクション等のリアルタイムビューとしても活用可能
ロジック検証／ステートリークの確認 テスト時にステートが消えていない／期待通りに更新されているかを目視で確認できる

4-3. 構造化ログと観測性の標準装備

Spark 4.0では運用の可視性向上のため、構造化ログ（JSON Lines）形式での出力とOpenTelemetry Exportが標準でサポートされました。

機能 内容
構造化ログ spark.log.structuredLogging.enabled=trueにより、イベント／エラー／メトリクスをJSON Lines 形式で出力。クラウドログ基盤（CloudWatch, Datadog, Elasticsearch 等）と統合しやすい。
OpenTelemetry Sink Sparkのメトリクス（task数・失敗回数・レイテンシなど）を OpenTelemetry Protocol：OTLPで外部へエクスポート。Grafana / Prometheus / Datadog などと連携可。
Spark UI 強化 FlameGraph表示・Thread Dump UIなどが統合。ボトルネックの検出が簡単に。

4-4. 実運用での使いどころ

シーン 活用方法
異常検知・DR制御（電力／車載） transformWithState()で特定条件の連続検出 → アラートをリアルタイム制御（ex: 10分連続で温度上昇したら充電停止）
在庫 or 売上の集計 transformWithState() + StateStore Viewを使って日次／週次累積値を保ったまま出力し、外部レポートと連携
検証時のロジック可視化 構造化ログ & State Viewでデバッグがしやすく、失敗レコード・遅延の原因を明確に特定可能

まとめ

Spark 4.0はStructured Streamingに「ステートフル処理の自由度と開発性」「実運用を前提とした監視・可視化」を本格的に統合しました。

  • transformWithState：状態を保持しながら、柔軟にイベントを処理
  • State Store View：状態の中身を直接確認できる
  • Structured Logging：ログ基盤・監視ツールとの統合が容易

リアルタイム系処理が求められるシステム（VPP・IoT・金融など）において、Spark 4.0は高い表現力と運用性を両立する基盤として、強力な選択肢となっています。

Discussion