Platform For AI:クイックスタート：DLC を使用した分散トレーニング

3.1 データセットの作成

データセットには、トレーニングコード、データ、モデルアーティファクトが保存されます。このセクションでは、Alibaba Cloud Object Storage Service (OSS) に基づくデータセットの使用方法について説明します。

1. PAI コンソールの左側のナビゲーションウィンドウで、データセット > カスタムデータセット > データセットの作成 を選択します。

2. 次の主要なパラメーターを設定し、その他のパラメーターはデフォルト値のままにします。

   • 名前：たとえば、dataset_mnist と入力します。

   • ストレージクラス：Object Storage Service (OSS) を選択します。

   • OSS パス：フォルダアイコン をクリックし、バケットを選択して、dlc_mnist などの新しいディレクトリを作成します。

     OSS を有効化していない場合、または現在のリージョンで利用可能なバケットがない場合は、次の手順に従って OSS を有効化し、バケットを作成します。

     (オプション) OSS の有効化とバケットの作成

     1. OSS を有効化します。

     2. OSS コンソールにログインし、バケットの作成 をクリックして、[バケット名] を入力し、PAI と同じ リージョン を選択します。その他のパラメーターはデフォルト値のままにし、[作成] をクリックします。

   [確認] をクリックしてデータセットを作成します。

3. トレーニングコードとデータをアップロードします。

   1. 提供されているトレーニングスクリプト mnist_train.py をダウンロードします。プロセスを簡略化するため、このスクリプトは実行時にトレーニングデータをデータセットの dataSet ディレクトリに自動的にダウンロードします。

      ご自身のプロジェクトでは、コードとトレーニングデータを事前に PAI データセットにアップロードできます。

      シングルノード・シングル GPU トレーニングコードの例：mnist_train.py

      import torch
      import torch.nn as nn
      import torch.nn.functional as F
      import torch.optim as optim
      from torch.utils.data import DataLoader
      from torchvision import datasets, transforms
      from torch.utils.tensorboard import SummaryWriter
      # ハイパーパラメーター
      batch_size = 64  # 各トレーニング反復のバッチサイズ
      learning_rate = 0.01  # 学習率
      num_epochs = 20  # トレーニングエポック数
      # GPU が利用可能か確認
      device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
      # データの前処理
      transform = transforms.Compose([
          transforms.ToTensor(),
          transforms.Normalize((0.5,), (0.5,))
      ])
      train_dataset = datasets.MNIST(root='/mnt/data/dataSet', train=True, download=True, transform=transform)
      val_dataset = datasets.MNIST(root='/mnt/data/dataSet', train=False, download=False, transform=transform)
      train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
      val_loader = DataLoader(val_dataset, batch_size=batch_size, shuffle=False)
      # シンプルなニューラルネットワークを定義
      class SimpleCNN(nn.Module):
          def __init__(self):
              super(SimpleCNN, self).__init__()
              # 最初の畳み込み層：入力チャンネル 1 (グレースケール画像)、出力チャンネル 10、カーネルサイズ 5x5
              self.conv1 = nn.Conv2d(1, 10, kernel_size=5)
              # 2 番目の畳み込み層：入力チャンネル 10、出力チャンネル 20、カーネルサイズ 3x3
              self.conv2 = nn.Conv2d(10, 20, kernel_size=3)
              # 全結合層：入力は 20 * 5 * 5 (畳み込みとプーリング後の特徴マップサイズ)、出力は 128
              self.fc1 = nn.Linear(20 * 5 * 5, 128)
              # 出力層：128 -> 10 (10 個の数字クラス用)
              self.fc2 = nn.Linear(128, 10)
          def forward(self, x):
              # 入力 x の形状：[batch, 1, 28, 28]
              x = F.max_pool2d(F.relu(self.conv1(x)), 2)  # [batch, 10, 12, 12]
              x = F.max_pool2d(F.relu(self.conv2(x)), 2)  # [batch, 20, 5, 5]
              x = x.view(-1, 20 * 5 * 5)  # [batch, 500] にフラット化
              x = F.relu(self.fc1(x))      # [batch, 128]
              x = self.fc2(x)              # [batch, 10]
              return x
      # モデルをインスタンス化し、利用可能であれば GPU に移動
      model = SimpleCNN().to(device)
      criterion = nn.CrossEntropyLoss()
      optimizer = optim.SGD(model.parameters(), lr=learning_rate)
      # TensorBoard SummaryWriter を作成してモデルのトレーニングプロセスを可視化
      writer = SummaryWriter('/mnt/data/output/runs/mnist_experiment')
      # 最高精度のモデルを保存するための変数
      best_val_accuracy = 0.0
      # モデルをトレーニングし、損失と精度をログに記録
      for epoch in range(num_epochs):
          model.train()
          for batch_idx, (data, target) in enumerate(train_loader):
              data, target = data.to(device), target.to(device)  # データとターゲットを GPU に移動
              # 勾配をゼロにする
              optimizer.zero_grad()
              # フォワードパス
              output = model(data)
              # 損失を計算
              loss = criterion(output, target)
              # バックワードパス
              loss.backward()
              # パラメーターを更新
              optimizer.step()
              # トレーニング損失を TensorBoard にログ記録
              if batch_idx % 100 == 0:  # 100 バッチごとにログ記録
                  writer.add_scalar('Loss/train', loss.item(), epoch * len(train_loader) + batch_idx)
                  print(f'Train Epoch: {epoch} [{batch_idx * len(data)}/{len(train_loader.dataset)} ({100. * batch_idx / len(train_loader):.0f}%)]\tLoss: {loss.item():.6f}')
          # モデルを検証し、検証損失と精度をログに記録
          model.eval()
          val_loss = 0
          correct = 0
          with torch.no_grad():  # 勾配を計算しない
              for data, target in val_loader:
                  data, target = data.to(device), target.to(device)  # データとターゲットを GPU に移動
                  output = model(data)
                  val_loss += criterion(output, target).item()  # 検証損失を累積
                  pred = output.argmax(dim=1, keepdim=True)  # 予測ラベルを取得
                  correct += pred.eq(target.view_as(pred)).sum().item()  # 正解予測数を累積
          val_loss /= len(val_loader)  # 平均検証損失を計算
          val_accuracy = 100. * correct / len(val_loader.dataset)  # 検証精度を計算
          print(f'Validation Loss: {val_loss:.4f}, Accuracy: {correct}/{len(val_loader.dataset)} ({val_accuracy:.0f}%)')
          # 検証損失と精度を TensorBoard にログ記録
          writer.add_scalar('Loss/validation', val_loss, epoch)
          writer.add_scalar('Accuracy/validation', val_accuracy, epoch)
          # 最高の検証精度を持つモデルを保存
          if val_accuracy > best_val_accuracy:
              best_val_accuracy = val_accuracy
              torch.save(model.state_dict(), '/mnt/data/output/best_model.pth')
              print(f'Model saved with accuracy: {best_val_accuracy:.2f}%')
      # SummaryWriter を閉じる
      writer.close()
      print('Training complete. writer.close()')

   2. コードをアップロードします。データセットの詳細ページで、データを表示 をクリックして OSS コンソールに移動します。次に、オブジェクト > ファイルの選択 > アップロード をクリックして、トレーニングスクリプトを OSS にアップロードします。

3.2 DLC ジョブの作成

1. PAI コンソールの左側のナビゲーションウィンドウで、コンテナトレーニング (DLC) > タスクの作成 を選択します。

2. 次の主要なパラメーターを設定し、その他のパラメーターはデフォルト値のままにします。パラメーターの完全なリストについては、「トレーニングジョブの作成」をご参照ください。

   • イメージの構成：イメージアドレス を選択し、ご利用の リージョン に適したイメージアドレスを入力します。

     コンソールの上部のナビゲーションバーで、[リージョン] セレクターを使用して、現在のリージョン (例：中国 (杭州)) を確認します。

     リージョン	イメージアドレス
     中国 (北京)	dsw-registry-vpc.cn-beijing.cr.aliyuncs.com/pai/modelscope:1.28.0-pytorch2.3.1tensorflow2.16.1-gpu-py311-cu121-ubuntu22.04
     中国 (上海)	dsw-registry-vpc.cn-shanghai.cr.aliyuncs.com/pai/modelscope:1.28.0-pytorch2.3.1tensorflow2.16.1-gpu-py311-cu121-ubuntu22.04
     中国 (杭州)	dsw-registry-vpc.cn-hangzhou.cr.aliyuncs.com/pai/modelscope:1.28.0-pytorch2.3.1tensorflow2.16.1-gpu-py311-cu121-ubuntu22.04
     その他	リージョン ID を検索し、イメージアドレスの <Region ID> を置き換えて完全な URL を取得します。 dsw-registry-vpc.<Region ID>.cr.aliyuncs.com/pai/modelscope:1.28.0-pytorch2.3.1tensorflow2.16.1-gpu-py311-cu121-ubuntu22.04

     このイメージは、「DSW 対話型モデリングのクイックスタート」トピックと同じ検証済み環境を使用します。PAI でモデリングを行う際の一般的なワークフローは、まず DSW で環境を検証してコードを開発し、次に DLC を使用してトレーニングを行うことです。

   • データセットのマウント：カスタムデータセット を選択し、前のステップで作成したデータセットを選択します。デフォルトの マウントパス は /mnt/data です。

   • 開始コマンド：python /mnt/data/mnist_train.py

     起動コマンドは、DSW またはローカルでコードを実行する場合と同じです。ただし、mnist_train.py は /mnt/data/ にマウントされているため、コードパスを /mnt/data/mnist_train.py に変更するだけで済みます。

   • リソースソース：パブリックリソース を選択し、リソース仕様 に ecs.gn7i-c8g1.2xlarge を選択します。

     このインスタンスタイプが在庫切れの場合は、別の GPU インスタンスタイプを選択できます。

   OK をクリックしてジョブを作成します。ジョブの実行には約 15 分かかります。実行中に、[ログ] をクリックしてトレーニングの進捗を確認できます。

   トレーニングログから、MNIST モデルがエポック 18 で 9886/10000 (約 99%) の検証精度に達したことがわかります。ジョブのステータスが[成功]に変更されると、ログに最終出力 Training complete. writer.close() が表示されます。

   ジョブが完了すると、DLC は最適なモデルチェックポイントと TensorBoard ログを、マウントされたデータセットの output パスに保存します。

   OSS バケットの dlc_mnist/ ディレクトリには、トレーニングデータ用の dataSet/ サブディレクトリ、モデルアーティファクト用の output/ サブディレクトリ、および mnist_train.py スクリプトが含まれるようになります。

3.3 (オプション) TensorBoard での表示

TensorBoard の可視化ツールを使用して損失曲線を表示し、トレーニングプロセスをより深く理解できます。

1. DLC job の詳細ページで、Tensorboard タブをクリックして、Tensorboardの新規作成 をクリックします。

2. 構成タイプ を タスク別 に設定します。Summary 目次 には、トレーニングコードで SummaryWriter がデータを保存するパス /mnt/data/output/runs/ を入力します。OK をクリックして開始します。

   これは、次のコードスニペットに対応します：writer = SummaryWriter('/mnt/data/output/runs/mnist_experiment')

3. Tensorboard の表示 をクリックして、train_loss 曲線 (トレーニングセットの損失) と validation_loss 曲線 (検証データセットの損失) を確認します。

   

   (オプション) ハイパーパラメーターを調整してモデルのパフォーマンスを向上させる

   損失値の傾向を観察することで、モデルのパフォーマンスを評価できます。

   • トレーニングが完了する前に train_loss と validation_loss がまだ減少傾向にある (学習不足)。

     num_epochs (トレーニングエポック数) を増やすか、learning_rate をわずかに上げてモデルを再トレーニングし、トレーニングデータにより良く適合させることができます。

   • train_loss は継続的に減少するが、トレーニングが完了する前に validation_loss が増加し始める (過学習)。

     num_epochs を減らすか、learning_rate をわずかに下げてモデルを再トレーニングし、過学習を防ぐことができます。

   • トレーニングが完了する前に train_loss と validation_loss の両方が安定している (良好な適合)。

     モデルがこの状態にある場合、次のステップに進むことができます。

3.4 トレーニング済みモデルのデプロイ

詳細については、「V. EAS を使用したモデルサービスのデプロイ」をご参照ください。

4.1 データセットとスクリプトの準備

「3. シングルノード・シングル GPU トレーニング」セクションで既にデータセットを作成している場合は、mnist_train_distributed.py をダウンロードしてアップロードするだけです。そうでない場合は、まず「3.1 データセットの作成」を行い、その後コードをアップロードします。

import os
import torch
import torch.distributed as dist
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torch.utils.data import DataLoader
from torch.utils.data.distributed import DistributedSampler
from torchvision import datasets, transforms
from torch.utils.tensorboard import SummaryWriter
class SimpleCNN(nn.Module):
    def __init__(self):
        super(SimpleCNN, self).__init__()
        self.conv1 = nn.Conv2d(1, 10, kernel_size=5)
        self.conv2 = nn.Conv2d(10, 20, kernel_size=3)
        self.fc1 = nn.Linear(20 * 5 * 5, 128)
        self.fc2 = nn.Linear(128, 10)
    def forward(self, x):
        x = F.max_pool2d(F.relu(self.conv1(x)), 2)
        x = F.max_pool2d(F.relu(self.conv2(x)), 2)
        x = x.view(-1, 20 * 5 * 5)
        x = F.relu(self.fc1(x))
        x = self.fc2(x)
        return x
def main():
    rank = int(os.environ["RANK"])
    world_size = int(os.environ["WORLD_SIZE"])
    local_rank = int(os.environ["LOCAL_RANK"])
    dist.init_process_group(backend='nccl')
    torch.cuda.set_device(local_rank)
    device = torch.device('cuda', local_rank)
    batch_size = 64
    learning_rate = 0.01
    num_epochs = 20
    transform = transforms.Compose([
        transforms.ToTensor(),
        transforms.Normalize((0.5,), (0.5,))
    ])
    # メインプロセス (rank=0) のみがダウンロードする必要があり、他のプロセスはそれを待つ必要があります。
    # rank != 0 のプロセスをバリアで待機させます。
    if rank != 0:
        dist.barrier()
    # すべてのプロセスがデータセットを作成します。
    # ただし、実際にダウンロードを実行するのは rank 0 のプロセスのみです。
    train_dataset = datasets.MNIST(root='/mnt/data/dataSet', train=True, download=(rank == 0), transform=transform)
    # rank 0 のプロセスがダウンロードを完了すると、それもバリアに到達し、すべてのプロセスを解放します。
    if rank == 0:
        dist.barrier()
    # この時点で、すべてのプロセスが同期され、続行できます。
    val_dataset = datasets.MNIST(root='/mnt/data/dataSet', train=False, download=False, transform=transform)
    train_sampler = DistributedSampler(train_dataset, num_replicas=world_size, rank=rank, shuffle=True)
    train_loader = DataLoader(train_dataset, batch_size=batch_size, sampler=train_sampler, num_workers=4, pin_memory=True)
    val_loader = DataLoader(val_dataset, batch_size=batch_size, shuffle=False, num_workers=4, pin_memory=True)
    model = SimpleCNN().to(device)
    model = torch.nn.parallel.DistributedDataParallel(model, device_ids=[local_rank])
    criterion = nn.CrossEntropyLoss().to(device)
    optimizer = optim.SGD(model.parameters(), lr=learning_rate)
    if rank == 0:
        writer = SummaryWriter('/mnt/data/output_distributed/runs/mnist_experiment')
    best_val_accuracy = 0.0
    for epoch in range(num_epochs):
        train_sampler.set_epoch(epoch)
        model.train()
        for batch_idx, (data, target) in enumerate(train_loader):
            data, target = data.to(device, non_blocking=True), target.to(device, non_blocking=True)
            optimizer.zero_grad()
            output = model(data)
            loss = criterion(output, target)
            loss.backward()
            optimizer.step()
            if batch_idx % 100 == 0:
                # 各 rank と local_rank が自身の損失を出力します。
                print(f"Rank: {rank}, Local_Rank: {local_rank} -- Train Epoch: {epoch} "
                      f"[{batch_idx * len(data) * world_size}/{len(train_loader.dataset)} "
                      f"({100. * batch_idx / len(train_loader):.0f}%)]\tLoss: {loss.item():.6f}")
                if rank == 0:
                    writer.add_scalar('Loss/train', loss.item(), epoch * len(train_loader) + batch_idx)
        # 検証
        model.eval()
        val_loss = 0
        correct = 0
        total = 0
        with torch.no_grad():
            for data, target in val_loader:
                data, target = data.to(device, non_blocking=True), target.to(device, non_blocking=True)
                output = model(data)
                val_loss += criterion(output, target).item() * data.size(0)
                pred = output.argmax(dim=1, keepdim=True)
                correct += pred.eq(target.view_as(pred)).sum().item()
                total += target.size(0)
        val_loss_tensor = torch.tensor([val_loss], dtype=torch.float32, device=device)
        correct_tensor = torch.tensor([correct], dtype=torch.float32, device=device)
        total_tensor = torch.tensor([total], dtype=torch.float32, device=device)
        dist.all_reduce(val_loss_tensor, op=dist.ReduceOp.SUM)
        dist.all_reduce(correct_tensor, op=dist.ReduceOp.SUM)
        dist.all_reduce(total_tensor, op=dist.ReduceOp.SUM)
        val_loss = val_loss_tensor.item() / total_tensor.item()
        val_accuracy = 100. * correct_tensor.item() / total_tensor.item()
        if rank == 0:
            print(f'Validation Loss: {val_loss:.4f}, Accuracy: {int(correct_tensor.item())}/{int(total_tensor.item())} ({val_accuracy:.0f}%)')
            writer.add_scalar('Loss/validation', val_loss, epoch)
            writer.add_scalar('Accuracy/validation', val_accuracy, epoch)
            if val_accuracy > best_val_accuracy:
                best_val_accuracy = val_accuracy
                torch.save(model.module.state_dict(), '/mnt/data/output_distributed/best_model.pth')
                print(f'Model saved with accuracy: {best_val_accuracy:.2f}%')
    if rank == 0:
        writer.close()
    dist.destroy_process_group()
    if rank == 0:
        print('Training complete. writer.close()')
if __name__ == "__main__":
    main()

4.2 分散 DLC ジョブの作成

1. PAI コンソールの左側のナビゲーションウィンドウで、分散トレーニング (DLC) > タスクの作成 を選択します。

2. 次の主要なパラメーターを設定し、その他のパラメーターはデフォルト値のままにします。パラメーターの完全なリストについては、「トレーニングジョブの作成」をご参照ください。

   • イメージの構成：イメージアドレス を選択し、ご利用の リージョン に適したイメージアドレスを入力します。

     リージョン	イメージアドレス
     中国 (北京)	dsw-registry-vpc.cn-beijing.cr.aliyuncs.com/pai/modelscope:1.28.0-pytorch2.3.1tensorflow2.16.1-gpu-py311-cu121-ubuntu22.04
     中国 (上海)	dsw-registry-vpc.cn-shanghai.cr.aliyuncs.com/pai/modelscope:1.28.0-pytorch2.3.1tensorflow2.16.1-gpu-py311-cu121-ubuntu22.04
     中国 (杭州)	dsw-registry-vpc.cn-hangzhou.cr.aliyuncs.com/pai/modelscope:1.28.0-pytorch2.3.1tensorflow2.16.1-gpu-py311-cu121-ubuntu22.04
     その他	リージョン ID を検索し、イメージアドレスの <Region ID> を置き換えて完全な URL を取得します。 dsw-registry-vpc.<Region ID>.cr.aliyuncs.com/pai/modelscope:1.28.0-pytorch2.3.1tensorflow2.16.1-gpu-py311-cu121-ubuntu22.04

     このイメージは、「DSW 対話型モデリングのクイックスタート」トピックと同じ検証済み環境を使用します。PAI でモデリングを行う際の一般的なワークフローは、まず DSW で環境を検証してコードを開発し、次に DLC を使用してトレーニングを行うことです。

   • データセットのマウント：カスタムデータセット を選択し、作成したデータセットを選択します。デフォルトの マウントパス は /mnt/data です。

   • 開始コマンド：torchrun --nproc_per_node=1 --nnodes=${WORLD_SIZE} --node_rank=${RANK} --master_addr=${MASTER_ADDR} --master_port=${MASTER_PORT} /mnt/data/mnist_train_distributed.py

     DLC は、一般的な環境変数 (たとえば、MASTER_ADDR や WORLD_SIZE) を自動的にインジェクトします。これらの変数には、$VARIABLE_NAME 構文でアクセスできます。

   • リソースソース：パブリックリソース を選択し、ノード数 を 2 に設定し、リソース仕様 に ecs.gn7i-c8g1.2xlarge を選択します。

     このインスタンスタイプが在庫切れの場合は、別の GPU インスタンスタイプを選択できます。

   OK をクリックしてジョブを作成します。ジョブの実行には約 10 分かかります。実行中に、概要 ページで両方のインスタンスのトレーニング ログ を表示できます。

   ジョブが完了すると、DLC は最適なモデルチェックポイントと TensorBoard ログを、マウントされたデータセットの output_distributed パスに保存します。

4.3 (オプション) TensorBoard での表示

TensorBoard の可視化ツールを使用して損失曲線を表示し、トレーニングプロセスをより深く理解できます。

1. DLC ジョブの詳細ページで、TensorBoard タブをクリックし、Tensorboardの新規作成 をクリックします。

2. 構成タイプ を タスク別 に設定します。Summary 目次 には、トレーニングコードがサマリーファイルを保存するパス /mnt/data/output_distributed/runs を入力します。OK をクリックして開始します。

   これは、コードスニペット writer = SummaryWriter('/mnt/data/output_distributed/runs/mnist_experiment')

3. Tensorboard の表示 をクリックして、train_loss 曲線 (トレーニングセットの損失) と validation_loss 曲線 (検証データセットの損失) を確認します。

   

   (オプション) ハイパーパラメーターを調整してモデルのパフォーマンスを向上させる

   損失値の傾向を観察することで、モデルのパフォーマンスを評価できます。

   • トレーニングが終了する前に、train_loss と validation_loss がまだ減少している (学習不足)。

     num_epochs (トレーニングエポック数) を増やすか、learning_rate をわずかに上げてモデルを再トレーニングし、トレーニングデータにより良く適合させることができます。

   • train_loss は継続的に減少するが、トレーニングが終了する前に validation_loss が増加し始める (過学習)。

     num_epochs を減らすか、learning_rate をわずかに下げてモデルを再トレーニングし、過学習を防ぐことができます。

   • トレーニングが終了する前に train_loss と validation_loss の両方が安定している (良好な適合)。

     モデルがこの状態にある場合、次のステップに進むことができます。

4.4 トレーニング済みモデルのデプロイ

詳細については、「V. EAS を使用したモデルサービスのデプロイ」をご参照ください。