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

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

注意

如需使用 SDK v1 建置管線的教學課程，請參閱教學課程：建置適用於影像分類的 Azure 機器學習 管線

機器學習管線的核心是將完整的機器學習工作分割成多步驟工作流程。 每個步驟都是可管理的元件，可以個別開發、優化、設定及自動化。 步驟會透過定義完善的介面進行連線。 Azure 機器學習 管線服務會自動協調管線步驟之間的所有相依性。 使用管線的優點是標準化 MLOps 實務、可調整的小組共同作業、訓練效率和降低成本。 若要深入瞭解管線的優點，請參閱什麼是 Azure 機器學習 管線。

在本教學課程中，您會使用 Azure 機器學習，使用 Azure 機器學習 Python SDK v2 來建立生產就緒的機器學習專案。

這表示您將能夠利用 Azure 機器學習 Python SDK 來：

• 取得 Azure 機器學習 工作區的句柄
• 建立 Azure 機器學習 數據資產
• 建立可重複使用的 Azure 機器學習元件
• 建立、驗證及執行 Azure 機器學習 管線

在本教學課程中，您會建立 Azure 機器學習 管線來定型模型以進行信用默認預測。 管線會處理兩個步驟：

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

下一個影像顯示簡單的管線，因為您在提交後會在 Azure Studio 中看到該管線。

這兩個步驟是第一個數據準備和第二個定型。

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

必要條件

1. 若要使用 Azure 機器學習，您必須先有工作區。 如果您沒有資源，請完成 建立資源，才能開始 建立工作區，並深入瞭解如何使用它。

2. 登入 Studio，如果工作區尚未開啟，請選取您的工作區。

3. 完成本教學課程 上傳、存取及探索您的數據 ，以建立您在本教學課程中所需的數據資產。 請確定您執行所有程式代碼來建立初始數據資產。 如有需要，請探索數據並加以修改，但您只需要本教學課程中的初始數據。

4. 在您的工作區開啟或建立筆記本：

   • 如果您想要將程式代碼複製/貼到儲存格中，請建立新的筆記本。
   • 或者，從 Studio 的範例區段開啟 tutorials/get-started-notebooks/pipeline.ipynb。 然後選取 [ 複製 ] 以將筆記本新增至您的 檔案。 （請參閱尋找範例的位置。

設定核心

1. 在開啟筆記本上方的頂端列上，如果您還沒有計算實例，請建立計算實例。

   

2. 如果計算實例已停止，請選取 [ 啟動計算 ]，並等候其執行。

   

3. 請確定位於右上方的核心是 Python 3.10 - SDK v2。 如果沒有，請使用下拉式清單來選取此核心。

   

4. 如果您看到橫幅指出您需要進行驗證，請選取 [ 驗證]。

重要

本教學課程的其餘部分包含教學課程筆記本的單元格。 將其複製/貼到新的筆記本中，或如果您複製筆記本，請立即切換至筆記本。

設定管線資源

Azure 機器學習 架構可從 CLI、Python SDK 或 Studio 介面使用。 在此範例中，您會使用 Azure 機器學習 Python SDK v2 來建立管線。

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

• 定型的數據資產
• 要執行管線的軟體環境
• 作業執行所在位置的計算資源

建立工作區的句柄

在深入探討程序代碼之前，您需要參考工作區的方法。 您將為工作區的句柄建立 ml_client 。 接著，您將用來 ml_client 管理資源和作業。

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

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

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 機器學習 中的數據開始。

• Azure 機器學習 會使用 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 機器學習 特定套件 （azureml-mlflow）。

Azure 機器學習 套件並非執行 Azure 機器學習 作業的必要專案。 不過，新增這些套件可讓您與 Azure 機器學習 互動，以記錄計量和註冊模型，全都在 Azure 機器學習 作業內。 在本教學課程稍後的訓練腳本中，您會使用它們。

使用 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 機器學習 管線是可重複使用的 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 機器學習 將數據集掛接為資料夾至計算，因此，我們建立了輔助函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 機器學習 元件。

使用可執行命令行動作的一般用途 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 機器學習 記錄功能來記錄和可視化學習進度。

您使用 類別 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 機器學習 環境。 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:AzureML-sklearn-1.0-ubuntu20.04-py38-cpu:1
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 機器學習 管線的特定@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 中執行。 這次您在 create_or_update 上使用 ml_client.jobs。

在這裡，您也會傳遞實驗名稱。 實驗是某個專案上所有反覆專案的容器。 所有以相同實驗名稱提交的作業，都會在 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，已自動記錄定型計量。

  

將模型部署為線上端點

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

清除資源

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

停止計算實例

如果您現在不會使用它，請停止計算實例：

1. 在 Studio 的左側導覽區域中，選取 [ 計算]。
2. 在頂端索引標籤中，選取 [ 計算實例]
3. 選取清單中的計算實例。
4. 在頂端工具列上，選取 [ 停止]。

刪除所有資源

重要

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

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

1. 在 Azure 入口網站中，選取最左邊的 [資源群組]。

2. 從清單中，選取您所建立的資源群組。

3. 選取 [刪除資源群組]。

   

4. 輸入資源群組名稱。 接著選取刪除。

下一步

瞭解如何 排程機器學習管線作業