使用元件搭配 Azure Machine Learning SDK v2 建立和執行機器學習管線

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

在本文中，您將了解如何使用 Python SDK 第 2 版建置 Azure Machine Learning 管線，以完成包含三個步驟的影像分類工作：準備資料、訓練影像分類模型，以及為模型評分。 機器學習管線會以速度、可攜性和重複使用性來將工作流程最佳化，讓您能夠專注於機器學習，而不是基礎結構和自動化。

此範例會訓練小型 Keras 卷積神經網路，以分類 Fashion MNIST 資料集中的影像。 管線看起來應如下所示。

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

• 準備管線作業的輸入資料
• 建立三個元件以準備資料、訓練和評分
• 從元件撰寫管線
• 使用計算取得工作區的存取權
• 提交管線作業
• 檢閱元件的輸出和經過訓練的神經網路
• (選擇性) 註冊元件，以便在工作區內進一步重複使用和共用

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

必要條件

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

• 您已在其中安裝 Azure Machine Learning Python SDK 第 2 版的 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 筆記本中以互動方式執行程式碼片段。

本文以 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 個類別的時尚影像資料集。 每個影像都是 28x28 灰階影像，而且有 60,000 個訓練影像和 10,000 個測試影像。 作為影像分類問題，Fashion-MNIST 比傳統 MNIST 手寫數位資料庫更困難。 其以壓縮的二進為形式散發，如同原本的手寫數位資料庫。

匯入您需要的所有 Azure Machine Learning 必要程式庫。

您可以藉由定義 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 檔案中。

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

• train.py：包含要訓練模型的實際邏輯。
• train_component.py：定義元件的介面，並在 train.py 中匯入函式。
• conda.yaml：定義元件的執行階段環境。

取得包含執行邏輯的指令碼

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

使用 Python 函式定義元件

成功定義訓練函式之後，您可以在 Azure Machine Learning SDK 第 2 版中使用 @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 輸入 (訓練資料的來源)、一個在訓練期間指定 Epoch 的 epochs 輸入，以及一個輸出模型檔案的 output_model 輸出。 epochs 的預設值為 10。 此元件的執行邏輯來自上述 train.py 的 train() 函式。

比起準備資料元件，訓練模型元件的設定稍微複雜一些。 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 元件規格格式建立元件規格。 此檔案會指定下列資訊：

• 中繼資料：name、display_name、version、type 等等。
• 介面：輸入和輸出
• 命令、程式碼和環境：用來執行元件的命令、程式碼和環境

$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 檔案和 train 資料夾下的 train_component 檔案匯入 prepare_data_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)

針對 yaml 所定義的評分元件，您可以使用 load_component() 函式來載入。

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

建置管線

現在您已建立並載入用於建置管線的所有元件和輸入資料。 您可以將這些資料撰寫至管線中：

注意

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

• 包含 default_compute="serverless", 的 default_compute=cpu_compute_target,
• 包含 train_node.resources = "ResourceConfiguration(instance_type="Standard_NC6s_v3",instance_count=2) 的 train_node.compute = gpu_compute_target

# 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。

• prepare_data_node 的 input_data 會使用 pipeline_input_data 的值。

• train_node 的 input_data 來自 prepare_data_node 的 training_data 輸出。

• score_node 的 input_data 來自 prepare_data_node 的 test_data 輸出，而 input_model 來自 train_node 的 output_model。

• 由於 train_node 會訓練 CNN 模型，因此您可以將計算指定為 gpu_compute_target，進而改善訓練效能。

提交管線作業

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

取得您工作區的存取權

設定認證

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

如果這不適用於您，請參考更多可用的認證：設定認證範例、azure-identity 參考文件。

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_job 會產生類似下列的輸出：

呼叫 submit 會快速完成 Experiment ，並產生類似下列的輸出：

Experiment	名稱	類型	狀態	詳細資料頁面
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 中偵錯管線，請參閱如何使用偵錯管線失敗。

(選擇性) 將元件註冊至工作區

在上一節中，您已使用三個元件建置管線，透過 E2E 的方式完成影像分類工作。 您也可以將元件註冊到您的工作區，以便在工作區內共用和重複使用這些元件。 以下是註冊準備資料元件的範例。

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 來提交和偵錯管線的說明，請參閱如何在 UI 中使用元件建立管線。
• 如需如何使用 Azure Machine Learning CLI 來建立元件和管線的說明，請參閱如何透過 CLI 使用元件建立管線。
• 如需如何使用 Batch 端點將管線部署到生產環境，請參閱如何使用批次端點部署管線。