Platform For AI:PyTorchモデルのトレーニングと展開

SDKのインストールと設定

次のコマンドを実行して、Python用PAI SDKをインストールします。

python -m pip install "alipai>=0.4.0"

CLIターミナルで次のコマンドを実行し、設定を実行します。 詳細については、「PAI Python SDKのインストールと設定」をご参照ください。

python -m pai.toolkit.config

トレーニングデータの準備

この例では、MNISTデータセットを使用して、画像分類モデルをトレーニングします。 PAIでトレーニングジョブを実行するには、データをOSSバケットにアップロードする必要があります。

• MNISTデータセットのダウンロード

次のShellスクリプトを実行して、MNISTデータセットをdataという名前のオンプレミスディレクトリにダウンロードします。

#!/bin/sh
set -e

url_prefix="https://ossci-datasets.s3.amazonaws.com/mnist/"
# You can use the following address if the download takes too long.
# url_prefix="http://yann.lecun.com/exdb/mnist/"

mkdir -p data/MNIST/raw/

wget -nv ${url_prefix}train-images-idx3-ubyte.gz -P data/MNIST/raw/
wget -nv ${url_prefix}train-labels-idx1-ubyte.gz -P data/MNIST/raw/
wget -nv ${url_prefix}t10k-images-idx3-ubyte.gz -P data/MNIST/raw/
wget -nv ${url_prefix}t10k-labels-idx1-ubyte.gz -P data/MNIST/raw/

• データセットをOSSにアップロード

OSSが提供するCLIツールossutilを使用して、データセットをアップロードできます。 詳細については、「ossutil overview」をご参照ください。 PAI Python SDKを使用して、トレーニングデータをOSSバケットの /mnist/data/ パスにアップロードすることもできます。

• ossutilを使用:

ossutil cp -rf。/data oss://<YourOssBucket>/mnist/data /

• PythonでPAI SDKを使用する:

from pai.common.oss_utils import upload
from pai.session import get_default_session

sess = get_default_session()
data_uri = upload("./data/", oss_path="mnist/data/", bucket=sess.oss_bucket)
print(data_uri)

トレーニングスクリプトを準備する

ジョブを送信する前に、PyTorchを使用してトレーニングスクリプトを作成する必要があります。 この例で使用されるトレーニングスクリプトは、PyTorchによって提供されるMNISTの例に基づいて変更されます。 変更には、データの読み込みとモデルの保存のロジックの変更が含まれます。 詳細については、「MNISTの例」をご参照ください。

• 環境変数を使用した入力データパスの取得

estimator.fit(input={"train_data":data_uri}) を使用して、OSSに保存されているデータをトレーニングコンテナにマウントします。 トレーニングスクリプトは、ローカルファイルを読み取ることにより、マウントされたデータを取得できます。

estimator.fitメソッドの入力はDICTタイプです。 各入力はチャネルであり、キーはチャネルの名前であり、値は格納されたデータのパスです。 トレーニングスクリプトは、PAI_INPUT_{ChannelNameUpperCase} 変数によって、作業コンテナ内のマウントされたデータのパスを取得できます。

次の内容に基づいてデータ読み込みのコードを変更します。

- dataset1 = datasets.MNIST("../data", train=True, download=True, transform=transform)
- dataset2 = datasets.MNIST("../data", train=False, transform=transform)

+ # Obtain the input data path by using environment variables.
+ data_path = os.environ.get("PAI_INPUT_TRAIN_DATA", "../data")
+ dataset1 = datasets.MNIST(data_path, train=True, download=True, transform=transform)
+ dataset2 = datasets.MNIST(data_path, train=False, transform=transform)

• 環境変数を使用した出力モデルパスの取得

トレーニング環境の指定されたパスにモデルを保存する必要があります。 パス内のデータとモデルがOSSバケットに保存されます。 PAI_OUTPUT_MODEL変数で指定されたパスにモデルを保存する必要があります。 デフォルトでは /ml/output/modelが使用されます。

次の内容に基づいてモデル出力のコードを変更します。

- if args.save_model:
-     torch.save(model.state_dict(), "mnist_cnn.pt")
+ # Save the model.
+ save_model(model)
+ 
+ def save_model(model):
+     """Convert the model to TorchScript and save it to the specified path."""
+     output_model_path = os.environ.get("PAI_OUTPUT_MODEL")
+     os.makedirs(output_model_path, exist_ok=True)
+     
+     m = torch.jit.script(model)
+     m.save(os.path.join(output_model_path, "mnist_cnn.pt"))

PAIが提供する組み込みのPyTorchプロセッサを使用してサービスを作成する場合、入力モデルはTorchScript形式である必要があります。 詳細は、「TorchScript」をご参照ください。 この例では、モデルはTorchScript形式でエクスポートされます。

サンプルトレーニングスクリプト:

# source: https://github.com/pytorch/examples/blob/main/mnist/main.py
from __future__ import print_function

import argparse
import os

import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torch.optim.lr_scheduler import StepLR
from torchvision import datasets, transforms


class Net(nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        self.conv1 = nn.Conv2d(1, 32, 3, 1)
        self.conv2 = nn.Conv2d(32, 64, 3, 1)
        self.dropout1 = nn.Dropout(0.25)
        self.dropout2 = nn.Dropout(0.5)
        self.fc1 = nn.Linear(9216, 128)
        self.fc2 = nn.Linear(128, 10)

    def forward(self, x):
        x = self.conv1(x)
        x = F.relu(x)
        x = self.conv2(x)
        x = F.relu(x)
        x = F.max_pool2d(x, 2)
        x = self.dropout1(x)
        x = torch.flatten(x, 1)
        x = self.fc1(x)
        x = F.relu(x)
        x = self.dropout2(x)
        x = self.fc2(x)
        output = F.log_softmax(x, dim=1)
        return output


def train(args, model, device, train_loader, optimizer, epoch):
    model.train()
    for batch_idx, (data, target) in enumerate(train_loader):
        data, target = data.to(device), target.to(device)
        optimizer.zero_grad()
        output = model(data)
        loss = F.nll_loss(output, target)
        loss.backward()
        optimizer.step()
        if batch_idx % args.log_interval == 0:
            print(
                "Train Epoch: {} [{}/{} ({:.0f}%)]\tLoss: {:.6f}".format(
                    epoch,
                    batch_idx * len(data),
                    len(train_loader.dataset),
                    100.0 * batch_idx / len(train_loader),
                    loss.item(),
                )
            )
            if args.dry_run:
                break


def test(model, device, test_loader):
    model.eval()
    test_loss = 0
    correct = 0
    with torch.no_grad():
        for data, target in test_loader:
            data, target = data.to(device), target.to(device)
            output = model(data)
            test_loss += F.nll_loss(
                output, target, reduction="sum"
            ).item()  # sum up batch loss
            pred = output.argmax(
                dim=1, keepdim=True
            )  # get the index of the max log-probability
            correct += pred.eq(target.view_as(pred)).sum().item()

    test_loss /= len(test_loader.dataset)

    print(
        "\nTest set: Average loss: {:.4f}, Accuracy: {}/{} ({:.0f}%)\n".format(
            test_loss,
            correct,
            len(test_loader.dataset),
            100.0 * correct / len(test_loader.dataset),
        )
    )


def main():
    # Training settings
    parser = argparse.ArgumentParser(description="PyTorch MNIST Example")
    parser.add_argument(
        "--batch-size",
        type=int,
        default=64,
        metavar="N",
        help="input batch size for training (default: 64)",
    )
    parser.add_argument(
        "--test-batch-size",
        type=int,
        default=1000,
        metavar="N",
        help="input batch size for testing (default: 1000)",
    )
    parser.add_argument(
        "--epochs",
        type=int,
        default=14,
        metavar="N",
        help="number of epochs to train (default: 14)",
    )
    parser.add_argument(
        "--lr",
        type=float,
        default=1.0,
        metavar="LR",
        help="learning rate (default: 1.0)",
    )
    parser.add_argument(
        "--gamma",
        type=float,
        default=0.7,
        metavar="M",
        help="Learning rate step gamma (default: 0.7)",
    )
    parser.add_argument(
        "--no-cuda", action="store_true", default=False, help="disables CUDA training"
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        default=False,
        help="quickly check a single pass",
    )
    parser.add_argument(
        "--seed", type=int, default=1, metavar="S", help="random seed (default: 1)"
    )
    parser.add_argument(
        "--log-interval",
        type=int,
        default=10,
        metavar="N",
        help="how many batches to wait before logging training status",
    )
    parser.add_argument(
        "--save-model",
        action="store_true",
        default=False,
        help="For Saving the current Model",
    )
    args = parser.parse_args()
    use_cuda = not args.no_cuda and torch.cuda.is_available()

    torch.manual_seed(args.seed)

    device = torch.device("cuda" if use_cuda else "cpu")

    train_kwargs = {"batch_size": args.batch_size}
    test_kwargs = {"batch_size": args.test_batch_size}
    if use_cuda:
        cuda_kwargs = {"num_workers": 1, "pin_memory": True, "shuffle": True}
        train_kwargs.update(cuda_kwargs)
        test_kwargs.update(cuda_kwargs)

    transform = transforms.Compose(
        [transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,))]
    )

    data_path = os.environ.get("PAI_INPUT_DATA")
    dataset1 = datasets.MNIST(data_path, train=True, download=True, transform=transform)
    dataset2 = datasets.MNIST(data_path, train=False, transform=transform)
    train_loader = torch.utils.data.DataLoader(dataset1, **train_kwargs)
    test_loader = torch.utils.data.DataLoader(dataset2, **test_kwargs)

    model = Net().to(device)
    optimizer = optim.Adadelta(model.parameters(), lr=args.lr)

    scheduler = StepLR(optimizer, step_size=1, gamma=args.gamma)
    for epoch in range(1, args.epochs + 1):
        train(args, model, device, train_loader, optimizer, epoch)
        test(model, device, test_loader)
        scheduler.step()

    # Save the model.
    save_model(model)


def save_model(model):
    """Convert the model to TorchScript and save it to the specified path."""
    output_model_path = os.environ.get("PAI_OUTPUT_MODEL")
    os.makedirs(output_model_path, exist_ok=True)

    m = torch.jit.script(model)
    m.save(os.path.join(output_model_path, "mnist_cnn.pt"))


if __name__ == "__main__":
    main()

上記のトレーニングコードをオンプレミスディレクトリに保存し、[推定値] を使用してPAIに送信します。 この例では、train_srcという名前の新しいディレクトリが作成され、トレーニングスクリプトがtrain_src/train.pyに保存されます。

|-- train_src # The directory of the training script to be uploaded.
    |-- requirements.txt # Optional. The third-party dependencies of the training job.
    '-- train.py # The saved training script 

トレーニングジョブを送信する

Estimatorでは、オンプレミスのトレーニングスクリプトとイメージを使用して、PAIでトレーニングジョブを実行できます。

• トレーニングジョブのスクリプトとコマンド

source_dirパラメーターで指定されたトレーニングスクリプトのディレクトリは、OSSにアップロードされ、ジョブが開始される前にジョブコンテナに準備されます。 デフォルトのディレクトリ /ml/usercodeが使用されます。 コマンドパラメーターで指定される起動コマンドの作業ディレクトリも /ml/usercodeです。

• トレーニングジョブのイメージ

この例では、PAIが提供するPyTorchイメージを使用してトレーニングジョブを実行します。

• トレーニングジョブのハイパーパラメータ

${PAI_CONFIG_DIR}/hyperparameters.jsonファイルを読み取るか、環境変数を使用して、トレーニングジョブのハイパーパラメーターを取得できます。 詳細については、「トレーニングジョブの送信」の「トレーニングジョブのプリセット環境変数」をご参照ください。

この例では、実行されるコマンドはpython train.py $PAI_USER_ARGSです。ここで、PAI_USER_ARGS変数はハイパーパラメーターによって取得された文字列です。 トレーニングジョブの最後の起動コマンドは、python train.py epochs 5 -- batch-size 256 -- lr 0.5です。

• metric_definitionsを使用したトレーニングメトリックの指定

PAIを使用すると、標準出力と標準エラーを含む出力トレーニングログから正規表現を照合することで、トレーニングメトリックを取得できます。 システムは、詳細ページへのリンクも印刷します。 詳細ページで、トレーニングジョブの詳細な設定、出力ログ、およびメトリクスを表示できます。

• instance_typeを使用して、トレーニングのインスタンスタイプを指定します。

PAIでサポートされているインスタンスタイプの詳細については、「Billing of DLC」の「付録: パブリックリソースグループの料金の詳細」セクションをご参照ください。

Estimatorを構築するためのサンプルコード:

pai.estimator import Estimatorからの

from pai.estimator import Estimator
from pai.image import retrieve

# Use the PyTorch 1.18 image for GPU-based trainings to run the training script.
image_uri = retrieve(
    "PyTorch", framework_version="1.8PAI", accelerator_type="GPU"
).image_uri
print(image_uri)

est = Estimator(
    # Startup command of the training job. The default working directory is /ml/usercode/.
    command="python train.py $PAI_USER_ARGS",
    # The relative path or absolute path of the training code directory to be uploaded.
  	# By default, the /ml/usercode directory of the training environment is used.
    source_dir="./train_src/",
    # The image of the training job.
    image_uri=image_uri,
    # Instance configuration.
    instance_type="ecs.gn6i-c4g1.xlarge",  # 4vCPU 15GB 1*NVIDIA T4
    # Hyperparameters of the training job.
    hyperparameters={
        "epochs": 5,
        "batch-size": 64 * 4,
        "lr": 0.5,
    },
    # Metrics configuration of the training job.
    metric_definitions=[
        {
            "Name": "loss",
            "Regex": r".*loss=([-+]?[0-9]*.?[0-9]+(?:[eE][-+]?[0-9]+)?).*",
        },
    ],
    base_job_name="pytorch_mnist",
)

est.fitメソッドを使用して、トレーニングジョブを実行のためにPAIに送信します。 ジョブが送信された後、SDKはジョブの詳細ページのリンクを印刷し、ジョブの実行が終了するまでトレーニングジョブのログを印刷し続けます。

estimator.fitメソッドのinputパラメーターを使用して、OSSに保存されているデータを使用できます。 inputで指定されたパスは、指定されたディレクトリにマウントされます。 次に、トレーニングスクリプトは、ファイルを読み取ることによってデータをロードすることができる。

この例では、トレーニングデータがOSSにアップロードされます。

# If you use ossutil to upload training data, you need to explicitly specify the OSS URI of the input data.
# data_uri = "oss://<YourOssBucket>/mnist/data/"

# Submit the training job.
est.fit(
    inputs={
        "train_data": data_uri,
    }
)

# The output path of the trained model.
print("TrainingJob output model data:")
print(est.model_data())

トレーニングジョブの送信の詳細については、「トレーニングジョブの送信」をご参照ください。

推論サービスのデプロイ

トレーニングジョブが完了したら、estimator.mo del_data() メソッドを使用して、トレーニング済みモデルのOSSパスを取得できます。 次のセクションでは、トレーニング済みモデルをPAIにデプロイしてオンライン推論サービスを作成する方法について説明します。

以下の手順を実行します。

• モデルを使用して推論サービスを構築する方法を説明するには、InferenceSpecを使用します。

プロセッサまたはカスタムイメージを使用して、モデルを配置できます。 両方の方法を以下の例で説明する。

• Model.de ployメソッドを使用して、サービスが使用するリソースやサービス名などの情報を設定し、推論サービスを作成します。

詳細については、「推論サービスのデプロイ」をご参照ください。

プロセッサを使用してモデルをデプロイする

プロセッサは、推論パッケージの抽象記述を含む。 モデルをロードし、モデル推論サービスを開始するために使用されます。 モデル推論サービスは、ユーザーが呼び出すためのAPIを提供します。 PAIは組み込みのPyTorchプロセッサを提供し、モデルをTorchScript形式でPAIにデプロイし、推論サービスを作成できます。 詳細については、「Pytorch」をご参照ください。

次の例では、PyTorchプロセッサを使用して、トレーニング済みモデルを推論サービスとして展開します。

from pai.model import Model, InferenceSpec
from pai.predictor import Predictor
from pai.common.utils import random_str


m = Model(
 model_data=est.model_data(),
 # Use the PyTorch processor provided by PAI.
 inference_spec=InferenceSpec(processor="pytorch_cpu_1.10"),
)

p: Predictor = m.deploy(
 service_name="tutorial_pt_mnist_proc_{}".format(random_str(6)),
 instance_type="ecs.c6.xlarge",
)

print(p.service_name)
print(p.service_status)

Model.de ployは新しい推論サービスを作成し、Predictorオブジェクトを返します。 Predictor.predictメソッドを使用して、推論サービスにリクエストを送信し、予測結果を取得できます。

この例では、numpyを使用してテストサンプルを作成し、推論サービスに送信します。

import numpy as np

# The input is of the Float32 type and in the format of (BatchSize, Channel, Weight, Height).
dummy_input = np.random.rand(2, 1, 28, 28).astype(np.float32)

# np.random.rand(1, 1, 28, 28).dtype
res = p.predict(dummy_input)
print(res)

print(np.argmax(res, 1))

予測が完了した後、Predictor.de lete_serviceを使用して推論サービスを削除できます。

p.de lete_service()

イメージを使用してモデルを展開する

パフォーマンスに依存するシナリオでは、プロセッサを使用してモデルを配置できます。 モデルにサードパーティの依存関係がある場合や、推論サービスで前処理と後処理が必要な場合など、カスタム要件があるシナリオでは、イメージを使用してモデルをデプロイできます。 PAI Python SDKは、pai.mo del.container_serving_spec() メソッドを提供し、PAIが提供するオンプレミスコードとイメージを使用して推論サービスを作成できます。

モデルをデプロイする前に、モデルのロード、HTTPサーバーの起動、および推論要求の処理に使用するコードを準備する必要があります。 この例では、コードはFlaskを使用して記述されます。 サンプルコード：

import json
from flask import Flask, request
from PIL import Image
import os
import torch
import torchvision.transforms as transforms
import numpy as np
import io

app = Flask(__name__)
# The model is loaded to the current path by default.  
MODEL_PATH = "/eas/workspace/model/"

device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
model = torch.jit.load(os.path.join(MODEL_PATH, "mnist_cnn.pt"), map_location=device).to(device)
transform = transforms.Compose(
    [transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,))]
)


@app.route("/", methods=["POST"])
def predict():
    # Preprocess the image data.
    im = Image.open(io.BytesIO(request.data))
    input_tensor = transform(im).to(device)
    input_tensor.unsqueeze_(0)
    # Perform inference with the model.
    output_tensor = model(input_tensor)
    pred_res =output_tensor.detach().cpu().numpy()[0] 

    return json.dumps(pred_res.tolist())


if __name__ == '__main__':
    app.run(host="0.0.0.0", port=int(os.environ.get("LISTENING_PORT", 8000)))

上記のコードをオンプレミスのコンピューターに保存して、後で使用できるようにアップロードします。 この例では、ディレクトリinfer_srcを作成し、上記のコードをinfer_src/run.pyに保存します。 サンプルディレクトリ構造:

|-- infer_src # The code directory of the inference service to be uploaded.
    |-- requirements.txt # Optional. The third-party dependencies of the inference service.
    '-- run.py # The inference service script.

pai.mo del.container_serving_specを通じてPAIが提供するオンプレミスのスクリプトとPyTorchイメージに基づいてInferenceSpecオブジェクトを作成します。

• モデルサービスのコードと起動コマンド

source_dirパラメーターで指定されたオンプレミスのスクリプトディレクトリがOSSにアップロードされ、サービスコンテナーにマウントされます。 デフォルトでは、/ml/usercodeディレクトリが使用されます。

• 推論サービスに使用するイメージ

pai.image.retrieveメソッドを使用して、PAIが提供するイメージを取得できます。 イメージを取得するときに、image_scopeパラメーターをImageScope.INFERENCEに指定します。

• モデルサービスのサードパーティ依存

requirementsパラメーターを使用して、モデルサービスのコードまたは依存関係を指定できます。 このようにして、依存関係はサービスの開始前に環境にインストールされます。

Model.de ploy操作を呼び出して、トレーニング済みモデルとInferenceSpecを使用してオンライン推論サービスをデプロイできます。

from pai.model import InferenceSpec, container_serving_spec, Model
from pai.image import retrieve, ImageScope
from pai.common.utils import random_str
import numpy as np

torch_image_uri = retrieve(
    framework_name="pytorch", framework_version="1.12", accelerator_type="CPU"
).image_uri

inf_spec = container_serving_spec(
    command="python run.py",
    source_dir="./infer_src/",
    image_uri=torch_image_uri,
    requirements=["flask==2.0.0"],
)
print(inf_spec.to_dict())

m = Model(
    model_data=est.model_data(),
    inference_spec=inf_spec,
)

predictor = m.deploy(
    service_name="torch_container_{}".format(random_str(6)),
    instance_type="ecs.c6.xlarge",
)

この例では、MNISTイメージが推論サービスに送信されます。

import base64
from PIL import Image
from IPython import display
import io


# raw_data is an MNIST image, which corresponds to the number 9.
raw_data = base64.b64decode(b"/9j/4AAQSkZJRgABAQAAAQABAAD/2wBDAAgGBgcGBQgHBwcJCQgKDBQNDAsLDBkSEw8UHRofHh0aHBwgJC4nICIsIxwcKDcpLDAxNDQ0Hyc5PTgyPC4zNDL/wAALCAAcABwBAREA/8QAHwAAAQUBAQEBAQEAAAAAAAAAAAECAwQFBgcICQoL/8QAtRAAAgEDAwIEAwUFBAQAAAF9AQIDAAQRBRIhMUEGE1FhByJxFDKBkaEII0KxwRVS0fAkM2JyggkKFhcYGRolJicoKSo0NTY3ODk6Q0RFRkdISUpTVFVWV1hZWmNkZWZnaGlqc3R1dnd4eXqDhIWGh4iJipKTlJWWl5iZmqKjpKWmp6ipqrKztLW2t7i5usLDxMXGx8jJytLT1NXW19jZ2uHi4+Tl5ufo6erx8vP09fb3+Pn6/9oACAEBAAA/APn+rVhpmoarP5GnWNzeTYz5dvE0jfkoJovNMv8ATmK3tjc2zByhE8TIQw6jkdR6VVq9oumPrWuWGlxyLG95cRwK7dFLMFyfzr3aXwp4ltAfB3gWwudI01JNuoa7eZhku5AMHafvFOw2Dn6ZJ4z4yeLk1HUbXwrZSSy2Oh5heeaQu88wG1mLHk4wR9c+1eXUqsVYMpIIOQR2r1D4QazqOs/FnSG1fVLi9ZI5vL+2TNKc+U2ApYnB7/hXml5LLNfXEsxLSvIzOSMEsTk1DRVnT7+60vULe/spmhureQSRSL1Vh0NWNd1mXX9ZuNUuLe2gmuCGkS2QohbABbBJwTjJ9yelZ1f/2Q==")

im = Image.open(io.BytesIO(raw_data))
display.display(im)

推論サービスは、HTTPリクエストボディのデータを入力イメージとして使用します。 raw_predictメソッドは、データがbytesタイプのリクエストを受け取ります。 次に、PAI Python SDKはPOSTメソッドを使用してリクエスト本文に推論データを含め、そのデータを推論サービスに送信します。

from pai.predictor import RawResponse
import numpy as np

resp: RawResponse = predictor.raw_predict(data=raw_data)
print(resp.json())

print(np.argmax(resp.json()))

テストの完了後にサービスを削除できます。

predictor.de lete_service()