教學課程 3：啟用週期性具體化並執行批次推斷

本教學課程系列會示範功能如何順暢地整合機器學習生命週期的所有階段：原型設計、定型和運算化。

這很重要

Azure Cache for Redis 宣布了所有 SKU 的退役時間表。 我們建議盡快將現有的 Azure Cache for Redis 實例移至 Azure Managed Redis 上。

遷移指引：

有關退休的更多詳細資訊：

教學一展示了如何透過自訂轉換建立特徵集規格，然後用該特徵集產生訓練資料、啟用物質化，以及進行回填。 教學二展示了如何實驗特徵以提升模型效能，然後訓練並註冊使用特徵的模型。

本教學課程說明如何：

  • 啟用 transactions 功能集的週期性具體化。
  • 在已註冊的模型上執行批次推斷管線。

必要條件

在你繼續這個教學之前，先完成本系列的教學1和2。

設定

  1. 配置 Azure Machine Learning Spark 筆記本。

    若要執行本教學課程，您可以建立新的筆記本，並逐步執行指示。 您也可以開啟並執行以下名稱的現有筆記本：3.啟用週期性具體化並執行批次推斷。 您可以在 featurestore_sample/notebooks 目錄中找到該筆記本，以及此系列中的所有筆記本。 您可以選擇 sdk_only 或 sdk_and_cli。 讓本教學課程保持開啟狀態，並參閱本教學課程以取得文件連結和更多說明。

    1. 在Compute 的下拉選單中，選擇 Azure Machine Learning Serverless Spark 下的 Serverless Spark Compute。

    2. 設定會話：

      1. 在頂端狀態列中選擇 設定工作階段。
      2. 選擇Python packages 標籤。
      3. 選取 上傳 Conda 檔案。
      4. 從本機電腦中選取 azureml-examples/sdk/python/featurestore-sample/project/env/online.yml 檔案。
      5. 選擇性增加工作階段逾時 (閒置時間)，以避免頻繁重新執行必要條件。

  2. 啟動 Spark 工作階段。

    # run this cell to start the spark session (any code block will start the session ). This can take around 10 mins.
print("start spark session")

  3. 設定範例的根目錄。

    import os

# please update the dir to ./Users/<your_user_alias> (or any custom directory you uploaded the samples to).
# You can find the name from the directory structure in the left nav
root_dir = "./Users/<your_user_alias>/featurestore_sample"

if os.path.isdir(root_dir):
    print("The folder exists.")
else:
    print("The folder does not exist. Please create or fix the path")

  4. 設定 CLI。

    不適用。

  1. 初始化專案工作區 CRUD (建立、讀取、更新和刪除) 用戶端。

    教學課程筆記本從此目前的工作區執行。

    ### Initialize the MLClient of this project workspace
import os
from azure.ai.ml import MLClient
from azure.ai.ml.identity import AzureMLOnBehalfOfCredential

project_ws_sub_id = os.environ["AZUREML_ARM_SUBSCRIPTION"]
project_ws_rg = os.environ["AZUREML_ARM_RESOURCEGROUP"]
project_ws_name = os.environ["AZUREML_ARM_WORKSPACE_NAME"]

# connect to the project workspace
ws_client = MLClient(
    AzureMLOnBehalfOfCredential(), project_ws_sub_id, project_ws_rg, project_ws_name
)

  2. 初始化特徵存儲變數。

    要反映你在教學 1 中創建的內容，請更新該 featurestore_name 值。

    from azure.ai.ml import MLClient
from azure.ai.ml.identity import AzureMLOnBehalfOfCredential

# feature store
featurestore_name = (
    "<FEATURESTORE_NAME>"  # use the same name from part #1 of the tutorial
)
featurestore_subscription_id = os.environ["AZUREML_ARM_SUBSCRIPTION"]
featurestore_resource_group_name = os.environ["AZUREML_ARM_RESOURCEGROUP"]

# feature store ml client
fs_client = MLClient(
    AzureMLOnBehalfOfCredential(),
    featurestore_subscription_id,
    featurestore_resource_group_name,
    featurestore_name,
)

  3. 初始化功能存放區 SDK 用戶端。

    # feature store client
from azureml.featurestore import FeatureStoreClient
from azure.ai.ml.identity import AzureMLOnBehalfOfCredential

featurestore = FeatureStoreClient(
    credential=AzureMLOnBehalfOfCredential(),
    subscription_id=featurestore_subscription_id,
    resource_group_name=featurestore_resource_group_name,
    name=featurestore_name,
)

在交易功能集上啟用週期性具體化

在教學 1 中，你啟用了實體化，並對 transactions 功能集執行了回填。 回填是按需執行的一次性操作，它會計算特徵值並將其放入物化存儲庫中。

若要在生產環境中處理模型的推斷，您可能想要設定週期性具體化作業，以讓具體化存放區保持最新狀態。 這些作業會在使用者定義的排程上執行。 週期性作業排程的運作方式如下：

  • 間隔和頻率值會定義視窗。 例如，下列值會定義一個三小時的時段：

    • interval = 3
    • frequency = Hour

  • 第一個窗口會從 start_time 中定義的 RecurrenceTrigger 值開始，以此類推。

  • 第一個週期性作業會在更新時間之後的下一個時間範圍開始時提交。

  • 後續的週期性作業會在第一個作業之後的每個時間範圍提交。

如先前的教學課程所述，在將資料具體化 (回填或週期性具體化) 後，功能擷取預設會使用已具體化的資料。

from datetime import datetime
from azure.ai.ml.entities import RecurrenceTrigger

transactions_fset_config = fs_client.feature_sets.get(name="transactions", version="1")

# create a schedule that runs the materialization job every 3 hours
transactions_fset_config.materialization_settings.schedule = RecurrenceTrigger(
    interval=3, frequency="Hour", start_time=datetime(2023, 4, 15, 0, 4, 10, 0)
)

fs_poller = fs_client.feature_sets.begin_create_or_update(transactions_fset_config)

print(fs_poller.result())

(選擇性) 儲存功能集資產的 YAML 檔案

請使用更新後的設定來儲存 YAML 檔案。

## uncomment and run
# transactions_fset_config.dump(root_dir + "/featurestore/featuresets/transactions/featureset_asset_offline_enabled_with_schedule.yaml")

執行批次推斷管線

批次推斷具有下列步驟：

  1. 你會使用與訓練流程中相同的內建特徵擷取元件來進行特徵擷取（詳見教學課程 2）。 針對管線訓練，您須提供功能擷取規格作為元件輸入。 針對批次推斷，您傳遞已註冊的模型以作為輸入。 元件在模型成品中尋找功能擷取規格。

    此外，針對訓練，觀察數據中包含目標變數。 不過，批次推斷觀察資料沒有目標變數。 功能擷取步驟會聯結觀察資料與功能，並輸出批次推斷的資料。

  2. 管線使用上一個步驟中的批次推斷輸入資料、在模型上執行推斷，並將預測值附加為輸出。

    注意

    在此範例中，您使用作業進行批次推斷。 你也可以在 Azure Machine Learning 中使用批次端點。

    from azure.ai.ml import load_job  # will be used later

# set the batch inference  pipeline path
batch_inference_pipeline_path = (
    root_dir + "/project/fraud_model/pipelines/batch_inference_pipeline.yaml"
)
batch_inference_pipeline_definition = load_job(source=batch_inference_pipeline_path)

# run the training pipeline
batch_inference_pipeline_job = ws_client.jobs.create_or_update(
    batch_inference_pipeline_definition
)

# stream the run logs
ws_client.jobs.stream(batch_inference_pipeline_job.name)

檢查批次推斷的輸出資料

在流程檢視中：

  1. 請在 inference_step 卡片中選取 outputs。

  2. 複製 Data 欄位值。 看起來像 azureml_995abbc2-3171-461e-8214-c3c5d17ede83_output_data_data_with_prediction:1。

  3. 將 Data 欄位值貼在下列資料格中，並分開名稱和版本值。 最後一個字元是版本，前面會加上冒號 (:)。

  4. 請注意批次推斷管線產生的 predict_is_fraud 欄。

    在批次推斷管線 (/project/fraud_mode/pipelines/batch_inference_pipeline.yaml) 的輸出中，系統建立了一個未追蹤的資料資產，名稱值為 GUID，版本值為 1。 這是因為您未提供 name 的 version 的 outputs 或 inference_step 值。 在此資料格中，您會先推導再顯示資產的資料路徑。

    inf_data_output = ws_client.data.get(
    name="azureml_1c106662-aa5e-4354-b5f9-57c1b0fdb3a7_output_data_data_with_prediction",
    version="1",
)
inf_output_df = spark.read.parquet(inf_data_output.path + "data/*.parquet")
display(inf_output_df.head(5))

清理

教學五：開發一個帶有自訂來源的功能集 ，說明如何刪除資源。

下一步

  • 上次更新時間：