イベントデータをインサイトに変える：transformWithStateによるスキーマ進化に対応した複雑なステート処理

業界を問わず、データエンジニアリングにおける最も根強い課題の1つはスキーマ進化です。ビジネス要件は変化し、データソースは変更され、新しいイベントフィールドが一夜にして出現し、データエンジニアリングチームはシステムを稼動させるためだけにパイプラインとステートストアを常に調整する必要があります。

スキーマが変更されると、従来のストリーミング手法は破綻します。ステートストアは互換性を失い、パイプラインは停止してしまいます。チームは、ヒストリカルデータを失うか、スキーマ移行のために高価なダウンタイムを受け入れるかという選択を迫られます。これは単なる技術的な問題ではなく、ビジネスアジリティを妨げる要因です。

Apache Spark™ 4.0は、ステートフル・ストリーミングにおけるスキーマ進化を「可能」にするだけでなく、シームレスにする画期的なAPIであるtransformWithStateInPandasを導入しました。インテリジェントな状態管理と自動スキーマ互換性により、ストリーミング・アプリケーションは、重要な状態情報を保持しながら、ビジネス・ニーズに合わせて進化します。

この記事は transformWithState シリーズの第 4 回目にして最新の投稿です: 

• 「Introducing transformWithState for Apache Spark™ Structured Streaming」では、この新しいAPIの設計原則とアーキテクチャについて詳しく解説しています。
• 「The Evolution of Arbitrary Stateful Stream Processing in Spark」では、transformWithStateInPandas が持つユニークな強みを掘り下げています。
• 「Continuous Environmental Monitoring Using the New transformWithState API」では、環境系IoTユースケースに transformWithState を適用した実例を紹介しています。

これらのブログは、transformWithStateAPIがいかにリアルタイム・ワークロードのための高度なステートフル処理を可能にし、IoTモニタリングからリアルタイム・セッションアナリティクスまでのユースケースをサポートしているかを紹介しています。

Sparkストリーミングにおけるスキーマ進化の課題

ステートフルなSparkストリーミングにおけるスキーマ進化は、運用に大きな摩擦を生み出します。applyInPandasWithState や session_windowのような従来のアプローチでは 、ステートがスキーマ情報ごとシリアライズされるため、データ構造と保存済みステートとの間に強い結合関係が形成されます。

スキーマに対してフィールド追加、型変更、カラム順の変更といった修正を行うと、既存の状態は新規入力データと非互換となります。その結果、スキーマ不整合がデシリアライズ失敗を引き起こし、運用担当者は手動での回避策の実装を余儀なくされます。

• 状態の非互換性:古い状態は、新しいスキーマ構造と整合性を取ることができません。
• 手動移行: スキーマを橋渡しするために、デベロッパーはカスタム変換ロジックを作成する必要があります。
• 必要なダウンタイム: スキーマを変更するには、ストリームを停止し、オフラインで状態を移行してから、再起動する必要があります。
• データ損失のリスク： 移行中に過去の状態が破損したり、失われたりする可能性があります。
• バージョン管理のオーバーヘッド: 複数のスキーマ バージョンをサポートするには、大量のボイラープレート コードが必要です。

なぜ スキーマ進化のためのtransformWithStateInPandas （TWS）なのか？

Sparkは、 基本的なセッション化のための　session_ windowや、 カスタムステートフル処理のための applyInPandasWithStateの ような実績のあるソリューションを提供していますが 、進化するビジネス要件は、しばしばシームレスなスキーマ進化のための追加的な柔軟性を必要とします。transformWithStateInPandas は、データとビジネスロジックの両方が継続的に進化する必要があるシナリオに対処するために、これらの基盤の上に構築されています。

transformWithStateInPandasが スキーマの進化に理想的な理由は次の通りです：

• スキーマの自動互換性： フィールドの追加、型の拡張、列の並び替えなど、既存の状態が新しいスキーマ・バージョンとシームレスに統合される。
• 進化時のステート保持：スキーマが変更されてもデータ損失は発生せず、履歴ステートは引き続き参照することができます。
• ダウンタイムを最小限に抑えたスキーマ進化: スキーマ変更を、サービスの中断を最小限に抑えてデプロイします。

シナリオの深掘り

リアルタイムでのカスタマー ジャーニーの再構築

あなたが、急成長中のオンライン小売プラットフォームである「StreamShop」のデータ エンジニアリング チームの一員であると想像してみてください。月曜日の朝、あなたのCEOが、コンバージョン率で競合他社があなたを上回っていることを示す競合他社のアナリティクスのプリントアウトを持って、エンジニアリング・スタンドアップに入ってきた。マーケティング チームは回答を求めています。

"広告に何百万ドルも費やしているが、ユーザーはどこで離脱しているのか？" "実際に購買につながるのはどの顧客経路なのか？""ユーザーが今何をしているかに基づいて、体験をパーソナライズすることは可能だろうか？"

ユーザーがクリックし、ブラウズし、商品をカートに入れ、購入するか、あるいは放棄するかによって展開されるユーザー・ジャーニー、つながりのあるシーケンス、行動ストーリーについてである。これらの旅はセッションの中で生きている。

あなたのクリックストリームデータは、Apache™ Kafkaを通してウェブやモバイルアプリから流れ込んできます。すべてのページビュー、クリック、"カートに追加、" 、あなたのストリーミングパイプラインはそれらを追跡し、定義されたスキーマを使ってセッション化し、flatMapGroupsWithStateを使ってレコードを保存します。 しかし今、新たな課題があります。先週、モバイルチームが新しいバージョンをデプロイし、 データチームに通知することなく、device_typeや page_categoryのような追加フィールドを送信し始めました。

現在使用しているウィンドウ集約ソリューションでは、このシナリオをそのままサポートしていないため、パイプラインを停止してスキーマを修正し、チェックポイントをリセットする必要があります。スキーマが変更されるたびにこの操作を行わなければならないので、これは非現実的です。より堅牢で、より柔軟で、スキーマ進化を透過的に扱えるものが必要になりました。

基礎：スキーマ進化でインテリジェントなセッションを構築する

セッションの進化を理解する

クリックストリームイベントは、セッションID、ユーザーID、timestamp、イベントタイプという基本的なV1スキーマでシンプルにスタートしました。しかし、StreamShopが進化するにつれ、イベントも進化しています。デバイスの種類、ページのカテゴリ、収益額のようなコマース特有のデータなど、豊富なコンテキスト情報を持つV2イベントを受信できるようになりました。

課題は単に2つのスキーマを扱うことではなく、既存の状態を失うことなく、あるいは最初からやり直すことなく進化させることです。セッションの連続性を維持したまま、このようなスキーマ進化を適切に処理できるセッション化ロジックが必要となります。

transformWithStateInPandas スキーマ進化 ：

状態ストアにおけるスキーマ進化とは？

スキーマ進化とは、ストリーミング クエリーが、状態ストア情報失ったり、ヒストリカルデータ完全に再処理したりすることなく、データの状態ストア スキーマの変更に対応できる能力を指します。transformWithStateInPandasの場合、クエリバージョン間でステート変数のスキーマを変更しても、既存のセッション状態を維持しつつ処理を継続できます。 以降では、その具体的な実装方法を示します。

セッショナイザーの実装

この例では、高度なセッション処理のために2つのカスタムクラス、SessionizerV1とSessionizerV2を作成しました。これらは、 transformWithStateInPandasが 、基本的なメトリクスを追跡するだけでなく、各ユーザージャーニーのコンテキストと進化を理解するのに役立つことを示しています。

V1プロセッサ： 基本形

セッショナイザーのV1では、event_count や total_revenue のようなカスタム値を追跡するための、シンプルなセッション用スキーマを定義します。

V2プロセッサ：進化したスキーマの定義

V2 は、自動スキーマ進化の真価を示すバージョンです。ここでは、2 つの新しいカラムを追加し、既存カラム（event_count）の型も拡張していますが、これらはすべて基盤となる状態ストア側でシームレスに更新されます。

従来のウィンドウベースのセッション化とは異なり、transformWithStateInPandasセッション化によって、各ユーザーのエンゲージメント・パターン、コマース行動、さらにはどのスキーマ・バージョンを使用したかなど、各ユーザーのジャーニーに関する豊富なコンテキストを維持することができます。

状態ストアにおけるスキーマ進化：

• 型の拡張（Type Widening）：event_count を IntegerType から LongType に拡張しても、自動的に変換されます。
• 新規フィールド（New Fields）：device_type と page_category を追加しても、既存の V1 状態から読む場合は、これらのフィールドは Noneとして扱われます。
• 同一ステート変数名（Same State Variable）：ステート変数名を "session_state" のままにしておくことで、自動スキーマ進化が有効になります。
• フィールド順序（Field Order）：互換性を保つため、新しいフィールドはスキーマの末尾に追加されます。

スキーマ進化でイベントを処理する

handleInputRowsメソッドは、V2が新しいV2イベントと進化したV1ステートの両方をインテリジェントに処理する方法を示しています：

スキーマ進化をマスターする：技術的に深く掘り下げる

スキーマ進化の課題を理解する

モダンなセッション化処理の本当の難しさは、イベントをグルーピングすることそのものではなく、「時間とともに変化していくイベントの姿」をどう扱うかにあります。 StreamShop では、モバイルアプリがアップデートでリッチなデータを送り始めた一方で、Web プラットフォームは従来どおりのスキーマのままだったため、この問題が一気に顕在化しました。 実際のスキーマ進化は、これから見ていくとおり、このような形で現場に表れます。

スキーマ進化の実際：

V1 イベント (元のスキーマ):

V2イベント（拡張スキーマ）：

完全なパイプラインの構成

同じ checkpointLocation を使うことで、V2 は V1 が最後にコミットしたオフセットから、その時点の状態ストアスキーマをそのまま引き継いで処理を再開できます。 これにより、過去データを再処理することなく、スキーマだけを自然に進化させることが可能になります。

Databricksがスキーマ進化を自動的に処理する方法

V2 プロセッサが V1 の状態を読み取ると、必要な変換が自動的に適用されます。これらの変換は、Databricks によって内部的に実行されます。

1.タイプ・ワイドニング（自動）

2. フィールドの追加 (自動)

3.進化検出（私たちのロジック）

ビジネスインパクト結果ストーリー

スキーマ進化を実際に見る

セッション化パイプラインをデプロイしたところ、結果はスキーマ進化の成功を示しています。次のようなことが起こります。

V1の結果 - 基本スキーマでの初回セッション：

• session_2とsession_4はターミナルイベント（purchase／logout）により完了状態です。
• session_1、session_3、session_5、session_6は、非ターミナルイベント（page_view）のみを持つため、アクティブなセッションとして継続しています。
• セッション 1、3、5、6 に対応するステートは、ステートストアに永続化されたまま保持されます。

Delta 出力：

V1セッション化出力：

StateStore の調査:

V2 セッション化の出力:

レコード状態スキーマのバージョンV2を読み取り - 進化したスキーマを持つ拡張セッション:

• 進化したV1の状態＋新しいV2のフィールドを使用して、session_1が完了する
• evolved_from_v1: true は、スキーマ進化が成功したことを確認します
• 強化されたコンテキスト：デバイスタイプ（モバイル、デスクトップ、タブレット）、ページカテゴリー（チェックアウト、プロフィール、製品）
• イベントの集積: 合計3つのイベントで収益$50。
• ステートの継続性を保ちながら、V1 プロセッサから V2 プロセッサへのクリーンな移行を示しています。

スキーマ進化分析：

現実世界でのスキーマ進化の成功

重要な知見：V2のセッション1でevolved_from_v1がtrueと表示されることです。その理由は次のとおりです。

1. V1 (page_view) で非終端イベントがありました → 状態が永続化されました
2. V2 プロセッサは、V1 の状態を読み取る際に、新しいフィールドの None 値を自動的に検出しました
3. DatabricksはV1の状態（5フィールド）をV2の状態（7フィールド）にシームレスに変換しました
4. 型の拡張（IntegerType→LongType）が自動的に行われました
5. 新しいフィールド (device_type、page_category) は V2 イベントから入力されます
6. 3 つのイベントが蓄積された状態でセッションが完了し、ステートの継続が成功したことを示しています。

前提条件設定

スキーマ進化は、Spark 構成オプションを組み合わせて設定した場合にのみ利用可能です。

1.Avroエンコーディング（必須）

2. RocksDB State Store (必須)

スキーマ進化のビジネス価値

スキーマ進化により、組織は日常業務を中断することなく、新しい機能や知見を追加できます。この柔軟性により、企業は継続的な改善、より迅速な革新、競争力の維持が可能になります。

• 最小限のダウンタイムで導入：新しいデータフィールドやトラッキング機能を最小限のダウンタイムで導入できます。顧客と社内チームはサービスを中断することなく、ビジネスはより豊かな知見を得ることができます。
• 段階的なロールアウト: スキーマの変更を段階的に導入できるため、リスクが軽減され、チームやシステム全体での導入がよりスムーズになります。ビジネス リーダーは、破壊的で大規模な変更を一度にコミットすることなく、新機能をテストして検証できます。
• 履歴の継続性： 組織は、既存のセッション状態と履歴コンテキストを維持することで、コストのかかるデータの再処理を回避できます。これにより、過去と現在のシームレスなビューが提供され、意思決定者が長期的なトレンドと知見に依拠できるようになります。
• アナリティクスの継続性： プラットフォームが進化しても、一貫したメトリクスが維持されます。この安定性により、経営幹部やアナリストは、データやレポートの不整合を心配することなく、意思決定に集中することができます。

最終的に、ステートを保持したままスキーマを進化させられる能力は、企業のリアルタイム分析への取り組み方そのものを変革します。これにより、イノベーションと運用上の卓越性を両立しつつ、サービス停止をほとんど、あるいはまったく発生させることなく継続的な成長を支援できます。

まとめ

Apache Spark™ 4.0 の transformWithState API は、単なる技術的なアップグレードではなく、リアルタイムの顧客アナリティクスを構築する方法における転換点です。

StreamShopでは、チーム全体のリアルタイムの可視化を実現しました：

• マーケティングは顧客ジャーニーをライブで追跡し、
• プロダクトチームは機能のインパクトを数時間以内に計測し、
• エンジニアリングチームはステートの透明性を通じてシステムの詳細なインサイトを得られます

組み込みのスキーマ進化機能により、セッション化パイプラインはビジネスの進化に合わせて自動的に適応し、新しいイベント、プラットフォーム、タッチポイントがシームレスに処理されます。

ジャーニー、エンゲージメント、コンバージョン ファネルのいずれを追跡する場合でも、transformWithStateInPandas は生のイベントを実用的な知見に変換し、成長を促進する継続的な顧客理解を構築します。

この記事は transformWithState シリーズの締めくくりとして、この API が IoT やセッションベースの アナリティクス ユースケース全体で、スケーラブルでステートフルな処理を可能にする仕組みを解説しました。

関連リソース

• Spark Structured Streaming プログラミング ガイド
• ステートフル・アプリケーションのスキーマ進化
• ステートフル・パイプラインのパフォーマンス向上
• Githubリポジトリ