Object Storage Service:OSS アクセラレータを使用してモデルトレーニングを高速化する

ステップ 1: GPU 高速化 ECS インスタンスの作成

モデルトレーニング用に GPU 高速化 ECS インスタンスを作成して接続するには、次の手順に従います。インスタンスタイプは [ecs.gn6i-c4g1.xlarge]、オペレーティングシステムは Ubuntu 22.04、CUDA バージョンは 12.4.1 です。インスタンス構成をカスタマイズする際は、最新の CUDA バージョンを選択してください。

ステップ 2: OSS バケットの作成と OSS アクセラレータの有効化

データセットを保存するためのバケットを作成し、GPU 高速化 ECS インスタンスと同じリージョンでバケットの OSS アクセラレータを有効にするには、次の手順に従います。これにより、データセットへのアクセス速度が向上します。ECS インスタンスとバケットが同じリージョンにあり、内部エンドポイント経由でバケットにアクセスする場合、トラフィック料金は発生しないことに注意してください。

1. バケットの作成と内部エンドポイントの取得

   重要

   OSS アクセラレータ機能は、中国 (杭州)、中国 (上海)、中国 (北京)、中国 (ウランチャブ)、中国 (深圳)、およびシンガポールリージョンでパブリックプレビュー中です。作成するバケットがこれらのリージョンのいずれかにあり、GPU 高速化 ECS インスタンスと同じリージョンにあることを確認してください。このトピックでは、中国 (杭州) リージョンを例として使用します。

   1. OSS コンソールの バケットリスト ページに移動し、[バケットの作成] をクリックします。

   2. [バケットの作成] パネルで、画面の指示に従ってバケットを作成します。

   3. 宛先バケットの [概要] ページで、ポートセクションに移動します。[ECS からの VPC 経由のアクセス (内部ネットワーク)] の [エンドポイント] をコピーし、後で使用するために保存します。このエンドポイントを使用して、トレーニング中にデータセットとチェックポイントを宛先バケットにアップロードします。

      

2. OSS アクセラレータの有効化とアクセラレータエンドポイントの取得

   1. OSS コンソールの バケットリスト ページに移動します。宛先バケットを選択します。左側のナビゲーションウィンドウで、[バケット設定] > [OSS アクセラレータ] を選択して [OSS アクセラレータ] ページに移動します。

   2. [OSS アクセラレータの作成] をクリックします。**[OSS アクセラレータの作成]** パネルで、アクセラレータの容量を設定します。この例では、500 GB が使用されます。[次へ] をクリックします。

   3. [指定パスの高速化] を選択し、データセットが保存されているディレクトリへの高速化パスを設定してから、[OK] をクリックします。その後、画面の指示に従ってアクセラレータを作成します。

      

   4. **[OSS アクセラレータ]** ページで、アクセラレータのエンドポイントをコピーし、後で使用するために保存します。このエンドポイントを使用して、トレーニング中に宛先バケットからデータセットをダウンロードします。

      

ステップ 3: モデルのトレーニング

GPU 高速化 ECS インスタンスで次の手順に従い、モデルトレーニング環境を設定し、データセットをアップロードし、OSS アクセラレータのエンドポイントを使用してモデルトレーニングを高速化します。

1. トレーニング環境の設定

   1. conda 環境を準備し、依存関係を構成します。

      1. 次のコマンドを実行して conda をダウンロードしてインストールします。

         curl -L https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh -o /tmp/miniconda.sh && bash /tmp/miniconda.sh -b -p /opt/conda/ && rm /tmp/miniconda.sh && /opt/conda/bin/conda clean -tipy && export PATH=/opt/conda/bin:$PATH  && conda init bash && source ~/.bashrc && conda update conda 

      2. vim environment.yaml コマンドを実行して、environment.yaml という名前の conda 環境ファイルを作成して開きます。次の構成を追加してファイルを保存します。

         name: py312
         channels:
           - defaults
           - conda-forge
           - pytorch
         dependencies:
           - python=3.12
           - pytorch>=2.5.0
           - torchvision 
           - torchaudio 
           - transformers 
           - torchdata
           - oss2

      3. 次のコマンドを実行して、環境ファイルから py312 という名前の conda 環境を作成します。

         conda env create -f environment.yaml

      4. conda activate py312 コマンドを実行して、py312 という名前の conda 環境をアクティブ化します。次の図は、環境がアクティブ化されたことを示しています。

         

         重要

         後続の手順は、アクティブ化された conda 環境で実行してください。

   2. 環境変数を構成します。

      次のコマンドを実行して、データセットのアップロードに必要なアクセス認証情報を設定します。<ACCESS_KEY_ID> と <ACCESS_KEY_SECRET> を RAM ユーザーの AccessKey ID と AccessKey シークレットに置き換えます。AccessKey ID と AccessKey シークレットの作成方法の詳細については、「AccessKey ペアの作成」をご参照ください。

      export OSS_ACCESS_KEY_ID=<ACCESS_KEY_ID>
      export OSS_ACCESS_KEY_SECRET=<ACCESS_KEY_SECRET>

   3. OSS コネクタを設定します。

      1. 次のコマンドを実行して OSS コネクタをインストールします。

         pip install osstorchconnector

      2. 次のコマンドを実行して、アクセス認証情報設定ファイルを作成します。

         mkdir -p /root/.alibabacloud && touch /root/.alibabacloud/credentials

      3. vim /root/.alibabacloud/credentials コマンドを実行して設定ファイルを開きます。次の構成を追加してファイルを保存します。OSS コネクタの設定方法の詳細については、「AI/ML 用 OSS コネクタの設定」をご参照ください。

         以下は、アクセス認証情報として AccessKey ID と AccessKey シークレットを使用する設定例です。<AccessKeyId> と <AccessKeySecret> を RAM ユーザーの AccessKey ID と AccessKey シークレットに置き換えます。AccessKey ID と AccessKey シークレットの作成方法の詳細については、「AccessKey ペアの作成」をご参照ください。

         {
           "AccessKeyId": "LTAI************************",
           "AccessKeySecret": "At32************************"
         }

      4. 次のコマンドを実行して、認証情報ファイルに読み取り専用権限を設定し、AccessKey ID と AccessKey シークレットを保護します。

         chmod 400 /root/.alibabacloud/credentials

      5. 次のコマンドを実行して、OSS コネクタの設定ファイルを作成します。

         mkdir -p /etc/oss-connector/ && touch /etc/oss-connector/config.json

      6. vim /etc/oss-connector/config.json コマンドを実行して設定ファイルを開きます。次の構成を追加してファイルを保存します。ほとんどの場合、デフォルトの構成を使用できます。

         {
             "logLevel": 1,
             "logPath": "/var/log/oss-connector/connector.log",
             "auditPath": "/var/log/oss-connector/audit.log",
             "datasetConfig": {
                 "prefetchConcurrency": 24,
                 "prefetchWorker": 2
             },
             "checkpointConfig": {
                 "prefetchConcurrency": 24,
                 "prefetchWorker": 4,
                 "uploadConcurrency": 64
             }
         }
         

2. データの準備

   1. トレーニングデータセットと検証データセットを宛先バケットにアップロードします。

      1. 次のコマンドを実行して、トレーニングデータセットと検証データセットを ECS インスタンスにダウンロードします。このデータセットは実際の実験で使用されたデータセットではなく、テスト目的のみであることに注意してください。

         wget https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20241216/jsnenr/n04487081.tar
         wget https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20241218/dxrciv/n10148035.tar
         wget https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20241216/senwji/val.tar

      2. 次のコマンドを実行して、ダウンロードしたデータセットを解凍し、現在のパスに作成された `dataset` ディレクトリに配置します。

         tar -zxvf n10148035.tar && tar -zxvf n04487081.tar && tar -zxvf val.tar
         mkdir dataset && mkdir ./dataset/train && mkdir ./dataset/val
         mv n04487081 ./dataset/train/ && mv n10148035 ./dataset/train/ && mv IL*.JPEG ./dataset/val/

      3. python3 upload_dataset.py コマンドを実行してスクリプトを実行します。これにより、解凍されたデータセットが指定されたバケットにアップロードされます。

         # upload_dataset.py
         
         from torchvision import transforms
         from PIL import Image
         import oss2
         import os
         from oss2.credentials import EnvironmentVariableCredentialsProvider
         
         # 中国 (杭州) リージョンの内部エンドポイントを例として使用します。
         OSS_ENDPOINT = "oss-cn-hangzhou-internal.aliyuncs.com"    # OSS の内部エンドポイント。
         OSS_BUCKET_NAME = "<YourBucketName>"    # 宛先バケットの名前。
         BUCKET_REGION = "cn-hangzhou"    # 宛先バケットのリージョン。
         
         # OSS_URI_BASE: OSS バケット内のカスタムストレージプレフィックス。
         OSS_URI_BASE = "dataset/imagenet/ILSVRC/Data"
         
         def to_tensor(img_path):
             IMG_DIM_224 = 224
             compose = transforms.Compose([
                     transforms.RandomResizedCrop(IMG_DIM_224),
                     transforms.RandomHorizontalFlip(),
                     transforms.ToTensor(),
                     transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
                 ])
             img = Image.open(img_path).convert('RGB')
             img_tensor = compose(img)
             numpy_data = img_tensor.numpy()
             binary_data = numpy_data.tobytes()
             return binary_data
         
         def list_dir(directory):
             for root, _, files in os.walk(directory):
                 rel_root = os.path.relpath(root, start=directory)
                 for file in files:
                     rel_filepath = os.path.join(rel_root, file) if rel_root != '.' else file
                     yield rel_filepath
         IMG_DIR_BASE = "./dataset" 
         """
             IMG_DIR_BASE は、イメージが保存されているローカルパスです。絶対パスまたは相対パスを使用できます。
             ディレクトリ構造は、以下に示すように、実際のデータセット構造と一致している必要があります。
             {IMG_DIR_BASE}/
                 train/
                     n10148035/
                         n10148035_10034.JPEG
                         n10148035_10217.JPEG
                         ... 
                     n11879895/
                         n11879895_10016.JPEG
                         n11879895_10019.JPEG
                         ...
                     ...
                 val/
                     ILSVRC2012_val_00000001.JPEG
                     ILSVRC2012_val_00000002.JPEG
                     ...
         """
         
         bucket_api = oss2.Bucket(oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider()), OSS_ENDPOINT, OSS_BUCKET_NAME, region=BUCKET_REGION)
                 
         for phase in [ "val", "train"]:
             IMG_DIR = "%s/%s" % (IMG_DIR_BASE, phase)
             for _, img_relative_path in enumerate(list_dir(IMG_DIR)):
                 img_bin_name = img_relative_path.replace(".JPEG", ".pt")
                 object_key = "%s/%s/%s" % (OSS_URI_BASE, phase, img_bin_name)
                 bucket_api.put_object(object_key, to_tensor("%s/%s" % (IMG_DIR,img_relative_path)))

   2. イメージデータセットのラベルファイルをダウンロードして、データセットの分類マッピングを構築します。

      wget https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20241220/izpskr/imagenet_class_index.json
      wget https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20241220/lfilrp/ILSVRC2012_val_labels.json

3. トレーニングプロセス

   1. ImageNet データセットを処理するためのユーティリティモジュールを構築します。このモジュールは、OSS アクセラレータのエンドポイントを使用して OSS アクセラレータからデータセットをダウンロードし、データローダーを構築します。

      oss_dataloader.py

      # oss_dataloader.py
      
      import json
      import numpy as np
      from torch.utils.data import DataLoader
      import torch
      
      class ImageCls():
          def __init__(self):
              self.__syn_to_class = {}
              self.__syn_to_label = {}
              with open("imagenet_class_index.json", "rb") as f:
                  cls_list = json.load(f)
                  for cls, v in cls_list.items():
                      syn = v[0]
                      label = v[1]
                      self.__syn_to_class[syn] = int(cls)
                      self.__syn_to_label[int(cls)] = label
      
          def __len__(self):
              return len(self.__syn_to_label)
          
          def __getitem__(self, syn):
              cls = self.__syn_to_class[syn]
              return cls
      
      class ImageValSet():
          def __init__(self):
              self.__val_to_syn = {}
              with open("ILSVRC2012_val_labels.json", "rb") as f:
                  val_syn_list = json.load(f)
                  for val, syn in val_syn_list.items():
                      self.__val_to_syn[val] = syn
          
          def __getitem__(self, val):
              return self.__val_to_syn[val]
      
      imageCls = ImageCls()
      imageValSet = ImageValSet()
      
      
      IMG_DIM_224 = 224
      OSS_URI_BASE = "oss://<YourBucketName>/dataset/imagenet/ILSVRC/Data"
      
      # OSS アクセラレータのエンドポイント。データセットのダウンロードに使用します。お使いの OSS アクセラレータのエンドポイントに置き換えてください。
      ENDPOINT = "cn-hangzhou-j-internal.oss-data-acc.aliyuncs.com" 
      
      def obj_to_tensor(object):
          data = object.read()
          numpy_array_from_binary = np.frombuffer(data, dtype=np.float32).reshape([3, IMG_DIM_224, IMG_DIM_224])
          return torch.from_numpy(numpy_array_from_binary)
      
      def train_tensor_transform(object):
          tensor_from_binary = obj_to_tensor(object)
          key = object.key
          syn = key.split('/')[-2]
          
          return tensor_from_binary, imageCls[syn]
      
      def val_tensor_transform(object):
          tensor_from_binary = obj_to_tensor(object)
          key = object.key
          image_name = key.split('/')[-1].split('.')[0] + ".JPEG"
          return tensor_from_binary, imageCls[imageValSet[image_name]]
      
      
      def make_oss_dataloader(dataset, batch_size, num_worker, shuffle):
          image_datasets = {
              'train': dataset.from_prefix(OSS_URI_BASE + "/train/", endpoint=ENDPOINT, transform=train_tensor_transform),
              'val': dataset.from_prefix(OSS_URI_BASE + "/val/", endpoint=ENDPOINT, transform=val_tensor_transform),
          }
          dataloaders = {
              'train': DataLoader(image_datasets['train'], batch_size=batch_size, shuffle=shuffle, num_workers=num_worker),
              'val': DataLoader(image_datasets['val'], batch_size=batch_size, shuffle=shuffle, num_workers=num_worker)
          }
          
          return dataloaders

   2. 事前トレーニング済みの ResNet18 モデルを初期化するためのユーティリティモジュールを構築します。

      pre_trained_model.py

      # pre_trained_model.py
      
      from torchvision import models
      import torch.nn as nn
      import torch
      
      def make_resnet_model(cls_count=1000):
          device = torch.device("cuda:0")
          model = models.resnet18(pretrained=True)
          num_ftrs = model.fc.in_features
          model.fc = nn.Linear(num_ftrs, cls_count)
          
          model = model.to(device)
          if torch.cuda.device_count() > 1:
              model = nn.DataParallel(model)
          
          return model, device

   3. ResNet モデルをトレーニングするためのユーティリティモジュールを構築します。このモジュールは、指定されたモデル、データローダー、およびエポック数 (epoch_num) に基づいてモデルをトレーニングします。

      resnet_train.py

      # resnet_train.py
      
      from osstorchconnector import OssCheckpoint
      import torch.optim as optim
      import torch
      import torch.nn as nn
      
      OSS_CHECKPOINT_URI = "oss://<YourBucketName>/checkpoints/resnet18.pt"
      
      # OSS の内部エンドポイント。
      ENDPOINT = "oss-cn-hangzhou-internal.aliyuncs.com" 
      
      def train(model, dataloaders, device, epoch_num):
          optimizer = optim.SGD(model.parameters(), lr=0.001, momentum=0.9)
          exp_lr_scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=7, gamma=0.1)
          criterion = nn.CrossEntropyLoss()
          
          best_acc = 0.0
          for epoch in range(epoch_num):
              for phase in ['train', 'val']:
                  if phase == 'train':
                      model.train()
                  else:
                      model.eval()
      
                  running_loss = 0.0
                  running_corrects = 0
      
                  # データを反復処理します。
                  dataset_size = 0
                  for (inputs, labels) in dataloaders[phase]:
                      inputs = inputs.to(device)
                      labels = labels.to(device)
      
                      with torch.set_grad_enabled(phase == 'train'):
                          outputs = model(inputs)
                          _, preds = torch.max(outputs, 1)
                          loss = criterion(outputs, labels)
      
                          # トレーニングフェーズでのみ逆伝播と最適化を実行します。
                          if phase == 'train':
                              optimizer.zero_grad()
                              loss.backward()
                              optimizer.step()
      
                      # 統計。
                      running_loss += loss.item() * inputs.size(0)
                      running_corrects += torch.sum(preds == labels.data)
                      dataset_size += inputs.size(0)
                  
                  if phase == 'train':
                      exp_lr_scheduler.step()
      
                  epoch_loss = running_loss / dataset_size
                  epoch_acc = running_corrects / dataset_size
      
                  print(f'[Epoch {epoch}/{epoch_num - 1}][{phase}] {dataset_size} imgs {epoch_acc}')
      
                  
                  if phase == 'val' and epoch_acc > best_acc:
                      best_acc = epoch_acc
                      # チェックポイントを OSS にアップロードします。
                      checkpoint = OssCheckpoint(endpoint=ENDPOINT)
                      with checkpoint.writer(OSS_CHECKPOINT_URI) as checkpoint_writer:
                          torch.save(model.state_dict(), checkpoint_writer)

   4. モデルトレーニングプロセスを統合するためのメインスクリプトファイルを構築します。このファイルは、先行するユーティリティモジュールを統合してモデルトレーニングを開始します。

      main.py

      # main.py
      
      from oss_dataloader import make_oss_dataloader
      from pre_trained_model import make_resnet_model
      from osstorchconnector import OssMapDataset
      from resnet_train import train
      
      # 基本的なトレーニングパラメーター。
      NUM_EPOCHS = 30 # エポック数
      BATCH_SIZE = 64 # バッチサイズ
      NUM_WORKER = 4 # データローダーのワーカー数
      
      # 事前トレーニング済みの resnet18 モデルを使用します。
      model, device = make_resnet_model()
      
      # OssMapDataset データセットを使用して Dataloader を構築します。
      dataloaders = make_oss_dataloader(OssMapDataset, BATCH_SIZE, NUM_WORKER, True)
      
      # メイントレーニングプロセスを呼び出します。
      train(model, dataloaders, device, NUM_EPOCHS)

   5. python3 main.py コマンドを実行してモデルトレーニングを開始します。次の図は、トレーニングプロセスが正常に開始されたことを示しています。

      

4. 結果の確認

   バケットリスト ページに移動し、宛先バケットを選択して [ファイル管理] > [ファイル] をクリックします。`checkpoints` ディレクトリに `resnet18.pt` ファイルがあるか確認します。次の図に示すように、トレーニング完了後、チェックポイントファイルは正常に OSS にアップロードされます。