Azure Machine Learningの特徴量ストア機能でできることを理解する

こんばんは。今日はAzure Machine Learningの特徴量ストア機能について調べて触れてみたので、わかったことなどをメモしておきたいと思います。

それでは参ります。

特徴量ストアとは？

こちらの記事が分かりやすかったので、リンクを貼っておきます。

Feature Storeについてふんわり理解する – Re:ゼロから始めるML生活 (nogawanogawa.com)

ChatGPTにも解説をお願いしてみました。

——– (以下、ChatGPTによる解説）——-

特徴量ストアの概念

特徴量ストアとは、データサイエンスや機械学習のプロジェクトでよく使われる「特徴量（features）」を管理、保存、再利用できるようにする仕組みだよ。特徴量ってのは、元のデータから何らかの計算や加工をして作られる属性で、機械学習モデルが学習する際に重要な入力になるんだ。

特徴量ストアを使うと、特徴量の再利用が楽になり、他のデータサイエンティストやエンジニアとの協業もスムーズになる。さらに、実稼働環境での予測などに使われる特徴量が、訓練時のそれと一致することを保証できるから、品質の確保にも役立つんだ。

—————

合わせて、特徴量ストアの紹介でよく登場するUberの事例記事も要約してもらいました。

https://www.uber.com/en-ID/blog/michelangelo-machine-learning-platform/

——– (以下、ChatGPTによる要約）——-

Michelangeloには、特徴量（フィーチャー）を管理するための「特徴量ストア」という機能があります。このストアは、Uber内のさまざまなチームが作成した特徴量を一元管理し、それらを共有できるようにする目的で設計されています。特徴量ストアには約10,000の特徴量が登録されており、これらは日々自動で計算・更新されます。

なぜ重要か？

1. 再利用と効率性: 同じまたは類似の特徴量を使用する多くのモデリング問題が存在するため、特徴量の再利用が可能になり、効率が向上します。
2. データ品質: 特徴量が一元管理されているため、データ品質が向上し、モデルの性能が高まる可能性があります。
3. 簡易性とアクセス: 特徴量ストアから簡単に特徴量を取得できるため、モデル設計とデプロイが容易になります。

Azure Machine Learningの特徴量ストア機能

概要

以下記事の通り、Azure Machine Learningの特徴量ストア機能は、2023年5月のMicrosoft Build 2023でパブリックプレビューとして発表された、比較的新しい機能となっています。

Azure Machine Learning – Public Preview for Build | Azure の更新情報 | Microsoft Azure

マネージド Feature Store とは – Azure Machine Learning | Microsoft Learn

この機能の特徴もChatGPTに解説をお願いしてみました。

——– (以下、ChatGPTによる解説）——-

Azure Machine Learning（Azure ML）の特徴量ストアも基本的には上記と同じ機能を提供するけど、Azure独自の特長がいくつかある。

1. 統合された環境: Azure MLはAzureの他のサービス（例：Azure Data Lake StorageやAzure SQL Databaseなど）と簡単に連携できる。特徴量ストアもこれらのサービスとシームレスに動作する。
2. スケーラビリティ: Azureはクラウドサービスだから、必要に応じてリソースをスケールすることができる。大量の特徴量データを扱う必要がある場合でも安心だよ。
3. セキュリティとガバナンス: Azureは企業向けの厳格なセキュリティとデータガバナンス機能を提供している。特徴量ストアもこれらの機能を利用できるから、データの安全性と整合性が保たれる。
4. 使いやすいUIとSDK: Azure ML特徴量ストアは、使いやすいウェブベースのUIと、Python SDKを提供しているから、コードを書くのが得意な人もそうでない人も使いやすい。

以上のように、Azure MLの特徴量ストアは一般的な特徴量ストアの機能に加えて、クラウドの柔軟性とセキュリティ、使いやすさを兼ね備えているんだ。

—————

特徴量ストアへ登録する時の流れ

こちらもChatGPTにまとめてもらいました。

１：特徴量セットの定義:
特徴量の名前、データ型、変換ロジックなどを定義する。

from azureml.core import Workspace
from azureml.core.feature import Feature, FeatureSet

features = [
    Feature("age", "int64"),
    Feature("income", "float64")
]
feature_set = FeatureSet(name="my_feature_set", features=features)

２：特徴量セットの登録:
定義した特徴量セットをAzure MLの特徴量ストアに登録する。

workspace = Workspace.from_config()
feature_set.register(workspace=workspace)

３：データのマテリアライゼーション（オフラインストア）:
特徴量データを計算し、オフラインストア（例えば、Azure Data Lake Storage）に保存する。

from azureml.core.feature_materialization import MaterializationConfig

materialization_config = MaterializationConfig(
    offline_datastore_name="my_offline_datastore"
)
feature_set.materialize(materialization_config)

４：データのマテリアライゼーション（オンラインストア）:
特徴量データを計算し、オンラインストア（例えば、Azure Redis Cache）に保存する。

materialization_config = MaterializationConfig(
    online_datastore_name="my_online_datastore"
)
feature_set.materialize(materialization_config)

特徴量ストアに保存したデータを利用する時の流れ

１：特徴量の検索と選択：
特徴量ストアから必要な特徴量を検索して選択する。

from azureml.core.feature import FeatureSet

workspace = Workspace.from_config()
feature_set = FeatureSet.get(workspace, name="my_feature_set")

２：モデルトレーニング：

選択した特徴量を使ってモデルを訓練する。

from sklearn.ensemble import RandomForestClassifier
from azureml.core.feature import get_feature_data

# 特徴量データを取得
X_train, y_train = get_feature_data(feature_set, "train_data")

# モデルの訓練
clf = RandomForestClassifier()
clf.fit(X_train, y_train)

３：モデルのデプロイ:
訓練したモデルをデプロイし、リアルタイムで特徴量を取得する設定も可能。

from azureml.core import Model
from azureml.core.webservice import AciWebservice

# モデルの登録
model = Model.register(workspace, model_path="my_model.pkl", model_name="my_model")

# モデルのデプロイ
deployment_config = AciWebservice.deploy_configuration()
service = Model.deploy(workspace, "my_service", [model], deployment_config)

４：モニタリングとメンテナンス:
特徴量やモデルのパフォーマンスを定期的にチェックする。

from azureml.core import Experiment

# モデルのパフォーマンスをログに保存
experiment = Experiment(workspace, "my_experiment")
run = experiment.start_logging()
run.log("accuracy", 0.95)
run.complete()

実際に使ってみた

さて、概要がわかったところで、この機能を実際に使ってみたいと思います。

以下のチュートリアルで一連の流れを体感できますので、私も今回こちらを試してみました。

チュートリアル #1: マネージド Feature Store (プレビュー) を使用して特徴量セットの開発と登録を行う – Azure ML managed Feature Store – Basics | Microsoft Learn

以下、気になったポイントを中心にメモ。

０−１：ノートブックの準備と実行環境のセットアップ

Docsにも書いてますが（ちゃんと読んでなくて詰まりました笑）、NotebookをAzure MLに実行する際、Spark Computerのセッションの構成から、パッケージの構成を記述したCondaファイルを設定しておかないとImportエラーでコードが動かないので注意。

０−２：特徴量ストアリソースの作成

まず、特徴量ストアリソースを別途作る必要があるんですね。

## 特徴量ストアリソースパラメータの構成
# We use the subscription, resource group, region of this active project workspace.
# You can optionally replace them to create the resources in a different subsciprtion/resourceGroup, or use existing resources
import os

featurestore_name = "my-featurestore"
featurestore_location = "eastus"
featurestore_subscription_id = os.environ["AZUREML_ARM_SUBSCRIPTION"]
featurestore_resource_group_name = os.environ["AZUREML_ARM_RESOURCEGROUP"]

## 特徴量ストアリソースの作成
!az ml feature-store create --subscription $featurestore_subscription_id --resource-group $featurestore_resource_group_name --location $featurestore_location --name $featurestore_name

１：特徴量ストアの定義

以下のコードを実行することで特徴量ストアが定義できます。

特徴ストアへのデータの登録は、現在SDK/CLI経由でのみ可能なようです。

特徴量ストアが作成されたら、次にサンプルデータで試し使いしていくステップに。こんなデータに対して・・・

以下のようなデータ形式の特徴量セットを作成。

上記のコード、馴染みがない書き方ですが、以下に解説がありました。

CLI (v2) 特徴量セット仕様 YAML スキーマ – Azure Machine Learning | Microsoft Learn

これに照らすと、上記コードの意味は以下のようになるっぽい。

transactions_featureset_code_path = (
    root_dir + "/featurestore/featuresets/transactions/transformation_code"
)

transactions_featureset_spec = create_feature_set_spec(
    source=FeatureSource(

        // ソースデータの種類：parquet
        type=SourceType.parquet,

        // ソースデータのパス
        path="wasbs://data@azuremlexampledata.blob.core.windows.net/feature-store-prp/datasources/transactions-source/*.parquet",

        // タイムスタンプを表す列は"timestamp"列
        timestamp_column=TimestampColumn(name="timestamp"),

        // ソースデータの遅延：20分
        source_delay=DateTimeOffset(days=0, hours=0, minutes=20),
    ),
    transformation_code=TransformationCode(
        // 変換コード定義が配置されているフォルダ
        path=transactions_featureset_code_path,

        // 上記フォルダ内の変換コードが記述されているファイルのクラス名
        transformer_class="transaction_transform.TransactionFeatureTransformer",
    ),

    //　インデックス列は"accountID"
    index_columns=[Column(name="accountID", type=ColumnType.string)],

    // ソースデータのルックバック：7日
    source_lookback=DateTimeOffset(days=7, hours=0, minutes=0),

    // 一時的な結合のループバック：1日
    temporal_join_lookback=DateTimeOffset(days=1, hours=0, minutes=0),

    // スキーマの推論：True
    infer_schema=True,
)

transformation_code_pathで指定されている内容もみてみた。なるほど、上の定義でtransformer_classで「transaction_transform.TransactionFeatureTransformer」と指定されていましたが、これはこのtransaction_transform.pyのTransactionFeatureTransformerクラスのことを指していたのですね。

特徴量作成の処理実体はこちらで記述されている模様。

#transaction_transform.pyファイル

from pyspark.sql import functions as F
from pyspark.sql.window import Window
from pyspark.ml import Transformer
from pyspark.sql.dataframe import DataFrame


class TransactionFeatureTransformer(Transformer):
    def _transform(self, df: DataFrame) -> DataFrame:
        days = lambda i: i * 86400
        w_3d = (
            Window.partitionBy("accountID")
            .orderBy(F.col("timestamp").cast("long"))
            .rangeBetween(-days(3), 0)
        )
        w_7d = (
            Window.partitionBy("accountID")
            .orderBy(F.col("timestamp").cast("long"))
            .rangeBetween(-days(7), 0)
        )
        res = (
            df.withColumn("transaction_7d_count", F.count("transactionID").over(w_7d))
            .withColumn(
                "transaction_amount_7d_sum", F.sum("transactionAmount").over(w_7d)
            )
            .withColumn(
                "transaction_amount_7d_avg", F.avg("transactionAmount").over(w_7d)
            )
            .withColumn("transaction_3d_count", F.count("transactionID").over(w_3d))
            .withColumn(
                "transaction_amount_3d_sum", F.sum("transactionAmount").over(w_3d)
            )
            .withColumn(
                "transaction_amount_3d_avg", F.avg("transactionAmount").over(w_3d)
            )
            .select(
                "accountID",
                "timestamp",
                "transaction_3d_count",
                "transaction_amount_3d_sum",
                "transaction_amount_3d_avg",
                "transaction_7d_count",
                "transaction_amount_7d_sum",
                "transaction_amount_7d_avg",
            )
        )
        return res

その他Lookbackや一時的な結合の設定の意味はいまひとつよく分からない・・・笑　一時的な結合のループバック（temporal_join_lookback）については、以下に少しヒントになりそうなことは書いてましたが・・まだプレビューだからですかね、今後このあたりの記載がもう少し充実したらいいですね・・

マネージド Feature Store のエラーのトラブルシューティング – Azure Machine Learning | Microsoft Learn