教學課程：建立生產機器學習管線

發行項
05/21/2024

適用於：Python SDK azure-ai-ml v2 (目前)

注意

如需使用 SDK 第 1 版建置管線的教學課程，請參閱教學課程：建置影像分類的 Azure Machine Learning 管線

機器學習管線的用意是將完整的機器學習工作分割成多步驟工作流程。每個步驟都是可管理的元件，可個別開發、最佳化、設定和自動化。步驟透過定義明確的介面而連接起來。 Azure Machine Learning 管線服務自動協調管線步驟之間的所有相依性。使用管線的好處是可將 MLOps 做法標準化、有可調整的小組共同作業、有定型效率以及可以降低成本。若要深入了解管線的好處，請參閱什麼是 Azure Machine Learning 管線。

在本教學課程中，您會使用 Azure Machine Learning，透過 Azure Machine Learning Python SDK v2 建立可用於生產環境的機器學習專案。

這表示您將能夠利用 Azure Machine Learning Python SDK 來：

取得 Azure Machine Learning 工作區的控制代碼
建立 Azure Machine Learning 資料資產
建立可重複使用的 Azure Machine Learning 元件
建立、驗證和執行 Azure Machine Learning 管線

本教學課程進行期間，您會建立 Azure Machine Learning 管線以定型用於預測信用違約的模型。管線會處理兩個步驟：

資料準備
定型和註冊已定型的模型

接下來的影像顯示了一個您會在提交後於 Azure 工作室中看到的簡單管線。

兩個步驟依序為準備資料和定型。

圖表顯示管線的概觀。

下面這段影片會示範如何開始使用 Azure Machine Learning 工作室，以便您可以遵循教學課程中的步驟。此影片會示範如何建立筆記本、建立計算執行個體，以及複製筆記本。下列各節也會說明這些步驟。

必要條件

若要使用 Azure Machine Learning，您必須先有工作區。如果您沒有工作區，請完成建立要開始使用所需要的資源以建立工作區，並深入了解其使用方式。
登入 Studio，如果工作區尚未開啟，請選取您的工作區。
完成上傳、存取和探索資料教學課程，以建立本教學課程所需的資料資產。請務必執行所有程式碼以建立初始資料資產。如有需要，請探索資料並加以修改，但在本教學課程中，您只需要初始資料。
在您的工作區開啟或建立筆記本：
- 如果您想要將程式碼複製/貼到資料格中，請建立新筆記本。
- 或者，從工作室的 [範例] 區段開啟 tutorials/get-started-notebooks/pipeline.ipynb。然後選取 [複製] 以將筆記本新增至您的 [檔案]。 (查看可在哪裡找到範例。)

設定核心

在開啟的筆記本上方的頂端列上，如果您還沒有計算執行個體，請建立計算執行個體。
如果計算執行個體已停止，請選取 [啟動計算]，並等到其執行為止。
確定位於右上方的核心是 Python 3.10 - SDK v2。如果不是，則使用下拉式清單選取此核心。
如果您看到橫幅指出您需要進行驗證，請選取 [驗證]。

重要

本教學課程的其餘部分包含教學課程筆記本的儲存格。將其複製/貼入您的新筆記本中，或立即切換至您複製的筆記本。

設定管線資源

您可以從 CLI、Python SDK 或工作室介面使用 Azure Machine Learning 架構。在此範例中，您會使用 Azure Machine Learning Python SDK 第 2 版來建立管線。

在建立管線前，您需要下列資源：

用於定型的資料資產
要執行管線的軟體環境
要在其中執行作業的計算資源

建立工作區的控制代碼

在深入探討程式碼之前，您需要可參考工作區的方法。您會建立 ml_client 以獲得工作區的控制代碼。接著，您會使用 ml_client 來管理資源和作業。

在下一個資料格中，輸入您的訂用帳戶識別碼、資源群組名稱和工作區名稱。若要尋找這些值：

在右上方的 Azure Machine Learning 工作室工具列中，選取您的工作區名稱。
將工作區、資源群組和訂閱識別碼的值複製到程式碼。
您必須複製一個值、關閉區域並貼上，然後返回處理下一個值。

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

# authenticate
credential = DefaultAzureCredential()

SUBSCRIPTION = "<SUBSCRIPTION_ID>"
RESOURCE_GROUP = "<RESOURCE_GROUP>"
WS_NAME = "<AML_WORKSPACE_NAME>"
# Get a handle to the workspace
ml_client = MLClient(
    credential=credential,
    subscription_id=SUBSCRIPTION,
    resource_group_name=RESOURCE_GROUP,
    workspace_name=WS_NAME,
)

注意

建立 MLClient 時不會連線至工作區。用戶端初始化具有延遲性，會等到第一次需要進行呼叫時才開始 (這會在下一個程式碼資料格發生)。

對 ml_client 進行呼叫以驗證連線。因為這是您第一次呼叫工作區，因此系統可能會要求您進行驗證。

# Verify that the handle works correctly.
# If you ge an error here, modify your SUBSCRIPTION, RESOURCE_GROUP, and WS_NAME in the previous cell.
ws = ml_client.workspaces.get(WS_NAME)
print(ws.location, ":", ws.resource_group)

存取已註冊的資料資產

一開始先取得您先前在教學課程：在 Azure Machine Learning 中上傳、存取和探索資料中註冊的資料。

Azure Machine Learning 會使用 Data 物件來註冊可重複使用的資料定義，並在管線內取用資料。

# get a handle of the data asset and print the URI
credit_data = ml_client.data.get(name="credit-card", version="initial")
print(f"Data asset URI: {credit_data.path}")

建立管線步驟的作業環境

到目前為止，您已在計算執行個體 (即開發機器) 上建立開發環境。您也需要管線的每個步驟所使用的環境。每個步驟都可以有本身的環境，或者您可以對多個步驟使用一些通用的環境。

在此範例中，您會使用 conda yaml 檔案為作業建立 conda 環境。首先，建立一個用來儲存檔案的目錄。

import os

dependencies_dir = "./dependencies"
os.makedirs(dependencies_dir, exist_ok=True)

現在，在相依性目錄中建立檔案。

%%writefile {dependencies_dir}/conda.yaml
name: model-env
channels:
  - conda-forge
dependencies:
  - python=3.8
  - numpy=1.21.2
  - pip=21.2.4
  - scikit-learn=0.24.2
  - scipy=1.7.1
  - pandas>=1.1,<1.2
  - pip:
    - inference-schema[numpy-support]==1.3.0
    - xlrd==2.0.1
    - mlflow== 2.4.1
    - azureml-mlflow==1.51.0

規格包含一些您會在管線中使用的一般套件 (numpy、pip)，以及一些 Azure Machine Learning 特定套件 (azureml-mlflow)。

Azure Machine Learning 套件並非執行 Azure Machine Learning 作業的必要條件。不過，新增這些套件可讓您與 Azure Machine Learning 互動，以記錄計量和註冊模型，全都在 Azure Machine Learning 作業內完成。您會在本教學課程稍後的定型指令碼中加以使用。

使用 yaml 檔案，在您的工作區中建立和註冊此自訂環境：

from azure.ai.ml.entities import Environment

custom_env_name = "aml-scikit-learn"

pipeline_job_env = Environment(
    name=custom_env_name,
    description="Custom environment for Credit Card Defaults pipeline",
    tags={"scikit-learn": "0.24.2"},
    conda_file=os.path.join(dependencies_dir, "conda.yaml"),
    image="mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04:latest",
    version="0.2.0",
)
pipeline_job_env = ml_client.environments.create_or_update(pipeline_job_env)

print(
    f"Environment with name {pipeline_job_env.name} is registered to workspace, the environment version is {pipeline_job_env.version}"
)

建置定型管線

現在您已有執行管線所需的所有資產，接下來即可建置管線本身。

Azure Machine Learning 管線是可重複使用的 ML 工作流程，通常由數個元件組成。元件的一般使用年限為：

撰寫元件的 yaml 規格，或使用 ComponentMethod 以程式設計方式加以建立。
(選擇性) 使用名稱和版本在工作區中註冊元件，使其可重複使用且可共用。
從管線程式碼載入該元件。
使用元件的輸入、輸出和參數來實作管線。
提交管線。

有兩種方式可以建立元件定義、程式設計定義及 yaml 定義。接下來的兩個章節會引導您以這兩種方式建立元件。您可以建立兩個元件來嘗試這兩個選項，也可以挑選您慣用的方法。

注意

在本教學課程中，為了簡單起見，我們會對所有元件使用相同的計算。不過，您可以為每個元件設定不同的計算 (例如，透過新增類似 train_step.compute = "cpu-cluster" 的程式行)。若要檢視針對每個元件使用不同的計算來建置管線的範例，請參閱 cifar-10 管線教學課程中的基本管線作業一節。

建立元件 1：資料準備 (使用程式設計定義)

首先，我們建立第一個元件。此元件會處理資料的前置處理。前置處理工作會在 data_prep.py Python 檔案中執行。

首先，建立 data_prep 元件的來源資料夾：

import os

data_prep_src_dir = "./components/data_prep"
os.makedirs(data_prep_src_dir, exist_ok=True)

此指令碼會執行將資料分割成定型和測試資料集的簡單工作。 Azure Machine Learning 將資料集掛接為計算的資料夾，因此，我們建立了輔助 select_first_file 函式來存取已掛接的輸入資料夾內的資料檔案。

MLFlow 會在管線執行期間用來記錄參數和計量。

%%writefile {data_prep_src_dir}/data_prep.py
import os
import argparse
import pandas as pd
from sklearn.model_selection import train_test_split
import logging
import mlflow


def main():
    """Main function of the script."""

    # input and output arguments
    parser = argparse.ArgumentParser()
    parser.add_argument("--data", type=str, help="path to input data")
    parser.add_argument("--test_train_ratio", type=float, required=False, default=0.25)
    parser.add_argument("--train_data", type=str, help="path to train data")
    parser.add_argument("--test_data", type=str, help="path to test data")
    args = parser.parse_args()

    # Start Logging
    mlflow.start_run()

    print(" ".join(f"{k}={v}" for k, v in vars(args).items()))

    print("input data:", args.data)

    credit_df = pd.read_csv(args.data, header=1, index_col=0)

    mlflow.log_metric("num_samples", credit_df.shape[0])
    mlflow.log_metric("num_features", credit_df.shape[1] - 1)

    credit_train_df, credit_test_df = train_test_split(
        credit_df,
        test_size=args.test_train_ratio,
    )

    # output paths are mounted as folder, therefore, we are adding a filename to the path
    credit_train_df.to_csv(os.path.join(args.train_data, "data.csv"), index=False)

    credit_test_df.to_csv(os.path.join(args.test_data, "data.csv"), index=False)

    # Stop Logging
    mlflow.end_run()


if __name__ == "__main__":
    main()

現在，您已有可執行所需工作的指令碼，請從中建立 Azure Machine Learning 元件。

使用可執行命令列動作的一般用途 CommandComponent。此命令列動作可直接呼叫系統命令或執行指令碼。輸入/輸出可透過 ${{ ... }} 標記法指定於命令列上。

from azure.ai.ml import command
from azure.ai.ml import Input, Output

data_prep_component = command(
    name="data_prep_credit_defaults",
    display_name="Data preparation for training",
    description="reads a .xl input, split the input to train and test",
    inputs={
        "data": Input(type="uri_folder"),
        "test_train_ratio": Input(type="number"),
    },
    outputs=dict(
        train_data=Output(type="uri_folder", mode="rw_mount"),
        test_data=Output(type="uri_folder", mode="rw_mount"),
    ),
    # The source folder of the component
    code=data_prep_src_dir,
    command="""python data_prep.py \
            --data ${{inputs.data}} --test_train_ratio ${{inputs.test_train_ratio}} \
            --train_data ${{outputs.train_data}} --test_data ${{outputs.test_data}} \
            """,
    environment=f"{pipeline_job_env.name}:{pipeline_job_env.version}",
)

(選擇性地) 在工作區中註冊元件，以供日後重複使用。

# Now we register the component to the workspace
data_prep_component = ml_client.create_or_update(data_prep_component.component)

# Create (register) the component in your workspace
print(
    f"Component {data_prep_component.name} with Version {data_prep_component.version} is registered"
)

建立元件 2：定型 (使用 yaml 定義)

您建立的第二個元件會取用定型和測試資料、定型樹狀式模型，並傳回輸出模型。請使用 Azure Machine Learning 記錄功能來記錄和視覺化學習進度。

您使用 CommandComponent 類別建立了第一個元件。這次，您會使用 yaml 定義來定義第二個元件。每種方法各有優點。 yaml 定義實際上可與程式碼一起簽入，並提供可讀取的歷程記錄追蹤。採用內建的類別文件和程式碼完成時，使用 CommandComponent 的程式設計方法會更容易執行。

建立此元件的目錄：

import os

train_src_dir = "./components/train"
os.makedirs(train_src_dir, exist_ok=True)

在目錄中建立定型指令碼：

%%writefile {train_src_dir}/train.py
import argparse
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.metrics import classification_report
import os
import pandas as pd
import mlflow


def select_first_file(path):
    """Selects first file in folder, use under assumption there is only one file in folder
    Args:
        path (str): path to directory or file to choose
    Returns:
        str: full path of selected file
    """
    files = os.listdir(path)
    return os.path.join(path, files[0])


# Start Logging
mlflow.start_run()

# enable autologging
mlflow.sklearn.autolog()

os.makedirs("./outputs", exist_ok=True)


def main():
    """Main function of the script."""

    # input and output arguments
    parser = argparse.ArgumentParser()
    parser.add_argument("--train_data", type=str, help="path to train data")
    parser.add_argument("--test_data", type=str, help="path to test data")
    parser.add_argument("--n_estimators", required=False, default=100, type=int)
    parser.add_argument("--learning_rate", required=False, default=0.1, type=float)
    parser.add_argument("--registered_model_name", type=str, help="model name")
    parser.add_argument("--model", type=str, help="path to model file")
    args = parser.parse_args()

    # paths are mounted as folder, therefore, we are selecting the file from folder
    train_df = pd.read_csv(select_first_file(args.train_data))

    # Extracting the label column
    y_train = train_df.pop("default payment next month")

    # convert the dataframe values to array
    X_train = train_df.values

    # paths are mounted as folder, therefore, we are selecting the file from folder
    test_df = pd.read_csv(select_first_file(args.test_data))

    # Extracting the label column
    y_test = test_df.pop("default payment next month")

    # convert the dataframe values to array
    X_test = test_df.values

    print(f"Training with data of shape {X_train.shape}")

    clf = GradientBoostingClassifier(
        n_estimators=args.n_estimators, learning_rate=args.learning_rate
    )
    clf.fit(X_train, y_train)

    y_pred = clf.predict(X_test)

    print(classification_report(y_test, y_pred))

    # Registering the model to the workspace
    print("Registering the model via MLFlow")
    mlflow.sklearn.log_model(
        sk_model=clf,
        registered_model_name=args.registered_model_name,
        artifact_path=args.registered_model_name,
    )

    # Saving the model to a file
    mlflow.sklearn.save_model(
        sk_model=clf,
        path=os.path.join(args.model, "trained_model"),
    )

    # Stop Logging
    mlflow.end_run()


if __name__ == "__main__":
    main()

在此定型指令碼中您會看到，在定型模型後，模型檔案就會儲存並註冊到工作區。現在，您可以在推斷端點中使用已註冊的模型。

針對此步驟的環境，您會使用其中一個內建 (策展) Azure Machine Learning 環境。標籤 azureml 會指示系統在策展環境中尋找名稱。首先，建立描述元件的 yaml 檔案：

%%writefile {train_src_dir}/train.yml
# <component>
name: train_credit_defaults_model
display_name: Train Credit Defaults Model
# version: 1 # Not specifying a version will automatically update the version
type: command
inputs:
  train_data: 
    type: uri_folder
  test_data: 
    type: uri_folder
  learning_rate:
    type: number     
  registered_model_name:
    type: string
outputs:
  model:
    type: uri_folder
code: .
environment:
  # for this step, we'll use an AzureML curate environment
  azureml://registries/azureml/environments/sklearn-1.0/labels/latest
command: >-
  python train.py 
  --train_data ${{inputs.train_data}} 
  --test_data ${{inputs.test_data}} 
  --learning_rate ${{inputs.learning_rate}}
  --registered_model_name ${{inputs.registered_model_name}} 
  --model ${{outputs.model}}
# </component>

接著，建立並註冊元件。註冊元件可讓您在其他管線中重複使用元件。此外，具有工作區存取權的其他人也可以使用已註冊的元件。

# importing the Component Package
from azure.ai.ml import load_component

# Loading the component from the yml file
train_component = load_component(source=os.path.join(train_src_dir, "train.yml"))

# Now we register the component to the workspace
train_component = ml_client.create_or_update(train_component)

# Create (register) the component in your workspace
print(
    f"Component {train_component.name} with Version {train_component.version} is registered"
)

從元件建立管線

現在您已定義並註冊這兩個元件，接下來即可開始實作管線。

在此，您會使用輸入資料、分割比率 和 已註冊的模型名稱作為輸入變數。然後，呼叫元件，並透過其輸入/輸出識別碼加以連接。每個步驟的輸出都可透過 .outputs 屬性來存取。

由 load_component() 傳回的 Python 函式，運作方式會和我們在管線內用來呼叫每個步驟的任何一般 Python 函式相同。

若要撰寫管線的程式碼，您可以使用識別 Azure Machine Learning 管線的特定 @dsl.pipeline 裝飾項目。在裝飾項目中，我們可以指定管線描述和預設資源，例如計算和儲存體。如同 Python 函式，管線也可以有輸入。然後，您可以用不同的輸入建立單一管線的多個執行個體。

在此，我們使用輸入資料、分割比率 和 已註冊的模型名稱作為輸入變數。然後，我們呼叫元件，並透過其輸入/輸出識別碼加以連接。每個步驟的輸出都可透過 .outputs 屬性來存取。

# the dsl decorator tells the sdk that we are defining an Azure Machine Learning pipeline
from azure.ai.ml import dsl, Input, Output


@dsl.pipeline(
    compute="serverless",  # "serverless" value runs pipeline on serverless compute
    description="E2E data_perp-train pipeline",
)
def credit_defaults_pipeline(
    pipeline_job_data_input,
    pipeline_job_test_train_ratio,
    pipeline_job_learning_rate,
    pipeline_job_registered_model_name,
):
    # using data_prep_function like a python call with its own inputs
    data_prep_job = data_prep_component(
        data=pipeline_job_data_input,
        test_train_ratio=pipeline_job_test_train_ratio,
    )

    # using train_func like a python call with its own inputs
    train_job = train_component(
        train_data=data_prep_job.outputs.train_data,  # note: using outputs from previous step
        test_data=data_prep_job.outputs.test_data,  # note: using outputs from previous step
        learning_rate=pipeline_job_learning_rate,  # note: using a pipeline input as parameter
        registered_model_name=pipeline_job_registered_model_name,
    )

    # a pipeline returns a dictionary of outputs
    # keys will code for the pipeline output identifier
    return {
        "pipeline_job_train_data": data_prep_job.outputs.train_data,
        "pipeline_job_test_data": data_prep_job.outputs.test_data,
    }

然後，使用管線定義，以及您的資料集、選擇的分割率和您為模型挑選的名稱，將管線具現化。

registered_model_name = "credit_defaults_model"

# Let's instantiate the pipeline with the parameters of our choice
pipeline = credit_defaults_pipeline(
    pipeline_job_data_input=Input(type="uri_file", path=credit_data.path),
    pipeline_job_test_train_ratio=0.25,
    pipeline_job_learning_rate=0.05,
    pipeline_job_registered_model_name=registered_model_name,
)

提交作業

接著即可提交作業，以在 Azure Machine Learning 中執行。這次您會在 ml_client.jobs 上使用 create_or_update。

在此您也會傳入實驗名稱。實驗是對特定專案執行的所有反覆項目的容器。所有以相同實驗名稱提交的作業，都會在 Azure Machine Learning 工作室中相鄰列出。

完成後，管線會在您的工作區中註冊模型，作為定型的結果。

# submit the pipeline job
pipeline_job = ml_client.jobs.create_or_update(
    pipeline,
    # Project's name
    experiment_name="e2e_registered_components",
)
ml_client.jobs.stream(pipeline_job.name)

您可以使用上一個資料格中產生的連結，來追蹤管線的進度。當您第一次選取此連結時，您可能會看到管線仍在執行中。在其執行完成後，您就可以檢查每個元件的結果。

按兩下 [定型信用違約模型] 元件。

您會想要看到關於定型的重要結果有兩個：

檢視您的記錄：
1. 選取 [輸出 + 記錄] 索引標籤。
2. 開啟資料夾至 user_logs>std_log.txt 此區段會顯示指令碼執行 stdout。
檢視您的計量：選取 [計量] 索引標籤。此區段會顯示不同的已記錄計量。在此範例中， mlflow autologging 自動記錄了定型計量。

將模型部署為線上端點

若要了解如何將模型部署至線上端點，請參閱將模型部署為線上端點教學課程。

清除資源

如果您打算立即繼續進行其他教學課程，請跳至後續步驟。

停止計算執行個體

如果現在不打算使用，請停止計算執行個體：

在工作室的左側導覽區域中，選取 [計算]。
在頂端索引標籤中，選取 [計算執行個體]
選取清單中的計算執行個體。
在頂端工具列中，選取 [停止]。

刪除所有資源

重要

您所建立的資源可用來作為其他 Azure Machine Learning 教學課程和操作說明文章的先決條件。

如果不打算使用您建立的任何資源，請刪除以免產生任何費用：

在 Azure 入口網站中，選取最左邊的 [資源群組]。
從清單中，選取您所建立的資源群組。
選取 [刪除資源群組]。
輸入資源群組名稱。接著選取刪除。

下一步

了解如何排定機器學習管線作業

共用方式為