ゲーム向けApache Sparkリアルタイムモード：リアルタイムセッション化を行うより優れた方法

ゲーム業界では、わずか1ミリ秒の遅延も許されません。ゲーム内のパーソナライズ、レコメンデーションエンジンの強化、動的なコンテンツ配信スケジュールの決定を実現するために、プラットフォームは世界中の何百万人ものプレイヤーのセッションデータを1秒未満（サブ秒）のレイテンシで処理する必要があります。

現在、このような超低レイテンシの要件を満たすために、複数のエンジンを組み合わせた複雑なアーキテクチャを採用する必要はもうありません。このブログでは、Apache Spark Real-Time Modeの実際の導入事例をご紹介します。複雑なステートフルロジックに対応する新しいtransformWithState演算子を活用することで、Sparkがどのようにエンドツーエンドでミリ秒単位のパフォーマンスを実現するかを実証します。使い慣れたStructured Streamingのエコシステムを使用して、開発を加速し、ミッションクリティカルな運用アプリケーションを構築する方法をご紹介します。

ユースケースの概要

ゲーム開始から終了まで — セッショントラッキングが重要な理由

ゲームプラットフォームにとって、どのデバイスがどのくらいの時間アクティブであるかを把握することは、単なるインフラ上の課題ではなく、ビジネスを推進する原動力となります。リアルタイムのセッションデータは、ゲーム内でのパーソナライズされた体験を提供し、レコメンデーションエンジンを強化し、コンテンツ配信スケジュールの決定に役立つほか、数百万台のコンソールやPCからデバイスの健全性シグナルを収集します。運用チームはこれを利用して、ペアレンタルコントロールを適用したり、異常なセッションパターンを検出したりします。

セッションイベントの基本

コンソールとPCの両方からのセッションイベントは、Kafkaトピックに流れ込みます。各イベントにはデバイスIDとセッションIDが含まれています。デバイスIDはコンソールまたはPCを識別し、セッションIDはゲームセッションを識別します。各デバイスで同時にアクティブにできるセッションは1つだけです。

パイプラインは、次の4つのシナリオを処理します。

• セッション開始（GameStart）：開始イベントが到着します。パイプラインはセッションIDと開始時刻を保存し、SessionActiveイベントを発行して、30秒の処理時間タイマーを登録します。そのデバイスで別のセッションがすでにアクティブであった場合は、まず古いセッションを終了します。
• セッションハートビート（Active）：タイマーが30秒ごとに起動します。パイプラインは「現在時刻 - 開始時刻」を計算し、現在の継続時間を含むSessionActiveハートビートを発行して、タイマーを再登録します。
• セッション終了（GameEnd）：アクティブなセッションに一致する終了イベントが到着します。パイプラインは最終的な継続時間を含むSessionEndを発行し、状態をクリアします。
• セッションタイムアウト（GameSessionTimeout）：タイマーが起動し、計算された継続時間が設定された最大値を超えた場合、パイプラインはハートビートを発行する代わりに、タイムアウトの理由を含むSessionEndを発行し、状態をクリーンアップします。

リアルタイムモードを備えたSparkが大きな変革をもたらす理由

マイクロバッチモードのSpark Structured Streamingでもステートフルなセッション化を処理できますが、入力処理とタイマー駆動型出力の両方で1秒未満（サブ秒）の精度が求められるユースケースでは、マイクロバッチでは不十分です。これまでは、このギャップを埋めるために、別の専用エンジンを管理するか、カスタムソリューションを構築する必要がありました。

Apache Flinkの場合：状態管理とタイマーを実装できますが、Flinkを採用するということは、Databricksプラットフォームと並行して、別のクラスター、状態バックエンド、デプロイモデル、監視スタック、コードベースなど、まったく別のエコシステムを導入することを意味します。その結果、インフラの断片化、運用の複雑化、そして2つ目のストリーミングエンジンを運用・保守するためのコストや人員が必要になります。

独自のカスタムソリューションの場合：一部のチームは、独自のセッション化サービスを構築しています。たとえば、各デバイスにセッション状態、タイマー、ハートビート送信を管理するアクターを割り当てるAkkaベースのアクターシステムなどです。これらはFlinkと同様のインフラおよび運用オーバーヘッドを伴うだけでなく、拡張性（スケール）がないという追加の課題もあります。何百万ものステートフルアクターをノード間に分散させる仕組みを自社で設計しなければなりません。これらのシステムは最初は機能しますが、時間が経つにつれてメンテナンスモードに陥り、稼働は安定していても、容易に拡張できない状態になってしまいます。

現在、Real-Time Modeはこのギャップを解消し、チームがすでに使用しているものと同じSpark APIを使用して、すべてを単一の統合エンジン内で1秒未満の精度で処理できるようにします。

transformWithStateを使用したReal-Time Mode

transformWithStateは、複雑なステートフル処理を柔軟かつスケーラブルにする、Spark Structured Streamingの次世代演算子です。主な機能には、オブジェクト指向の状態管理、複合データ型、タイマー駆動型ロジック、自動TTLサポート、スキーマ進化などがあります。Real-Time Modeと組み合わせることで、入力処理とタイマー駆動型出力の両方で1秒未満の精度を実現します。

ゲームのセッション化のユースケースでは、次の2つのことが求められます。

• リアクティブ処理：セッションの開始と終了が到着した時点で処理します。
• プロアクティブ出力：受信データとは無関係に、スケジュールに従ってアクティブなセッションごとにハートビートを生成します。

transformWithStateは、2つの専用メソッドを持つ単一のStatefulProcessorクラスでこれら両方を実現します。
handleInputRows()は、受信するKafkaイベントに反応し、セッションの開始と終了を処理し、イベントの到着に合わせてセッション化の状態を維持します。

handleExpiredTimer()はその間に発生するすべての処理を担当し、新しいデータが到着したかどうかに関係なく、ハートビートやタイムアウトなどのプロアクティブな出力を生成するために起動します。

仕組み：リアルタイムのゲームセッション化パイプラインの構築

パイプラインアーキテクチャの概要

• イベントの取り込み：コンソールやPCからのセッションイベント（開始と終了）がKafkaトピックに到着します。各イベントが解析され、デバイス固有の識別子からdeviceIdが派生されます。
• ステートフルなグループ化：ストリームはdeviceIdごとにグループ化され、特定のデバイスに対するすべてのイベントが同じステートフルプロセッサインスタンスにルーティングされるようにします。
• 処理：transformWithStateがセッション化プロセッサを適用します。このプロセッサは、セッションIDをキーとするMapStateを使用して、デバイスごとのアクティブなセッションを追跡します。セッション開始イベントが到着すると、handleInputRows()がセッション状態を保存し、SessionActiveイベントを発行して、最初の30秒タイマーを登録します。それ以降はhandleExpiredTimer()が処理を引き継ぎ、30秒ごとにハートビートを発行し、タイムアウトをチェックします。セッション終了イベントが到着すると、handleInputRows()が再び処理を行い、最終的な継続時間を含むSessionEndを発行し、状態をクリアしてタイマーのループを停止します。
• 出力：処理されたセッションイベント（開始、ハートビート、終了、タイムアウト）はJSONとして出力Kafkaトピックに書き込まれ、ダウンストリームで利用できるようになります。

実装の詳細

アーキテクチャ、コード実装、本番環境での考慮事項の詳細な解説については、こちらの関連ブログをご覧ください。このブログでは、StatefulProcessorのコード、タイマーのライフサイクル、状態管理パターン、およびStreamingQueryListenerを使用した監視について詳しく説明しています。以下の結果は、パイプラインのスループットとレイテンシの特性を示しており、マイクロバッチモード（MBM）とReal-Time Mode（RTM）の間の大幅なレイテンシの違いを浮き彫りにしています。

スループット

現実的な負荷の下でパイプラインを検証するため、以下の持続的なスループットでテストを行いました。

出力の大部分は受信データによってトリガーされるのではなく、完全にhandleExpiredTimer()によって生成され、スケジュールに従ってプロアクティブにハートビートを発行します。

レイテンシ

レイテンシは、Kafkaの入力トピックのタイムスタンプから出力トピックのタイムスタンプまで、エンドツーエンドで測定されます。Real-Time Modeを使用すると、パイプラインは432ミリ秒のp99レイテンシを達成し、これはマイクロバッチモードよりも20倍高速です。

まとめ

ゲームのセッショナイゼーション（セッション化）のようなユースケースでは、受信イベントの処理にとどまらないパイプラインが必要です。スケジュールに従ってプロアクティブにハートビートを送信し、数百万の同時セッションを追跡し、状態（ステート）を効率的に管理する必要があります。このパターンはゲームに限られません。タイマー駆動の出力を必要とするあらゆるワークロード（IoTのハートビート、セッション追跡、リアルタイムのアラート、機器のモニタリングなど）を、同様の方法で構築できます。

transformWithState のタイマー機能がこれを可能にします。単一の StatefulProcessor クラスが、リアクティブな入力処理とプロアクティブなタイマー駆動の出力という、セッションのライフサイクル全体を処理します。リアルタイムモード（Real-Time Mode）と組み合わせることで、入力レコードが処理され、タイマーがミリ秒（サブ秒）単位の精度で起動します。次のバッチ間隔を待つことなく、即座に実行されます。これらすべてを、別のエンジンを追加することなく、Databricks 内だけで完結できます。

すでに Structured Streaming パイプラインをマイクロバッチモードで実行しており、より低いレイテンシを実現するために別のエンジンを検討している場合は、まずリアルタイムモード（Real-Time Mode）をお試しください。切り替えはトリガーを1つ変更するだけで、コードの書き換えやプラットフォームの移行は不要です。

ぜひご自身でお試しください：

• データジェネレーター付きのコンパニオンノートブック：ゲームのセッショナイゼーションパイプライン全体を実行し、MBMとRTMのレイテンシをご自身で比較できます。
• transformWithState APIガイド：ステート変数、タイマー、TTL、スキーマエボリューション
• リアルタイムモード（Real-Time Mode）リファレンス：サポートされている演算子、実行モード、ソース、シンク、言語サポート

リアルタイムモード（Real-Time Mode）が一般提供開始（GA）されました。

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