Apache Spark リアルタイムモードによる超高速異常検知

本記事では、不正検出、IoTセンサー監視、リアルタイムのパーソナライズ、セキュリティシグナル処理など、ビジネス成果に即時対応が不可欠なあらゆるシナリオにおいて、実際に大きな効果をもたらす運用ワークロード向けの再利用可能なパターンを確立します。

中心となる目的：イベントが不審または無効であると思われる場合、即座にフラグを立て、適切なダウンストリームのアクションにルーティングすることです。

このブログでは、Ethereumブロックチェーン取引における異常検知の実演を行います。Ethereumブロックチェーンのデータを分析し、無効なパターンを持つ取引をリアルタイムでフラグ付けします。具体的には、以下を検出します：

1. データ品質違反：Ethereumプロトコルにおいて物理的に不可能なgas_used > gas_limit が存在するブロック。これはデータの破損、プロデューサーのバグ、またはスキーマ解析の失敗を示します。
2. ペイロード衛生違反：認識可能なPIIまたは認証情報のパターン（メールアドレス、JWTトークン、AWSアクセスキー）を含むextra_dataフィールド。これはデータ漏洩やプロデューサーの設定ミスを示します。

デモンストレーションにはEthereumデータを使用していますが、この「不審または無効」という分類は、多くの価値の高いユースケースに適用できます：

• 不正検出：取引が異常なパターンを示す → ダウンストリームの調査ワークフローをトリガーする
• IoT監視：センサーの読み取り値が物理的に不可能なパラメータの範囲外になる → 自動応答を開始する
• セキュリティ運用：ペイロードにシークレットまたはPIIパターンが含まれる → 統合ガバナンスによりリアルタイムで隔離する
• パーソナライズエンジン：特定の行動イベントに反応し、文脈に応じたオファーを即座に提示する

同じ検出ロジックは、予期しないPIIを含む金融取引、範囲外のセンサー読み取り値を持つIoTペイロード、編集（マスキング）されるべきシークレットを含むAPIイベントログなど、あらゆるドメインに直接マッピングできます。Ethereumストリームは、このパターンを大規模に実証するための、クリーンで再現可能なデータセットを提供します。

Sparkリアルタイムモードについて

リアルタイムモード（RTM）は、Apache Spark™ Structured Streamingの新しいトリガータイプであり、Apache Flinkのような別の専用エンジンを使用することなく、Spark APIにミリ秒レベルのレイテンシを提供します。

Structured Streamingのデフォルトのマイクロバッチモードは、乗客が満員になるのを待ってから出発する空港のシャトルバスのように動作しますが、RTMは高速の動く歩道のように動作し、各イベントが到着するたびに処理します。これは、3つのアーキテクチャ上の革新によって実現されています。すなわち、継続的なデータフロー（イベントは個別のチャンクではなく、到着した時点で処理されます）、パイプラインスケジューリング（すべてのクエリステージがブロッキングなしで同時に実行されます）、およびストリーミングシャッフル（データはディスクをバイパスして、メモリ内でタスク間に即座に渡されます）です。

RTMは、不正検出、リアルタイムのパーソナライズ、ML特徴量の計算、IoT監視など、レイテンシがビジネス成果に直接影響を与える運用ワークロード向けに専用設計されています。1〜2秒のレイテンシを許容できるワークロードの場合、従来のマイクロバッチが引き続きコスト効率の高い選択肢となります。

RTMがゲームチェンジャーである理由

1. 超高速：1秒未満のレイテンシが実現可能に

リアルタイムモードは、Apache Sparkで可能なことを根本から変えます。ワークロードの複雑さに応じて約5msから約300msの範囲におよぶエンドツーエンドのレイテンシにより、Sparkはこれまで専用のストリーム処理エンジンが独占していた領域に踏み込みます。従来のマイクロバッチは1〜2秒のレイテンシを提供しますが、リアルタイムモードは約5msから約300msを達成します。
このアーキテクチャは、事前割り当てされた実行パイプラインと非同期チェックポインティングによってこれを実現し、従来のマイクロバッチ処理を制約していたスケジューリングオーバーヘッドを排除します。ミリ秒単位が重要となる運用ワークロード（不正検出、IoT監視、リアルタイムオファーなど）にとって、このレベルのパフォーマンスは変革をもたらします。

2. スタックの簡素化：別のテクノロジーが不要に

多くの組織が、「Sparkはリアルタイムのユースケースに対して十分なパフォーマンスを発揮しないため、この1つの要件のために完全に別のスタックが必要である」という、コストのかかる誤解に直面しがちです。

1〜2秒のレイテンシを許容するワークロードの場合、Sparkのマイクロバッチは優れた費用対効果でデータをDelta Lakeに確実に配信します。1秒未満の応答時間を必要とする運用ワークロードの場合、リアルタイムモードは別のテクノロジーの必要性を完全に排除します。これは、分析ワークロードと運用ワークロードの両方を単一のSparkベースのスタックに統合したCoinbase、DraftKings、およびMakeMyTripのチームによって実証されています。

リアルタイムモードを使用すると、Sparkは単一の統合プラットフォーム内で、分析（秒単位）と運用（ミリ秒単位）の両方のワークロードを処理できます。これにより、以下が削減されます：

• 運用の複雑さ：管理、監視、トラブルシューティングを行うテクノロジースタックが1つで済みます
• トレーニングのオーバーヘッド：既存のSparkの専門知識をリアルタイムのユースケースに直接応用できます
• 統合の摩擦：個別のストリーミングエンジン間での複雑なデータの受け渡しが不要になります
• 総所有コスト（TCO）：統合により、インフラストラクチャ、ライセンス、および運用コストが削減されます

3. 開発者に優しい：トリガーの簡単な変更のみ、コードの書き換えは不要

リアルタイムモードの最も魅力的な側面は、すでにStructured Streamingに精通している開発者にとって、驚くほどシンプルであることでしょう。この強力な機能を有効にするために、複雑な移行や根本的なコードの再構築は必要ありません。

組織は、トリガー設定を変更するだけで、ミリ秒レベルのレイテンシを実現できます：

これだけです。使い慣れた同じStructured Streaming API、同じチェックポイント管理、同じ少なくとも1回（at-least-once）の配信セマンティクス。1つの設定変更だけで、1秒未満の運用インテリジェンスを有効にできます。

配信保証に関する注意：Kafkaシンクを使用したRTMは、少なくとも1回（at-least-once）の配信保証を提供します。ダウンストリームのコンシューマーは、べき等（idempotent）書き込みまたは重複排除ロジックを介して、潜在的な重複を処理する必要があります。

このシームレスな統合は、極めて重要なメリットをもたらします。チームは、完全に別のテクノロジースタックを学習、デプロイ、管理するという大きなオーバーヘッドを伴うことなく、運用ワークロードのプロトタイプ作成と本番環境への導入を行うことができます。このアプローチにより、新しいリアルタイム機能の導入に従来伴っていたリスクを軽減しながら、イノベーションを劇的に加速させることができます。

リアルタイムモードが重要である理由を説明したところで、実際にこのパターンを実装する方法を見ていきましょう。以下のセクションでは、これらの機能をビジネス価値へと変換する運用パターンである、本番環境に対応したガードレールパイプラインの実演を行います。

アーキテクチャの概要：運用ガードレールストリームの構築

入ってくるすべてのイベントは即座に評価され、以下を含む強化されたダウンストリームイベントが生成されます：

• 判定：ALLOW 対 QUARANTINE
• 理由：トリガーされたフラグ（データ品質違反、ペイロード衛生上の懸念）の詳細な説明

この運用パターンは、リアルタイムの意思決定における信頼できる唯一の情報源（Single Source of Truth）として機能します：

• このイベントは調査のために隔離すべきか？
• ダウンストリームシステムが即座に反応できるように、どのようにイベントを強化するか？

このデモではイーサリアムのブロックデータを使用していますが、このパターンは金融取引、センサーデータ、認証ログ、API呼び出しなど、あらゆるユースケースに共通して適用でき、アーキテクチャは一貫しています。

大規模環境での検証

本番環境品質のソリューションを提供するというコミットメントに基づき、このパターンを大規模環境で検証しました。イーサリアムの全チェーン（4つのパーティションに分散された約95 GB、約2,300万メッセージに相当）をテスト用にKafkaにロードしました。

イベント分類のための検証ルールの定義

意図的にシンプルで、シグナル強度の高い検証ルールを実装します。

ルール1：ペイロードの健全性検証

本番環境のストリームに存在すべきではないデータを明確に示すパターンがないか、extra_dataフィールドをスキャンします。コード例では、基本的なパターン検出（メールアドレス、JWTトークン、AWSキーの形式）を示しています。

組織は、これらをご自社のコンプライアンス要件（PIIパターン、内部識別子、API認証情報、および同様の機密データ）に特化したルールに置き換える必要があります。

重要なポイント：リアルタイムのガードレールは、インシデント発生後の分析で発見するものではなく、統合ガバナンスの一部としてパイプライン内に組み込むべきです。

ルール2：データ品質の検証

gas_used > gas_limit

この状態は、有効なデータでは決して発生しないはずです。これが検出された場合、以下のいずれかの問題が発生していることを示しています。

• 送信中のデータ破損
• プロデューサー側でのデータ生成エラー
• スキーマ解析の不整合
• アップストリームシステムの障害

運用上の観点から、これはまさに即座にフラグを立てたいタイプのアノマリー（異常）であり、ダウンストリームシステムが影響を受ける前に迅速に対応できるようにします。

検証ロジックが確立されたので、次は1秒未満（サブ秒）の実行を可能にするストリーミング設定について見ていきましょう。

リアルタイムモード：必須の設定

リアルタイムモードは、リアルタイムトリガーを介して有効になり、アップデートモードで動作します。PySparkでは、インターバルパラメータ（例："5 minutes"）を指定します。

2つの重要な設定要件：

1. クラスター設定：Databricksのドキュメントには、必要なジョブクラスター設定とRTM有効化フラグが指定されています。
2. 出力モード：RTMトリガーではupdateモードを使用する必要があります。

このデモで使用した設定：

• ランタイム：Databricks Runtime 16.4 LTS以降
• コンピュート：固定ワーカー数の専用（シングルユーザー）クラスター（自動スケーリングは無効）
• Photon：無効（RTMではサポートされていません）
• ワーカー：このワークロードに対して4ワーカー
• 出力モード：update（RTMに必須）

完全なcluster_config.template.jsonについては、付属のリポジトリを参照してください。

実装：リアルタイムガードレールパイプライン

このシングルパスパイプラインは、KafkaとSparkのリアルタイムモード間のシームレスな統合を示しています。

1. Kafkaソースへの接続
2. 受信するJSONペイロードの解析
3. 判定と理由の計算
4. エンリッチされたJSONをKafkaに書き戻す

コードはこちらからお読みいただけます。

結果：大規模環境でのパフォーマンス

連続ストリーミングモードにおいて、4つのKafkaパーティションに分散された約2,300万メッセージのイーサリアムブロックチェーンデータを処理することで、リアルタイムモードのパフォーマンスを検証しました。

スループットのパフォーマンス

このパイプラインは、優れたスループット特性を示しました。

• 入力レート：65,592行/秒
• 処理レート：69,713行/秒（持続的）
• 処理された総レコード数：約23,213,628メッセージ
• クラスター設定：DBR 16.4 LTS、4ワーカー（i3.xlarge）、専用シングルユーザーモード、Photon無効

レイテンシーの指標

ドライバーログは、spark.streaming.madeProgressイベントを介して詳細なレイテンシー指標をキャプチャします。リアルタイムモードはprocessingLatencyMsをレポートします。これは、クエリがレコードを読み取ってからダウンストリームのシンクに書き込むまでの時間を測定するものです。

約2,300万件のレコードを処理するこのステートレス検証パイプラインでは、以下が観察されました。

• P0、P50、P90、P95：1ミリ秒未満（メトリクスでは0に丸められます）、P99：1ミリ秒、処理レート：69,713行/秒（持続的）
• processingLatencyMsの理解：このメトリクスは、RTMがレコードを読み取ってからダウンストリームのシンクに書き込むまでの時間を測定します。タスクごとに測定され、StreamingQueryProgressのrtmMetrics.processingLatencyMsセクションにパーセンタイル（P0、P50、P90、P95、P99）とともにレポートされます。このようなシングルステージのKafka-to-Kafkaパイプラインの場合、これは実質的にエンドツーエンドのレコードごとのレイテンシーを表します。
• これが意味すること：大多数のレコード（95パーセンタイル）は0.5ミリ秒未満で処理され、最も遅い1%であっても1ミリ秒以内に完了しました。P0〜P95で「0」と表示されている値は、0.5ミリ秒未満のレイテンシーを示しています（メトリクスシステムによって切り捨てられています）。

注：これらの結果は、ステートレス検証パイプラインでのパフォーマンスを示しています。より複雑なステートフル操作（集計、ウィンドウ処理）では、ワークロードの複雑さに応じて、約5ms〜約300msのRTM範囲内でより高いレイテンシーが発生する可能性があります。

主要なパフォーマンスに関するインサイト

• 解釈：全レコードの99%が1ミリ秒未満で処理され、最も遅い1%のみが1ミリ秒に達しました。これは、極めて一貫した低レイテンシーパフォーマンスを示しています。
• スループット：パイプラインは約2,300万レコードを処理しながら69,713行/秒を維持し、継続的な負荷の下で安定したパフォーマンスを示しました。

これらのメトリクスは、リアルタイムモードが毎秒約70,000行の大量ストリームを処理しながら、本番環境レベルのサブミリ秒レイテンシー（P95 0.5ms、P99 = 1ms）を提供することを示しています。これにより、統合プラットフォームと低レイテンシー要件 of 従来のトレードオフが解消されます。

まとめ

リアルタイムモードは、Apache Spark™ Structured Streamingを新しいクラスのワークロード、つまりストリーミングデータへの即時応答が求められる、レイテンシーに敏感な運用アプリケーションへと拡張します。チームがすでに使用しているSpark APIにサブ秒のレイテンシーをもたらすことで、最も時間的な制約が厳しいパイプラインのために、別の専用エンジンを運用する必要がなくなります。

その価値提案は非常に魅力的です：

• 1つの統合プラットフォーム：分析ワークロード（秒単位）と運用ワークロード（ミリ秒単位）の両方を処理します。
• 既存のSparkの専門知識をそのまま活用可能：個別の専門知識は不要です。
• 最小限の移行リスク：トリガー設定を1つ変更するだけで、リアルタイム機能が有効になります。
• 本番環境で検証済みのパフォーマンス：何百万ものイベントを処理しながら、P99レイテンシーは1ms、P95は0.5ms未満を達成します。

不正検出パイプライン、パーソナライズエンジン、ML特徴量計算システムのいずれを構築する場合でも、リアルタイムモードはSparkのシンプルさとエコシステムの広さを維持しながら、アプリケーションが求めるレイテンシーを提供します。

はじめに

付属のリポジトリをクローンして、このガードレールパイプラインをエンドツーエンドで実行してください。これには、完全な実装、クラスラー設定、およびデプロイガイドが含まれています。

さらに詳しく知るには、設定オプションやサポートされているソース/シンクについてリアルタイムモードのドキュメントを参照するか、アーキテクチャ全体の解説についてリアルタイムモードの技術ディープダイブをご覧ください。

リソース

• リアルタイムモードの技術的ディープダイブ：Apache Spark™に300ミリ秒未満のストリーミングを構築した方法
• Databricks上の運用ワークロード向けに1秒未満のレイテンシを実現
• Databricks リアルタイムモード ドキュメント
• 付属コードリポジトリ - リアルタイムモードのガードレールデモ - 包括的なテストスイート、クラスター構成、デプロイガイドを完備した、本番環境に対応した実装

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