Feature Views を使ったリアルタイム不正検出

リアルタイムの不正検知は、顧客が数秒前に行った行動、通常の支出傾向、そして今回の支払いの比較という3種類の特徴量に依存します。これらの特徴量は、モデルのトレーニングとライブスコアリングで同じ方法で計算される必要があります。そうしないと、データリークやトレーニング・サービングスキューによってモデルの品質が徐々に低下してしまいます。Databricks Feature Storeの一部であるFeature Viewは、この課題を解決します。各特徴量を一度定義するだけで、その定義からポイントインタイム（時点）に正確なトレーニングデータと、オンライン推論用のマネージドパイプラインの両方が生成されます。

900ドルの支払いは、ある顧客にとっては日常茶飯事でも、別の顧客にとっては異例の金額である場合があります。支払いが実行された際、不正検知モデルがそれが正常なものか、審査が必要なものか、あるいはブロックすべきかを判断できる時間はわずか数秒しかありません。不正な取引を承認してしまうと、顧客の金銭的損失につながります。逆に正当な取引をブロックしてしまうと、支払い体験が損なわれ、プラットフォームへの信頼が低下します。

取引データ単体では全容を把握できません。モデルは、その顧客が普段どのように取引を行っているか、そして数秒、数分、数日、あるいは数週間前に何をしたかを知る必要もあります。トレーニング時には、各支払いが行われた時点で利用可能であった情報のみを使用して特徴量を計算する必要があります。ライブスコアリング時には、不審なアクティビティをすばやく検出するために、最近の取引を十分に反映させる必要があります。2つのパスでロジックを別々に実装すると、微細な差異が生じてしまいます。

このウォークスルーでは、Kafkaストリームからの900ドルの支払いが、Model Servingによって提供される不正スコアに至るまでの流れを、各ステップのコードとともに追います。核心にあるのは、トレーニングとサービングの両方を駆動する1つの特徴量定義により、ポイントインタイムで正確なトレーニングデータと一貫性のあるオンライン特徴量が得られるという点です。パスごとに個別のパイプラインを構築する必要はありません。

Feature Viewとは何ですか？

Feature Viewは、特徴量のソース、エンティティキー、時間的セマンティクス、および変換を指定するDatabricks Feature Storeの定義です。同じ定義によって、トレーニング用の過去の値が計算され、サービング向けに特徴量を最新の状態に保つマネージドパイプラインが駆動されます。1つのAPIでバッチ、ストリーミング、およびオンデマンドの特徴量がカバーされます。Unity Catalogにより、それらの定義、権限、およびリネージがガバナンス管理されます。

この設計が求められる背景には、2つの失敗モードがあります。

• トレーニング・サービングスキューは、特徴量がトレーニング時と本番環境で異なる方法で計算された場合に発生します。推論時に、モデルが学習時と一致しない入力を受け取る原因となります。
• データリークは、予測時に利用できなかった情報がトレーニングデータに含まれている場合に発生します。ポイントインタイムで正確な特徴量を使用することで、予測対象の時点より前に発生したイベントのみを利用し、これを防ぎます。

Feature Viewsを使用すると、ストリーミング特徴量の計算を、過去のトレーニングやライブスコアリングと同じデータサイエンスワークフローに統合できます。過去の取引に対して特徴量を評価し、同じ定義を使用してライブ決定のための顧客コンテキストを最新の状態に保つことができます。

リアルタイム不正検知モデルにはどのような特徴量が必要ですか？

目標は、正当な支払いに誤検知を発生させることなく、不審な支払いをリアルタイムでフラグ付けすることです。900ドルの支払い単体で見れば問題ないように思えるかもしれません。しかし、それが5分間で10回目の支払いである場合、不審な連続アクセスのの一部である可能性があります。その連続アクセスを認識する遅延が生じるほど、侵害されたアカウントにアクセスした第三者が追加の支払いを行う時間を与えてしまうことになります。

Databricks Appsで構築された不正検知アプリケーションは、金融取引のストリームを処理します。各イベントには、顧客ID、金額、支払い方法、タイムスタンプ、ならびに発信国および宛先国が含まれます。アプリケーションは各取引をModel Servingエンドポイントに送信します。その後、ライブの取引量、フラグ付けされた取引、および国間のフローとともに、取引のリスクスコアが表示されます。
 

モデルは、3種類のコンテキストを組み合わせます。

• 最近のアクティビティ： RollingWindowを使用して計算される5分間の取引回数と10分間の合計金額により、顧客の支払い頻度と支出額が捉えられます。
• 支出の基準値： SawtoothWindowで維持される30日間の平均と標準偏差により、顧客の通常の支払い金額とその変動の大きさが記述されます。
• 取引の詳細：今回の支払いの金額、支払い方法、ならびに発信国および宛先国が、判断のためのコンテキストを提供します。

このアーキテクチャには、トレーニング用の過去のパスと、推論用のオンラインパスがあります。どちらもUnity Catalogの同じ特徴量定義から開始されます。過去のパスは、Streamの取り込みテーブルを読み込んでポイントインタイムのトレーニングセットを構築します。オンラインパスは、集計データをOnline Feature Storeにマテリアライズし、Model Servingがリクエスト時にそれらを読み込みます。

Kafkaからストリーミング特徴量を定義する方法

提供する特徴量を決定する前に、過去の取引に対して代替案をテストできます。5分間の取引回数は、15分間の回数よりも強力なシグナルとなりますか？今回の金額と顧客の支出の変動性を比較することは、平均値単体を使用する以上の効果がありますか？

この例では、顧客の取引イベントがKafkaストリームとして到着します。登録されたStreamが接続、スキーマ、および取り込み設定を保持し、 StreamSourceがそれを参照してリアルタイム特徴量を計算します。Kafkaトピックは、モデルトレーニングに必要なだけの十分な履歴を保持しません。そのため、Databricksは取り込みテーブル内でストリームのオフラインコピーを管理します。トレーニングやオンラインストアのバックフィルのために、Databricksは取り込みテーブルから過去の値を計算し、最新のデータにはライブのKafkaフィードを使用します。

取り込み時に、 transformation_sqlはKafkaの valueペイロードからのネストされたフィールドを、特徴量に必要なスキーマに再構築します。プロジェクションはソース定義の一部となるため、個別の前処理パイプラインをデプロイする必要はありません。今回の支払いについて、Kafkaの valueペイロードには以下が含まれます。

デモでは、 demo_user_idを customer_idとともに使用して、異なるアプリユーザーの顧客履歴を分離します。

 RollingWindowを使用して5分間のカウントを定義します。付属のノートブックには、インポート、設定、および完全なソースプロジェクションが含まれています。

ノートブックで compute_features(features=[recent_transaction_count])を使用して、その履歴値を確認できます。10分間の合計には、同じソースとキーを使用し、10分間の RollingWindowにわたるSumを使用します。

ソートゥースウィンドウ（Sawtooth window）は、最も古い境界が日単位で進む長期のルックバック集計です。バッチで計算された履歴と、最近のアクティビティのストリーミング更新を組み合わせます。900ドルの支払いでは、 SawtoothWindowによって30日間の支出基準値が短いローリングウィンドウに追加されます。どちらの種類のウィンドウも、最近のアクティビティを最新の状態に保ちます。ソートゥースウィンドウの最も古い境界は連続的ではなく日単位で移動するため、有効なルックバック期間は30日よりわずかに長くなります。基準値では、同じソートゥースウィンドウでの支払い金額に対して、組み込みの Avg、 StddevPop、および Countの集計を使用します。

Streamが作成されると、Databricksは管理対象の取り込みパイプラインを開始し、着信イベントを取り込みテーブルに保持します。Streamが存在する前のイベントが必要な場合も、 StreamBackfillSourceを使用すると、Streamのセットアップ時に独自の履歴を提供できます。テーブルが完全なルックバック期間をカバーすると、イベントが蓄積されるのを30日間待つことなく、新しいソートゥース特徴量をすぐに使用できるようになります。

ライブの決済データと顧客のベースラインを比較する方法

オンデマンド特徴量は、スコアリングリクエストの値と保存された特徴量を組み合わせて、リクエスト時に計算されます。ライブのトランザクションを顧客の過去のベースラインと比較することは、異常な挙動を見つけるための一般的な方法です。Feature Viewsでは、推論時の情報は RequestSource（ここではアプリケーションのリクエストからの transaction_amount）で表されます。

2人の顧客の平均決済額がどちらも300ドルであっても、その使い方は大きく異なる場合があります。標準偏差が100ドルと600ドルの場合、同じ900ドルの決済でも、一方の顧客の平均からは標準偏差6回分上回るのに対し、もう一方の顧客の平均からは1回分しか上回りません。モデルはこの比較を最近のアクティビティや決済の詳細と組み合わせることで、不正リスクを推定します。

Sawtooth集計は30日間の平均と標準偏差を維持します。オンデマンド特徴量はこれらを現在の金額と組み合わせて (current_amount - average) / standard_deviationを計算します。この計算は、Unity Catalogに登録されたPython UDFとして実装します：

特徴量の定義では、登録済みのアップストリームのFeatureオブジェクトを使用します。 FeatureViewSourceは現在の金額と3つの30日間統計をまとめ、 CustomUDFはその関数を呼び出します：

十分な利用可能な履歴がない場合、関数はフォールバックとしてゼロを返します。別のインジケーターを使用することで、モデルはそのフォールバックを真のゼロ（つまり、決済が顧客の平均と一致していること）と区別できるようになります。同じ派生特徴量は、トレーニング時には過去の入力に対して評価され、推論時には現在のリクエストと取得した統計に対して評価されます。

デプロイする前に特徴量の候補をテストする方法

特徴量の開発中、ノートブックで compute_features() を使用して特徴量の定義を検査できます。これは、入力が完全に過去のソースデータから取得される派生特徴量に対して機能します。 amount_deviation のようなリクエストに依存する比較の場合は、代わりに create_training_set() を使用してください。これにより、ラベル付きの例から現在の決済フィールドが提供されます。Unity Catalogの関数が存在している必要がありますが、候補をテストしている間は特徴量の定義を未登録のままにしておくことができます。

これにより、ライブスコアリング用にデプロイする前に、特徴量が不正検出モデルのオフラインパフォーマンスを向上させるかどうかを確認できます。たとえば、後続のホールドアウト決済に対して、 amount_deviation の有無によるモデルを比較できます。これらの実験はトランザクション履歴に対するバッチコンピュートで実行されるため、オンラインのマテリアライゼーションを設定せずに各候補を評価できます。特徴量を選択したら、その定義を登録し、サービング用のローリング集計とSawtooth集計をマテリアライズします。

時点に正確なトレーニングデータを作成する方法

過去の各トレーニング例には、その決済のタイムスタンプ時点り計算された特徴量が必要です。午前10:00の900ドルの決済の場合、5分間のカウントには、午前9:55から午前10:00まで（午前10:00自体は含めない）のトランザクションが含まれている必要があります。午前10:03に取得されたカウントには、午前10:02に行われた決済が含まれている可能性があり、これは元の決済がスコアリングされた時点では利用できませんでした。これがデータリークです。モデルは将来のアクティビティから学習するため、オフラインでのパフォーマンスが実際よりも優れているように見えてしまいます。Feature Viewsは、各例の顧客キーとタイムスタンプを使用して、Streamの取り込みテーブル内のイベントから正しい履歴ウィンドウを計算します。

選択した特徴量を登録した後、 create_training_set() はモデルのトレーニングデータを構築します。ラベル付きのDataFrameは、顧客キー、イベント時刻、現在の決済フィールド、および is_fraud を提供します。この抜粋では、30日間のベースラインを含む、チュートリアルで登録された特徴量を選択し、トレーニング用に選択したカットオフ前の行を使用します：

キーとタイムスタンプによって顧客履歴が選択されますが、モデルの入力からは除外されます。リクエストフィールドは特徴量の計算にフィードされ、分類器は結果として得られた名前付き特徴量を使用します。後続の行は、同じ定義を使用して検証セットを形成します。

ランダムフォレスト分類器をトレーニングし、後者で平均精度とROC AUCを評価します。 その後、Feature Engineeringクライアントを介してモデルをログに記録します。 training_set=train_set を渡すことで、サービングに必要な特徴量の依存関係が記録されます：

オンライン推論のためにリアルタイム特徴量を提供する方法

上記で選択した特徴量をライブスコアリングに移行するには、パイプラインでストリーミング集計を維持し、過去データを処理し、オンラインストアを最新の状態に保つ必要があります。これらのパイプラインの構築と運用には、不正検出モデルの開発に加えて、数カ月のデータエンジニアリング作業が追加で必要になります。Feature Viewsでは、登録された特徴量の定義がこのプロセスを駆動します。 materialize_features() を呼び出すことで管理されたパイプラインが作成され、Lakebaseを基盤とするDatabricks Online Feature Storeで集計を利用できるようになります。

ストリーミングパスでは Spark Real-Time Mode (RTM) を使用して、受信イベントを継続的に処理し、到着したローリング集計を更新します。

ローリング集計とSawtooth集計をマテリアライズします。30日間のSawtoothベースラインについて、Databricksはバッチコンピュートで過去の部分を、ストリーミングで最近のアクティビティを維持します。これら2つは読み込み時に単一の特徴量にマージされ、 StreamingMode() が別のバッチ呼び出しなしでこのハイブリッドパスを管理します。サービングの前に、Catalog Explorerで両方のマテリアライゼーションパスを確認してください。Sawtoothのガイダンスでは、本番環境で使用する前に丸2日間待つことを推奨しています。

RequestSource は、スコアリングリクエストごとに値を供給し、派生フィーチャはオンデマンドで計算されます。モデルがデプロイされると、アプリケーションは顧客のキーと現在の支払いフィールドをそのエンドポイントに送信します。Model Servingは最新の顧客フィーチャを取得してオンデマンドの偏差を評価し、その入力を不正検知モデルに渡します。以下は、スコアラーからのSDK呼び出しを使用した、アプリのリクエスト形式での900ドルの支払いデータの例です。

アプリはこのリクエストで event_time を送信しますが、オンラインルックアップではそのタイムスタンプによって選択された過去のスナップショットではなく、利用可能な最新の顧客フィーチャが使用されます。このデモでは、アプリは各支払いをKafkaにパブリッシュする前にスコアリングします。これにより、現在の支払いが自身の履歴から除外されるため、トレーニングウィンドウにおけるその支払いの除外と一致します。フィーチャの更新とスコアリングは引き続き非同期で実行されるため、次の支払いのスコアリング時には他のイベントが処理中である可能性があります。

エンドポイントは、 fraud_probability で確率を返し、 fraud_prediction でフラグを返します。アプリケーションはスコアリングされたトランザクションをダッシュボード用にLakebaseに保存します。

Unity CatalogでMLフィーチャをガバナンスする方法

本番稼働する不正検知モデルでは、どのモデルがどのフィーチャを使用しているか、どこから取得されたか、誰が読み取れるかといったガバナンスの疑問が生じます。Unity Catalogは、フィーチャの定義、権限、リネージを統合してこれらの疑問に答えます。MLflowは各モデルが使用するフィーチャを記録し、Model Servingは推論リクエストの処理時にそれらの依存関係を使用します。

フィーチャの権限によってライフサイクルを制御します。

• CREATE FEATURE は登録を制御します。
• READ FEATURE はフィーチャのメタデータへのアクセスを許可します。
• MANAGE はマテリアライゼーションなどの操作を承認します。

フィーチャの権限は、トレーニングやサービングのためにソーステーブルまたはマテリアライズドテーブルに必要な SELECT 権限を置き換えるものではありません。

Catalog Explorerでは、フィーチャビューは定義、ソースリンク、マテリアライズドテーブル、ステータスとともに Functions の下に表示されます。ストリーミングのマテリアライゼーションは、検査のためにシステム管理のLakeflowパイプラインにリンクします。 マテリアライゼーションタグを使用すると、チームはパイプラインやジョブのコンピュートコストを特定のチームやプロジェクトに割り当てることができます。

リアルタイム不正検知のための重要なポイント

私たちの900ドルの支払いには、最近のアクティビティ、長期的なベースライン、現在のリクエストというコンテキストが追加されました。それを可能にした3つのプラクティスは以下の通りです。

• 一度定義し、二度使用する。 トレーニングとサービングでフィーチャ定義を共有することで、実装の違いに起因するトレーニング・サービングの乖離を軽減します。
• 時点 (Point in time) で計算する。 将来のイベントを除外したトレーニングウィンドウにより、データリークを防ぎ、オフラインのメトリクスを正確に保ちます。
• プラットフォームにパイプラインを実行させる。 マテリアライゼーションとオンラインルックアップは管理されているため、より優れた不正シグナルの選択に集中できます。

このパターンは、このウォークスルーの範囲にとどまりません。 LastN は、顧客の直近10件の支払いの宛先国などのシーケンスを保持し、モデルが支払いルートの変化を検出するのに役立ちます。バッチフィーチャは、ライフタイム集計を使用してエンティティの全履歴を要約することもできます。

サブ秒単位のストリーミングの鮮度を支えるインフラストラクチャの詳細については、 How Databricks Feature Store serves features with sub-second freshness をご覧ください。実装の詳細については、 Feature Views API、 マテリアライゼーション、および Unity Catalogでのフィーチャビューの探索のドキュメントを参照してください。

FAQ

Databricks Feature Viewsとは何ですか？

フィーチャビューは、MLフィーチャを一度定義してトレーニングと本番の両方で使用するためのDatabricks Feature Storeの機能です。各定義では、ソース、エンティティキー、時間的セマンティクス、変換を指定します。同じAPIでバッチ、ストリーミング、オンデマンドのフィーチャをカバーし、Unity Catalogがそれらをガバナンスします。

不正検知モデルのトレーニング時に、フィーチャビューはどのようにデータリークを防ぎますか？

フィーチャビューは、各トレーニングサンプルのタイムスタンプ時点でのフィーチャを計算します。午前10:00の支払いの場合、5分間のトランザクション数には、午前9:55から午前10:00まで（午前10:00の支払い自体は除く）の支払いが含まれます。これにより、将来のアクティビティがトレーニングデータに含まれるのを防ぎ、オフラインのパフォーマンスが過大評価されるのを防ぎます。

トレーニング・サービングの乖離とは何ですか？また、フィーチャビューはどのようにそれを軽減しますか？

トレーニング・サービングの乖離は、トレーニング用とライブ推論用でフィーチャの計算方法が異なる場合に発生します。フィーチャビューは、過去の計算とオンラインマテリアライゼーションの両方に単一の登録済み定義を使用することで、この乖離を軽減します。チームは本番環境用にフィーチャロジックを再実装する必要がありません。

フィーチャビューはKafkaストリームからフィーチャを計算できますか？

はい。登録されたStreamはKafkaの接続、スキーマ、取り込み設定を保持し、 StreamSource はフィーチャ定義内でそれを参照します。Databricksは、トレーニングとバックフィルのために、ストリームのオフラインコピーを取り込みテーブルに保持します。リアルタイムモードでは、新しいイベントが到着するとストリーミング集計が更新されます。

本番環境の不正検知モデルには、どのようにリアルタイムフィーチャが提供されますか？

 materialize_features() を呼び出すと、Lakebaseを基盤とするOnline Feature Storeの集計を最新の状態に保つ管理されたパイプラインが作成されます。モデルが log_model() でログに記録され、Model Servingにデプロイされると、エンドポイントは最新の顧客フィーチャを取得し、リクエストからオンデマンドフィーチャを計算します。その後、不正の確率を返します。

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