教學課程：建置影像分類的 Azure Machine Learning 管線

適用於： Python SDK azureml v1 (部分機器翻譯)

在本教學課程中，您將了解如何建置 Azure Machine Learning 管線來準備資料並訓練機器學習模型。 機器學習管線會以速度、可攜性和重複使用性來將工作流程最佳化，讓您能夠專注於機器學習，而不是基礎結構和自動化。

此範例會訓練小型 Keras 卷積神經網路，以分類 Fashion MNIST 資料集中的影像。

在本教學課程中，您會完成下列工作：

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

必要條件

• 如果您還沒有 Azure Machine Learning 工作區，請完成建立資源以開始使用。
• 已安裝 azureml-core 和 azureml-pipeline 套件的 Python 環境。 此環境用於定義和控制您的 Azure Machine Learning 資源，與執行階段上用於訓練的環境不同。

啟動互動式 Python 工作階段

本教學課程使用適用於 Azure Machine Learning 的 Python SDK 來建立和控制 Azure Machine Learning 管線。 本教學課程假設您將在 Python REPL 環境或 Jupyter 筆記本中以互動方式執行程式碼片段。

• 本教學課程是以 Azure Machine Learning 範例存放庫中 python-sdk/tutorial/using-pipelines 目錄的 image-classification.ipynb 筆記本為基礎。 步驟自身的原始程式碼位於 keras-mnist-fashion 子目錄中。

匯入類型

匯入本教學課程所需的所有 Azure Machine Learning 類型：

import os
import azureml.core
from azureml.core import (
    Workspace,
    Experiment,
    Dataset,
    Datastore,
    ComputeTarget,
    Environment,
    ScriptRunConfig
)
from azureml.data import OutputFileDatasetConfig
from azureml.core.compute import AmlCompute
from azureml.core.compute_target import ComputeTargetException
from azureml.pipeline.steps import PythonScriptStep
from azureml.pipeline.core import Pipeline

# check core SDK version number
print("Azure Machine Learning SDK Version: ", azureml.core.VERSION)

Azure Machine Learning SDK 版本應為 1.37 或更新版本。 如果並非或低於該本版，請透過 pip install --upgrade azureml-core 升級。

設定工作區

從現有的 Azure Machine Learning 工作區建立工作區物件。

workspace = Workspace.from_config()

建立管線的基礎結構

建立 Experiment 物件來保留管線執行的結果：

exp = Experiment(workspace=workspace, name="keras-mnist-fashion")

建立 ComputeTarget，來代表管線執行所在的電腦資源。 本教學課程中使用的簡易神經網路只需幾分鐘即可訓練完畢，即使在以 CPU 為基礎的電腦上也一樣。 如果您想要使用 GPU 進行訓練，請將 use_gpu 設定為 True。 佈建計算目標通常需要約五分鐘時間。

use_gpu = False

# choose a name for your cluster
cluster_name = "gpu-cluster" if use_gpu else "cpu-cluster"

found = False
# Check if this compute target already exists in the workspace.
cts = workspace.compute_targets
if cluster_name in cts and cts[cluster_name].type == "AmlCompute":
    found = True
    print("Found existing compute target.")
    compute_target = cts[cluster_name]
if not found:
    print("Creating a new compute target...")
    compute_config = AmlCompute.provisioning_configuration(
        vm_size= "STANDARD_NC6" if use_gpu else "STANDARD_D2_V2"
        # vm_priority = 'lowpriority', # optional
        max_nodes=4,
    )

    # Create the cluster.
    compute_target = ComputeTarget.create(workspace, cluster_name, compute_config)

    # Can poll for a minimum number of nodes and for a specific timeout.
    # If no min_node_count is provided, it will use the scale settings for the cluster.
    compute_target.wait_for_completion(
        show_output=True, min_node_count=None, timeout_in_minutes=10
    )
# For a more detailed view of current AmlCompute status, use get_status().print(compute_target.get_status().serialize())

建立 Azure 預存資料的資料集

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

若要建立參考 Web 型資料的 Dataset，請執行：

data_urls = ["https://data4mldemo6150520719.blob.core.windows.net/demo/mnist-fashion"]
fashion_ds = Dataset.File.from_files(data_urls)

# list the files referenced by fashion_ds
print(fashion_ds.to_path())

此行程式碼會快速完成。 基礎資料會保留在 data_urls 陣列中指定的 Azure 儲存體資源。

建立資料準備管線步驟

此管線的第一個步驟會將 fashion_ds 的壓縮資料檔案轉換為您工作區中的資料集，以可供訓練使用的 CSV 檔案組成。 在工作區註冊之後，您的共同作業者即可存取此資料，用於自己的分析與訓練等等

datastore = workspace.get_default_datastore()
prepared_fashion_ds = OutputFileDatasetConfig(
    destination=(datastore, "outputdataset/{run-id}")
).register_on_complete(name="prepared_fashion_ds")

上述程式碼會指定以管線步驟輸出為基礎的資料集。 基礎處理的檔案將會放在工作區預設的資料存放區 Blob 儲存體中，位於 destination 中指定的路徑。 資料集將會在工作區中以 prepared_fashion_ds 名稱進行註冊。

建立管線步驟的來源

您截至目前為止執行的程式碼已建立並控制 Azure 資源。 現在是時候撰寫在網域中執行第一個步驟的程式碼。

如果您遵循 Azure Machine Learning 範例存放庫中的範例，便可以 keras-mnist-fashion/prepare.py 形式取得來源檔案。

如果您是從頭開始，請建立名為 keras-mnist-fashion/ 的子目錄。 建立新檔案、將下列程式碼新增至檔案，並將檔案命名為 prepare.py。

# prepare.py
# Converts MNIST-formatted files at the passed-in input path to a passed-in output path
import os
import sys

# Conversion routine for MNIST binary format
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()

# The MNIST-formatted source
mounted_input_path = sys.argv[1]
# The output directory at which the outputs will be written
mounted_output_path = sys.argv[2]

# Create the output directory
os.makedirs(mounted_output_path, exist_ok=True)

# Convert the training data
convert(
    os.path.join(mounted_input_path, "mnist-fashion/train-images-idx3-ubyte"),
    os.path.join(mounted_input_path, "mnist-fashion/train-labels-idx1-ubyte"),
    os.path.join(mounted_output_path, "mnist_train.csv"),
    60000,
)

# Convert the test data
convert(
    os.path.join(mounted_input_path, "mnist-fashion/t10k-images-idx3-ubyte"),
    os.path.join(mounted_input_path, "mnist-fashion/t10k-labels-idx1-ubyte"),
    os.path.join(mounted_output_path, "mnist_test.csv"),
    10000,
)

prepare.py 中的程式碼會接收兩個命令列引數：第一個會指派給 mounted_input_path，而第二個引數則會指派給 mounted_output_path。 如果該子目錄不存在，則呼叫 os.makedirs 建立子目錄。 然後，程式會轉換訓練與測試資料，並將以逗號分隔的檔案輸出至 mounted_output_path。

指定管線步驟

返回您用於指定管線的 Python 環境，執行此程式碼來為準備程式碼建立 PythonScriptStep：

script_folder = "./keras-mnist-fashion"

prep_step = PythonScriptStep(
    name="prepare step",
    script_name="prepare.py",
    # On the compute target, mount fashion_ds dataset as input, prepared_fashion_ds as output
    arguments=[fashion_ds.as_named_input("fashion_ds").as_mount(), prepared_fashion_ds],
    source_directory=script_folder,
    compute_target=compute_target,
    allow_reuse=True,
)

當管線步驟執行時，PythonScriptStep 的呼叫會指定下列動作：

• script_folder 目錄中的所有檔案都會上傳至 compute_target
• 在上傳的來源檔案中將執行 prepare.py 檔案
• fashion_ds 和 prepared_fashion_ds 資料集將掛接於 compute_target，並顯示為目錄
• fashion_ds 檔案的路徑為 prepare.py 的第一個引數。 在 prepare.py 中，此引數會指派給 mounted_input_path
• prepared_fashion_ds 的路徑將是 prepare.py 的第二個引數。 在 prepare.py 中，此引數會指派給 mounted_output_path
• 由於 allow_reuse 是 True，因此來源檔案或輸入變更後，此項目才會重新執行
• 此 PythonScriptStep 會命名為 prepare step

模組化和重複使用為管線的重要優點。 Azure Machine Learning 可以自動判斷原始程式碼或資料集變更。 如果 allow_reuse 為 True，則不受影響的步驟輸出將會重複使用，無需再次重新執行該步驟。 如果步驟依賴可能發生變化且位於 Azure Machine Learning 外部的資料來源 (例如包含銷售資料的 URL)，請將 allow_reuse 設為 False，而管線步驟則會在每次管線執行時一併執行。

建立訓練步驟

資料從壓縮格式轉換為 CSV 檔案後即可用於訓練卷積神經網路。

建立訓練步驟的來源

使用較大的管線時，最好將每個步驟的原始程式碼放在個別的目錄中 (例如 src/prepare/、src/train/ 等等)，但在本教學課程中，只會在相同 keras-mnist-fashion/ 來源目錄中使用或建立 train.py 檔案。

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

import numpy as np
import pandas as pd
import os
import matplotlib.pyplot as plt
from sklearn.model_selection import train_test_split
from azureml.core import Run

# dataset object from the run
run = Run.get_context()
dataset = run.input_datasets["prepared_fashion_ds"]

# split dataset into train and test set
(train_dataset, test_dataset) = dataset.random_split(percentage=0.8, seed=111)

# load dataset into pandas dataframe
data_train = train_dataset.to_pandas_dataframe()
data_test = test_dataset.to_pandas_dataframe()

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

X = np.array(data_train.iloc[:, 1:])
y = to_categorical(np.array(data_train.iloc[:, 0]))

# here we split validation data to optimiza classifier during training
X_train, X_val, y_train, y_val = train_test_split(X, y, test_size=0.2, random_state=13)

# test data
X_test = np.array(data_test.iloc[:, 1:])
y_test = to_categorical(np.array(data_test.iloc[:, 0]))


X_train = (
    X_train.reshape(X_train.shape[0], img_rows, img_cols, 1).astype("float32") / 255
)
X_test = X_test.reshape(X_test.shape[0], img_rows, img_cols, 1).astype("float32") / 255
X_val = X_val.reshape(X_val.shape[0], img_rows, img_cols, 1).astype("float32") / 255

batch_size = 256
num_classes = 10
epochs = 10

# construct neuron network
model = Sequential()
model.add(
    Conv2D(
        32,
        kernel_size=(3, 3),
        activation="relu",
        kernel_initializer="he_normal",
        input_shape=input_shape,
    )
)
model.add(MaxPooling2D((2, 2)))
model.add(Dropout(0.25))
model.add(Conv2D(64, (3, 3), activation="relu"))
model.add(MaxPooling2D(pool_size=(2, 2)))
model.add(Dropout(0.25))
model.add(Conv2D(128, (3, 3), activation="relu"))
model.add(Dropout(0.4))
model.add(Flatten())
model.add(Dense(128, activation="relu"))
model.add(Dropout(0.3))
model.add(Dense(num_classes, activation="softmax"))

model.compile(
    loss=keras.losses.categorical_crossentropy,
    optimizer=keras.optimizers.Adam(),
    metrics=["accuracy"],
)

# start an Azure ML run
run = Run.get_context()


class LogRunMetrics(Callback):
    # callback at the end of every epoch
    def on_epoch_end(self, epoch, log):
        # log a value repeated which creates a list
        run.log("Loss", log["loss"])
        run.log("Accuracy", log["accuracy"])


history = model.fit(
    X_train,
    y_train,
    batch_size=batch_size,
    epochs=epochs,
    verbose=1,
    validation_data=(X_val, y_val),
    callbacks=[LogRunMetrics()],
)

score = model.evaluate(X_test, y_test, verbose=0)

# log a single value
run.log("Final test loss", score[0])
print("Test loss:", score[0])

run.log("Final test accuracy", score[1])
print("Test accuracy:", score[1])

plt.figure(figsize=(6, 3))
plt.title("Fashion MNIST with Keras ({} epochs)".format(epochs), fontsize=14)
plt.plot(history.history["accuracy"], "b-", label="Accuracy", lw=4, alpha=0.5)
plt.plot(history.history["loss"], "r--", label="Loss", lw=4, alpha=0.5)
plt.legend(fontsize=12)
plt.grid(True)

# log an image
run.log_image("Loss v.s. Accuracy", plot=plt)

# create a ./outputs/model folder in the compute target
# files saved in the "./outputs" folder are automatically uploaded into run history
os.makedirs("./outputs/model", exist_ok=True)

# serialize NN architecture to JSON
model_json = model.to_json()
# save model JSON
with open("./outputs/model/model.json", "w") as f:
    f.write(model_json)
# save model weights
model.save_weights("./outputs/model/model.h5")
print("model saved in ./outputs/model folder")

ML 開發人員應該熟悉此程式碼的絕大部分：

• 資料會分割為訓練和用於訓練的驗證集，以及最終用於評分的單獨測試子集
• 輸入圖形尺寸為 28x28x1 (1 是因為輸入的項目為灰階)，一個批次中會有 256 個輸入項目，共 10 個類別
• 訓練 Epoch 的數量為 10
• 此模型有三個卷積層，均具備最大池化 (max pooling) 以及隨機關閉神經元 (dropout) 功能，後方接著有密集層 (dense layer) 和 softmax 頭層 (softmax head)
• 模型適用大小為 10 個 Epoch，接著會進行評估
• 模型架構會寫入 outputs/model/model.json，並將權數寫入 outputs/model/model.h5

不過，有些程式碼專屬於 Azure Machine Learning。 run = Run.get_context() 會擷取 Run 物件，其中包含目前的服務內容。 train.py 來源會使用此 run 物件來透過其名稱截取輸入資料集 (此為在 prepare.py 中程式碼透過指令碼引數的 argv 陣列截取資料集的替代方案)。

run 物件也可用於記錄每次 Epoch 結束時以及訓練結束時的訓練進度，來記錄隨時間變化的遺失和準確度圖表。

建立訓練管線步驟

設定訓練步驟稍微比準備步驟複雜。 準備步驟僅使用標準 Python 程式庫。 在訓練步驟中，您將更常需要修改執行原始程式碼的執行階段環境。

建立含有下列內容的檔案 conda_dependencies.yml：

dependencies:
- python=3.7
- pip:
  - azureml-core
  - azureml-dataset-runtime
  - keras==2.4.3
  - tensorflow==2.4.3
  - numpy
  - scikit-learn
  - pandas
  - matplotlib

Environment 類別代表機器學習作業執行的執行階段環境。 透過下方程式碼將上述的規格與訓練程式碼建立關聯：

keras_env = Environment.from_conda_specification(
    name="keras-env", file_path="./conda_dependencies.yml"
)

train_cfg = ScriptRunConfig(
    source_directory=script_folder,
    script="train.py",
    compute_target=compute_target,
    environment=keras_env,
)

建立訓練步驟本身會使用類似用於建立準備步驟的程式碼：

train_step = PythonScriptStep(
    name="train step",
    arguments=[
        prepared_fashion_ds.read_delimited_files().as_input(name="prepared_fashion_ds")
    ],
    source_directory=train_cfg.source_directory,
    script_name=train_cfg.script,
    runconfig=train_cfg.run_config,
)

建立並執行管線

現在您已指定資料輸入和輸出並建立管線的步驟，即可將上述步驟組成管線並加以執行：

pipeline = Pipeline(workspace, steps=[prep_step, train_step])
run = exp.submit(pipeline)

建立的 Pipeline 物件將在您的 workspace 中執行，並會由您所指定的準備與訓練步驟組成。

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

Submitted PipelineRun 5968530a-abcd-1234-9cc1-46168951b5eb
Link to Azure Machine Learning Portal: https://ml.azure.com/runs/abc-xyz...

您可以開啟連結來監控管線執行，或也可以執行下列程式碼來進行封鎖，直到管線完成為止：

run.wait_for_completion(show_output=True)

管線完成後，您可以擷取在訓練步驟中記錄的計量：

run.find_step_run("train step")[0].get_metrics()

如果您對計量值感到滿意，即可在工作區中註冊該模型：

run.find_step_run("train step")[0].register_model(
    model_name="keras-model",
    model_path="outputs/model/",
    datasets=[("train test data", fashion_ds)],
)

清除資源

如果您打算執行其他 Azure Machine Learning 教學課程，請不要完成本節。

停止計算執行個體

若您已使用計算執行個體，在您不使用時請停止該 VM 以降低成本。

1. 在您的工作區中，選取 [計算]。

2. 從清單中選取計算執行個體的名稱。

3. 選取 [停止]。

4. 當您準備好再次使用伺服器時，請選取 [啟動]。

刪除所有內容

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

1. 在 Azure 入口網站中，選取左側功能表中的 [資源群組]。
2. 在資源群組清單中，選取您所建立的資源群組。
3. 選取 [刪除資源群組]。
4. 輸入資源群組名稱。 然後，選取 [刪除]。

您也可以保留資源群組，但刪除單一工作區。 顯示工作區屬性，然後選取 [刪除]。

下一步

在本教學課程中，您已使用下列類型：

• Workspace 代表您的 Azure Machine Learning 工作區。 此工作區包含：
  • 包含管道訓練執行結果的 Experiment
  • 延遲載入保留在 Fashion-MNIST 資料存放區中資料的 Dataset
  • 代表執行管線步驟機器的 ComputeTarget
  • Environment，即執行管線步驟的執行階段環境
  • 將 Pipeline 步驟組成一體的 PythonScriptStep
  • 對於訓練流程滿意後您所註冊的 Model

Workspace 物件會包含在本教學課程中未使用資源 (筆記本、端點等項目) 的參考。 如需詳細資訊，請參閱什麼是 Azure Machine Learning 工作區？。

OutputFileDatasetConfig 會將執行個輸出升級為檔案型資料集。 如需如何將資料集搭配資料使用的詳細資訊，請參閱如何存取資料。

如需計算目標和環境的詳細資訊，請參閱 Azure Machine Learning 中的計算目標是什麼？以及 Azure Machine Learning 環境是什麼？

ScriptRunConfig 會將 ComputeTarget 和 Environment 與 Python 來源檔案產生關聯。 PythonScriptStep 會採用 ScriptRunConfig 並定義其輸入與輸出，在此管線中即為 OutputFileDatasetConfig 所建立的檔案資料集。

如需使用機器學習 SDK 來建置管線的更多範例，請參閱範例存放庫。