使用 Machine Learning SDK v2 的元件來建立和執行機器學習管線

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

在本文中，您將瞭解如何使用 Azure Machine Learning Python SDK v2 來建置 Azure Machine Learning 管線 ，以完成包含三個步驟的影像分類工作：準備數據、定型影像分類模型，以及為模型評分。 機器學習管線會使用速度、可移植性和重複使用來優化工作流程，因此您可以專注於機器學習，而不是基礎結構和自動化。

此範例管線會訓練小型 Keras 卷積神經網路，以分類 Fashion MNIST 數據集中的影像。 管線看起來像這樣：

在本文中，您會完成下列工作：

• 準備管線作業的輸入數據。
• 建立三個元件來準備數據、定型影像，以及為模型評分。
• 從元件建置管線。
• 取得具有計算之工作區的存取權。
• 提交管線作業。
• 檢閱元件和定型神經網路的輸出。
• （選擇性）註冊元件，以在工作區中進一步重複使用和共用。

如尚未擁有 Azure 訂用帳戶，請在開始之前先建立免費帳戶。 立即試用免費或付費版本的 Azure Machine Learning。

必要條件

• Azure Machine Learning 工作區。 如果您沒有資源，請完成 建立資源教學課程。

• 已安裝 Azure Machine Learning Python SDK v2 的 Python 環境。 如需安裝指示，請參閱 用戶入門。 此環境用於定義和控制您的 Azure Machine Learning 資源環境，並且與執行期間用於訓練的環境是分開的。

• 範例存放庫的複本。

  若要執行定型範例，請先複製範例存放庫並移至 sdk 目錄：

  git clone --depth 1 https://github.com/Azure/azureml-examples
  cd azureml-examples/sdk
  

啟動互動式 Python 工作階段

本文使用 Azure Machine Learning Python SDK 來建立及控制 Azure Machine Learning 管線。 本文是以假設您將在 Python REPL 環境或 Jupyter Notebook 中以互動方式執行代碼段所撰寫。

本文是以 image_classification_keras_minist_convnet.ipynb 筆記本為基礎，您可以在 Azure Machine Learning 範例存放庫的目錄中找到sdk/python/jobs/pipelines/2e_image_classification_keras_minist_convnet此筆記本。

匯入必要的程式庫

匯入本文所需的所有 Azure Machine Learning 連結庫：

# import required libraries
from azure.identity import DefaultAzureCredential, InteractiveBrowserCredential

from azure.ai.ml import MLClient
from azure.ai.ml.dsl import pipeline
from azure.ai.ml import load_component

準備管線作業的輸入資料

您必須準備影像分類管線的輸入數據。

Fashion MNIST 是一個時尚影像數據集，分為 10 個類別。 每個影像都是 28 x 28 灰階影像。 有 60,000 個訓練影像和 10,000 個測試影像。 作為影像分類問題，Fashion MNIST 比傳統 MNIST 手寫數位資料庫更具挑戰性。 其以壓縮的二進為形式散發，如同原本的手寫數位資料庫。

您可以藉由定義 Input 來建立資料來源位置的參考。 資料會保留在現有的位置，因此不會產生額外的儲存成本。

建立用於建置管線的元件

影像分類工作可以分成三個步驟：準備數據、定型模型，以及為模型評分。

Azure Machine Learning 元件是獨立的程式代碼片段，可完成機器學習管線中的一個步驟。 在本文中，您會為影像分類工作建立三個元件：

• 準備數據進行訓練和測試。
• 使用定型數據來定型神經網路以進行影像分類。
• 使用測試數據為模型評分。

針對每個元件，您需要完成下列步驟：

1. 準備包含執行邏輯的 Python 腳本。

2. 定義元件的介面。

3. 新增元件的其他元數據，包括執行環境和執行元件的命令。

下一節說明如何以兩種方式建立元件。 針對前兩個元件，您會使用 Python 函式。 針對第三個元件，您可以使用 YAML 定義。

建立數據準備元件

此管線中的第一個元件會將 的 fashion_ds 壓縮數據檔轉換成兩個 .csv 檔案，一個用於定型，另一個用於評分。 您可以使用 Python 函式來定義此元件。

如果您遵循 Azure Machine Learning 範例存放庫中的範例，則資料夾中已有來源檔案可供使用 prep 。 此資料夾包含兩個檔案來建構元件： prep_component.py，此元件會定義元件的 和 conda.yaml，其會定義元件的運行時間環境。

使用 Python 函式定義元件

藉由使用函式 command_component() 作為裝飾器，您可以輕鬆地定義元件的介面、其元數據，以及要從 Python 函式執行的程式碼。 每個裝飾的 Python 函式都會轉換成管線服務可以處理的單一靜態規格 (YAML)。

# Converts MNIST-formatted files at the passed-in input path to training data output path and test data output path
import os
from pathlib import Path
from mldesigner import command_component, Input, Output


@command_component(
    name="prep_data",
    version="1",
    display_name="Prep Data",
    description="Convert data to CSV file, and split to training and test data",
    environment=dict(
        conda_file=Path(__file__).parent / "conda.yaml",
        image="mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04",
    ),
)
def prepare_data_component(
    input_data: Input(type="uri_folder"),
    training_data: Output(type="uri_folder"),
    test_data: Output(type="uri_folder"),
):
    convert(
        os.path.join(input_data, "train-images-idx3-ubyte"),
        os.path.join(input_data, "train-labels-idx1-ubyte"),
        os.path.join(training_data, "mnist_train.csv"),
        60000,
    )
    convert(
        os.path.join(input_data, "t10k-images-idx3-ubyte"),
        os.path.join(input_data, "t10k-labels-idx1-ubyte"),
        os.path.join(test_data, "mnist_test.csv"),
        10000,
    )


def convert(imgf, labelf, outf, n):
    f = open(imgf, "rb")
    l = open(labelf, "rb")
    o = open(outf, "w")

    f.read(16)
    l.read(8)
    images = []

    for i in range(n):
        image = [ord(l.read(1))]
        for j in range(28 * 28):
            image.append(ord(f.read(1)))
        images.append(image)

    for image in images:
        o.write(",".join(str(pix) for pix in image) + "\n")
    f.close()
    o.close()
    l.close()

使用 @command_component 裝飾器，前述程式碼定義了一個顯示名稱為 Prep Data 的元件。

• name 是元件的唯一識別碼。

• version 是元件的目前版本。 元件可以有多個版本。

• display_name 是UI元件的易記顯示名稱。 這不是唯一的。

• description 通常描述元件可以完成的工作。

• environment 指定元件的運行時間環境。 此元件的環境會指定 Docker 映像，並參考 conda.yaml 檔案。

  檔案 conda.yaml 包含元件所使用的所有套件：

  name: imagekeras_prep_conda_env
  channels:
    - defaults
  dependencies:
    - python=3.7.11
    - pip=20.0
    - pip:
      - mldesigner==0.1.0b4
  

• prepare_data_component 函式會針對 input_data 定義一個輸入，以及針對 training_data 和 test_data 定義兩個輸出。 input_data 是輸入數據路徑。 training_data 和 test_data 是訓練資料和測試資料的輸出資料路徑。

• 元件會將來自 input_data 的數據轉換成 training_data .csv，以訓練數據和 test_data .csv 來測試數據。

這是元件在 Studio UI 中的外觀：

• 元件是管線圖形中的區塊。
• input_data、 training_data和 test_data 是元件的埠，會連線到其他元件以進行數據流處理。

您現在已為 Prep Data 元件準備好了所有的原始檔案。

建立模型訓練元件

在本節中，您將建立元件，以在 Python 函式中定型影像分類模型，就像使用 Prep Data 元件一樣。

由於定型邏輯較複雜，因此您會將定型程式代碼放在個別的 Python 檔案中。

此元件的來源檔案位於 trainAzure Machine Learning 範例存放庫的資料夾中。 此資料夾包含三個用來建構元件的檔案：

• train.py 包含定型模型的邏輯。
• train_component.py 會定義元件的介面，並匯入 中的 train.py函式。
• conda.yaml 定義元件的運行時間環境。

取得包含邏輯的腳本

檔案 train.py 包含一般 Python 函式，可執行邏輯來定型影像分類的 Keras 類神經網路。 若要檢視程式碼，請參閱 GitHub 上的 train.py 檔案。

使用 Python 函式定義元件

定義定型函式之後，您可以在 Azure Machine Learning SDK v2 中使用 @command_component，將您的函式包裝為可在 Azure Machine Learning 管線中使用的元件。

import os
from pathlib import Path
from mldesigner import command_component, Input, Output


@command_component(
    name="train_image_classification_keras",
    version="1",
    display_name="Train Image Classification Keras",
    description="train image classification with keras",
    environment=dict(
        conda_file=Path(__file__).parent / "conda.yaml",
        image="mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04",
    ),
)
def keras_train_component(
    input_data: Input(type="uri_folder"),
    output_model: Output(type="uri_folder"),
    epochs=10,
):
    # avoid dependency issue, execution logic is in train() func in train.py file
    from train import train

    train(input_data, output_model, epochs)

上述程式代碼會使用 @command_component定義具有顯示名稱Train Image Classification Keras的元件。

• keras_train_component 函式會定義兩個輸入：一個是來源訓練數據的輸入 input_data，一個是指定訓練期間所用世代數目的輸入 epochs，以及一個輸出 output_model，用來指定模型檔案的輸出路徑。 epochs 的預設值為 10。 此元件的邏輯來自 train() train.py 中的函 式。

訓練模型元件的配置比資料準備元件稍微複雜一點。 conda.yaml 的外觀如下：

name: imagekeras_train_conda_env
channels:
  - defaults
dependencies:
  - python=3.8
  - pip=20.2
  - pip:
    - mldesigner==0.1.0b12
    - azureml-mlflow==1.50.0
    - tensorflow==2.7.0
    - numpy==1.21.4
    - scikit-learn==1.0.1
    - pandas==1.3.4
    - matplotlib==3.2.2
    - protobuf==3.20.0

您現在已備妥 Train Image Classification Keras 元件的所有原始檔案。

建立模型評分元件

在本節中，您會建立元件，以透過 YAML 規格和腳本為定型模型評分。

如果您遵循 Azure Machine Learning 範例存放庫中的範例，則資料夾中已有來源檔案可供使用 score 。 此資料夾包含三個用來建構元件的檔案：

• score.py 包含元件的原始程式碼。
• score.yaml 會定義元件的介面和其他詳細數據。
• conda.yaml 定義元件的運行時間環境。

取得包含邏輯的腳本

檔案 score.py 包含執行定型模型邏輯的一般 Python 函式：

from tensorflow import keras
from keras.models import Sequential
from keras.layers import Dense, Dropout, Flatten
from keras.layers import Conv2D, MaxPooling2D
from tensorflow.keras.layers import BatchNormalization
from tensorflow.keras.utils import to_categorical
from keras.callbacks import Callback
from keras.models import load_model

import argparse
from pathlib import Path
import numpy as np
import pandas as pd
import os
import matplotlib.pyplot as plt
import mlflow


def get_file(f):

    f = Path(f)
    if f.is_file():
        return f
    else:
        files = list(f.iterdir())
        if len(files) == 1:
            return files[0]
        else:
            raise Exception("********This path contains more than one file*******")


def parse_args():
    # setup argparse
    parser = argparse.ArgumentParser()

    # add arguments
    parser.add_argument(
        "--input_data", type=str, help="path containing data for scoring"
    )
    parser.add_argument(
        "--input_model", type=str, default="./", help="input path for model"
    )

    parser.add_argument(
        "--output_result", type=str, default="./", help="output path for model"
    )

    # parse args
    args = parser.parse_args()

    # return args
    return args


def score(input_data, input_model, output_result):

    test_file = get_file(input_data)
    data_test = pd.read_csv(test_file, header=None)

    img_rows, img_cols = 28, 28
    input_shape = (img_rows, img_cols, 1)

    # Read test data
    X_test = np.array(data_test.iloc[:, 1:])
    y_test = to_categorical(np.array(data_test.iloc[:, 0]))
    X_test = (
        X_test.reshape(X_test.shape[0], img_rows, img_cols, 1).astype("float32") / 255
    )

    # Load model
    files = [f for f in os.listdir(input_model) if f.endswith(".h5")]
    model = load_model(input_model + "/" + files[0])

    # Log metrics of the model
    eval = model.evaluate(X_test, y_test, verbose=0)

    mlflow.log_metric("Final test loss", eval[0])
    print("Test loss:", eval[0])

    mlflow.log_metric("Final test accuracy", eval[1])
    print("Test accuracy:", eval[1])

    # Score model using test data
    y_predict = model.predict(X_test)
    y_result = np.argmax(y_predict, axis=1)

    # Output result
    np.savetxt(output_result + "/predict_result.csv", y_result, delimiter=",")


def main(args):
    score(args.input_data, args.input_model, args.output_result)


# run script
if __name__ == "__main__":
    # parse args
    args = parse_args()

    # call main function
    main(args)

中的 score.py 程式代碼會採用三個命令行自變數： input_data、 input_model和 output_result。 程式會使用輸入數據來評分輸入模型，然後輸出結果。

透過 YAML 定義元件

在本節中，您將瞭解如何以有效的 YAML 元件規格格式建立元件規格。 此檔案會指定下列資訊：

• 元數據。 名稱、顯示名稱、版本、類型等等。
• 介面。 輸入和輸出。
• 命令、程式代碼和環境。 用來執行元件的命令、程式代碼和環境。

$schema: https://azuremlschemas.azureedge.net/latest/commandComponent.schema.json
type: command

name: score_image_classification_keras
display_name: Score Image Classification Keras
inputs:
  input_data: 
    type: uri_folder
  input_model:
    type: uri_folder
outputs:
  output_result:
    type: uri_folder
code: ./
command: python score.py --input_data ${{inputs.input_data}} --input_model ${{inputs.input_model}} --output_result ${{outputs.output_result}}
environment:
  conda_file: ./conda.yaml
  image: mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04

• name 是元件的唯一識別碼。 其顯示名稱為 Score Image Classification Keras。
• 此元件有兩個輸入和一個輸出。
• 原始碼路徑定義於 code 區段中。 當元件在雲端中執行時，來自該路徑的所有檔案都會上傳為元件的快照集。
• 區 command 段會指定要在元件執行時執行的命令。
• environment 部分包含 Docker 映像和 conda YAML 檔案。 來源檔案位於範例存放庫中。

您現在擁有模型評分元件的所有來源檔案。

載入元件以建置管線

您可以匯入數據準備元件和模型定型元件，這些元件是由 Python 函式所定義，就像一般 Python 函式一樣。

下列程式代碼會分別從 prep 資料夾中的 prep_component.py 檔案匯入 prepare_data_component() 函式，以及從 train 資料夾中的 train_component 檔案匯入 keras_train_component() 函式。

%load_ext autoreload
%autoreload 2

# load component function from component python file
from prep.prep_component import prepare_data_component
from train.train_component import keras_train_component

# print hint of components
help(prepare_data_component)
help(keras_train_component)

您可以使用 函 load_component() 式來載入 YAML 所定義的分數元件。

# load component function from yaml
keras_score_component = load_component(source="./score/score.yaml")

建置管線

您已建立並載入所有元件和輸入數據，以建置管線。 您現在可以將它們組合成流程：

注意

若要使用 無伺服器計算，請將 新增 from azure.ai.ml.entities import ResourceConfiguration 至檔案頂端。 然後取代：

• default_compute=cpu_compute_target 與 default_compute="serverless"。
• train_node.compute = gpu_compute_target 與 train_node.resources = "ResourceConfiguration(instance_type="Standard_NC6s_v3",instance_count=2)。

# define a pipeline containing 3 nodes: Prepare data node, train node, and score node
@pipeline(
    default_compute=cpu_compute_target,
)
def image_classification_keras_minist_convnet(pipeline_input_data):
    """E2E image classification pipeline with keras using python sdk."""
    prepare_data_node = prepare_data_component(input_data=pipeline_input_data)

    train_node = keras_train_component(
        input_data=prepare_data_node.outputs.training_data
    )
    train_node.compute = gpu_compute_target

    score_node = keras_score_component(
        input_data=prepare_data_node.outputs.test_data,
        input_model=train_node.outputs.output_model,
    )


# create a pipeline
pipeline_job = image_classification_keras_minist_convnet(pipeline_input_data=mnist_ds)

管線具有預設計算 cpu_compute_target。 如果您未指定特定節點的計算，該節點將會在預設計算上執行。

管線具有管線層級輸入 pipeline_input_data。 您在提交管線作業時，可以為管線輸入指定一個值。

管線包含三個節點： prepare_data_node、 train_node和 score_node。

• input_data 的 prepare_data_node 會使用 pipeline_input_data 的值。

• train_node的 input_data 是 training_data 的prepare_data_node輸出。

• input_data 是 prepare_data_node 的 test_data 輸出，而 input_model 是 train_node 的 output_model。

• 因為 train_node 將 CNN 模型定型，因此您可以將其計算指定為 gpu_compute_target。 這樣做可以改善訓練效能。

提交管線作業

現在您已建構管線，您可以將作業提交至工作區。 若要提交作業，您必須先連線到工作區。

取得您工作區的存取權

設定認證

您將使用 DefaultAzureCredential 來取得工作區的存取權。 DefaultAzureCredential 應該能夠處理大部分的 Azure SDK 驗證案例。

如果 DefaultAzureCredential 不適用於您，請參閱 此設定認證範例 和 身分識別套件。

try:
    credential = DefaultAzureCredential()
    # Check if given credential can get token successfully.
    credential.get_token("https://management.azure.com/.default")
except Exception as ex:
    # Fall back to InteractiveBrowserCredential in case DefaultAzureCredential not work
    credential = InteractiveBrowserCredential()

找到配置運算資源的工作區句柄

建立 MLClient 物件來管理 Azure Machine Learning 服務。 如果您使用 無伺服器計算，則不需要建立這些計算。

# Get a handle to workspace
ml_client = MLClient.from_config(credential=credential)

# Retrieve an already attached Azure Machine Learning Compute.
cpu_compute_target = "cpu-cluster"
print(ml_client.compute.get(cpu_compute_target))
gpu_compute_target = "gpu-cluster"
print(ml_client.compute.get(gpu_compute_target))

重要

此代碼段預期工作區組態 JSON 檔案會儲存在目前目錄或其父系中。 如需建立工作區的詳細資訊，請參閱建立工作區資源。 如需將組態儲存至檔案的詳細資訊，請參閱 建立工作區組態檔。

將管線作業提交至工作區

既然您已掌握工作區，您可以提交管線作業：

pipeline_job = ml_client.jobs.create_or_update(
    pipeline_job, experiment_name="pipeline_samples"
)
pipeline_job

上述程式代碼會將此影像分類管線作業提交至名為 pipeline_samples的實驗。 如果實驗不存在，它會自動建立實驗。 pipeline_input_data 使用 fashion_ds。

提交實驗的呼叫會快速完成。 其會產生類似以下的輸出：

實驗	名稱	類型	狀態	詳細資料頁面
pipeline_samples	sharp_pipe_4gvqx6h1fb	管線	準備	Azure Machine Learning 工作室的連結。

透過選擇這個連結，您可以監控管線執行。 或者，您可以執行下列程式代碼來封鎖它，直到它完成為止：

# wait until the job completes
ml_client.jobs.stream(pipeline_job.name)

重要

管線的第一次執行大約需要 15 分鐘。 系統會下載所有相依性、建立 Docker 映像，以及布建和建立 Python 環境。 再次執行管線所需的時間會大幅縮短，因為過程中會重複使用這些資源，而不會再次建立。 不過，管線的總運行時間取決於腳本的工作負載，以及每個管線步驟中執行的進程。

檢查UI中的輸出並偵錯您的管線

您可以選取 Link to Azure Machine Learning studio，這是管線的作業詳細數據頁面。 您會看到管線圖形：

您可以以滑鼠右鍵按下元件來檢查每個元件的記錄和輸出，或選取元件以開啟其詳細資料窗格。 若要深入瞭解如何在UI中偵錯 管線，請參閱使用 Azure Machine Learning Studio 偵錯管線失敗。

（選擇性）向工作區註冊元件

在上一節中，您會使用三個元件來建置管線，以完成影像分類工作。 您也可以將元件註冊到工作區，以便在工作區中共用和重複使用它們。 下列範例示範如何註冊數據準備元件：

try:
    # try get back the component
    prep = ml_client.components.get(name="prep_data", version="1")
except:
    # if not exists, register component using following code
    prep = ml_client.components.create_or_update(prepare_data_component)

# list all components registered in workspace
for c in ml_client.components.list():
    print(c)

您可以使用 ml_client.components.get() 依名稱和版本取得已註冊的元件。 您可以使用 ml_client.components.create_or_update() 來註冊先前從 Python 函式或 YAML 載入的元件。

下一步

• 如需使用機器學習 SDK 來建置管線的更多範例，請參閱範例存放庫。
• 如需使用 Studio UI 來提交和偵錯管線的相關信息，請參閱 使用元件搭配 Azure Machine Learning Studio 建立和執行機器學習管線。
• 如需使用 Azure Machine Learning CLI 來建立元件和管線的相關信息，請參閱 搭配 Azure Machine Learning CLI 使用元件建立和執行機器學習管線。
• 如需使用批次端點將管線部署至生產環境的資訊，請參閱 如何使用批次端點部署管線。