使用搭配 Azure 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 筆記本中以互動方式執行程式碼片段。

本文以 Azure Machine Learning 範例存放庫目錄中的 sdk/python/jobs/pipelines/2e_image_classification_keras_minist_convnetimage_classification_keras_minist_convnet.ipynb 筆記本為基礎。

匯入必要的程式庫

匯入本文所需的所有 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

準備管線作業的輸入資料

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

時尚 MNIST 是一個時尚圖像數據集，分為 10 個類。 每個影像都是 28 x 28 灰階影像。 有 60,000 個訓練影像和 10,000 個測試影像。

import urllib3
import shutil
import gzip
import os
from pathlib import Path
from azure.ai.ml import Input

base_url = "https://azureopendatastorage.blob.core.windows.net/mnist/"
base_dir = Path("mnist")
if not base_dir.exists():
    base_dir.mkdir(parents=True)

c = urllib3.PoolManager()
for target_file in [
    "train-images-idx3-ubyte.gz",
    "train-labels-idx1-ubyte.gz",
    "t10k-images-idx3-ubyte.gz",
    "t10k-labels-idx1-ubyte.gz",
]:
    if (base_dir / target_file[:-3]).exists():
        continue
    with c.request("GET", base_url + target_file, preload_content=False) as resp, open(
        base_dir / target_file, "wb"
    ) as out_file:
        shutil.copyfileobj(resp, out_file)
        resp.release_conn()
    with gzip.open(base_dir / target_file, "rb") as f_in, open(
        base_dir / target_file[:-3], "wb"
    ) as f_out:
        shutil.copyfileobj(f_in, f_out)
    os.unlink(base_dir / target_file)

mnist_ds = Input(path=base_dir.as_posix())

您可以藉由定義 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()

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

• name 是元件的唯一識別碼

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

• display_name 是UI元件的易記顯示名稱

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

• environment 使用 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定義兩個輸出。

  • input_data 是輸入資料路徑
  • training_data 和 test_data 分別是訓練資料和測試資料的輸出資料路徑。

• 元件會將資料從 input_data 轉換為 training_data 的訓練資料 .csv 檔案和 test_data 的測試資料 .csv 檔案。

在工作室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 函數定義元件

定義訓練函式之後，您可以使用 @command_component 來將該函式包裝為元件，然後在 Azure Machine Learning SDK v2 中用於 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)

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

定義函數 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 函式一樣。

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

%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")

從工作區載入已登錄的元件

附註

若要從工作區載入已註冊的元件，您必須先設定工作區連線，如 取得工作區的存取權 一節中所述。 下列作業需要此 ml_client 物件。

如果您的工作區中已註冊元件，則可以使用該 ml_client.components.get() 方法直接載入它們。 當您想要重複使用先前由您註冊或由其他團隊成員共用的元件時，此方法非常有用。

# Load a registered component by name and version
registered_component = ml_client.components.get(
    name="my_registered_component", 
    version="1.0.0"
)

# Load the latest version of a registered component
latest_component = ml_client.components.get(
    name="my_registered_component"
)

您可以列出工作區中所有可用的元件，以尋找您需要的元件：

# List all components in the workspace
components = ml_client.components.list()
for component in components:
    print(f"Name: {component.name}, Version: {component.version}")

載入之後，您可以在管線中使用已註冊的元件，就像從本機檔案或 Python 函式載入的元件一樣。

建置管線

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

附註

若要使用 無伺服器計算，請將 新增 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：

• input_data 的 prepare_data_node 使用 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 驗證案例。

如果 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 建置管線的更多範例，請參閱 範例存放庫
• 如需有關使用工作室 UI 來提交和偵錯管線的資訊，請參閱如何利用 Azure Machine Learning 工作室和元件來建立與執行機器學習管線
• 如需使用 Azure Machine Learning CLI 建立元件和管線的相關資訊，請參閱使用 Azure Machine Learning CLI 的元件來建立和執行機器學習管線
• 如需使用批次端點將管線部署至生產環境的相關資訊，請參閱如何使用 批次端點部署管線