使用OSS加速器提升模型训练速度

步骤一：创建GPU云服务器

以下步骤旨在创建并连接一个适用于模型训练任务的GPU云服务器实例。该实例的规格为ecs.gn6i-c4g1.xlarge，操作系统为Ubuntu 22.04、CUDA版本为12.4.1。需注意，自定义实例配置时，CUDA版本请选择最新版本。

步骤二：创建OSS Bucket并开通OSS加速器

以下步骤旨在与目标GPU云服务器同一地域下，创建一个用于存储数据集的Bucket并开启该Bucket的OSS加速器空间，以提高数据集的访问速度。需注意，云服务器和Bucket处于同一地域且通过内网域名访问不会产生任何流量费用。

1. 创建Bucket并获取内网域名

   重要

   OSS加速器功能目前在华东1（杭州）、华东2（上海）、华北2（北京）、华北6（乌兰察布）、华南1（深圳）以及新加坡地域开放公测。请确保您创建的Bucket位于上述地域之一，并且与目标GPU云服务器处于同一地域。本文将以华东1（杭州）地域为例进行说明。

   1. 进入对象存储OSS控制台的Bucket列表页面，单击创建 Bucket。

   2. 在创建 Bucket面板，依据面板所给出的提示信息来完成Bucket的创建工作。

   3. 在目标Bucket概览页面，访问端口区域，复制ECS 的 VPC 网络访问（内网）的Endpoint留作备用，以便在后续训练时上传数据集和checkpoint到目标Bucket。

2. 开通OSS加速器并获取加速器域名

   1. 进入对象存储OSS控制台的Bucket列表页面，选择目标Bucket后在左侧导航栏，选择Bucket 配置 > OSS 加速器进入OSS 加速器页面。

   2. 单击创建OSS加速器，在创建OSS加速器面板设置加速器容量，本次实验以配置500 GB为例。单击下一步。

   3. 选择指定路径加速，并配置加速路径为数据集所在目录后单击确定，然后根据页面指引完成加速器的创建。

      加速路径填写 dataset/，并勾选同步预热。开启同步预热后，客户端通过加速器域名写入的数据会同时写入 OSS Bucket 和 OSS 加速器，再次读取时可获得更低延迟。

   4. 在OSS加速器界面复制加速器域名留作备用，以便在后续训练时用于从目标Bucket下载数据集。

步骤三：训练模型

以下步骤旨在GPU云服务器上，完成模型训练环境的配置、数据集的上传，以及使用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. 执行以下命令，基于conda环境文件创建名为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、AccessKeySecret。有关如何创建AccessKey ID和AccessKeySecret请参见创建AccessKey。

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

   3. 配置OSS Connector。

      1. 执行以下命令，安装OSS Connector。

         pip install osstorchconnector

      2. 执行以下命令，创建访问凭证配置文件。

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

      3. 执行vim /root/.alibabacloud/credentials命令，打开配置文件添加如下配置并保存。有关OSS Connector更多配置，请参见配置OSS Connector for AI/ML。

         使用AccessKey ID和AccessKey Secret作为访问凭证的配置示例，示例中的<AccessKeyId>、<AccessKeySecret>请分别替换为RAM用户的AccessKey ID、AccessKeySecret。如何创建AccessKey ID和AccessKeySecret请参见创建AccessKey。

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

      4. 执行以下命令，设置credentials文件只读权限，以保障AK、SK密钥安全。

         chmod 400 /root/.alibabacloud/credentials

      5. 执行以下命令，创建OSS Connector配置文件。

         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. 上传训练数据集和验证数据集到目标Bucket。

      1. 执行以下命令下载训练数据集和验证数据集至云服务器。需注意，此数据集并非真实实验场景所采用数据集，仅作测试用途。

         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命令运行此脚本，将解压好的数据集上传至指定的Bucket中。

         # 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名称 
         BUCKET_REGION = "cn-hangzhou"    #目标Bucket地域 
         # OSS_URI_BASE: 自定义OSS bucket中的存储前缀
         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加速器空间下载数据集并构建dataloader。

      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模型的工具模块。该模块通过给定的模型（model）、数据加载器（dataloaders），以及训练轮数（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
                      # checkpoint上传到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 # epoch number
      BATCH_SIZE = 64 # batch size
      NUM_WORKER = 4 # dataloader worker number
      # 使用预训练的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命令来开始模型训练，如下所示，已成功启动训练流程。

      (py312) root@xxx :~# python3 main.py
      /opt/conda/envs/py312/lib/python3.12/site-packages/torchvision/models/_utils.py:208: UserWarning: The parameter 'pretrained' is deprecated since 0.13 and may be removed in the future, please use 'weights' instead.
        warnings.warn()
      /opt/conda/envs/py312/lib/python3.12/site-packages/torchvision/models/_utils.py:223: UserWarning: Arguments other than a weight enum or `None` for 'weights' are deprecated since 0.13 and may be removed in the future. The current behavior is equivalent to passing `weights=ResNet18_Weights.IMAGENET1K_V1`. You can also use `weights=ResNet18_Weights.DEFAULT` to get the most up-to-date weights.
        warnings.warn(msg)
      Downloading: "https://download.pytorch.org/models/resnet18-f37072fd.pth" to /root/.cache/torch/hub/checkpoints/resnet18-f37072fd.pth
      100%|████████████████████████████████████████████████████████████████████████████████████████████████████████████| 44.7M/44.7M [00:10<00:00, 4.40MB/s]
      2024/12/20 13:36:34.853415| INFO |th=00000000000000000|dataset.cpp:222|new_oss_dataset:new oss dataset, uuid: 92251a5f-7f00-45b5-aaba-b8894ab15b0d id: 0, total: 1, pid: 20655, endpoint: cn-hangzhou-internal.oss-data-acc.aliyuncs.com
      2024/12/20 13:36:34.853437| INFO |th=00000000000000000|dataset.cpp:230|new_oss_dataset:[cred_path=/root/.alibabacloud/credentials][config_path=/etc/oss-connector/config.json]
      2024/12/20 13:36:34.853518| INFO |th=00000000000000000|dataset.cpp:255|new_oss_dataset:set log level: 1
      2024/12/20 13:36:34.853591| INFO |th=00000000000000000|dataset.cpp:262|new_oss_dataset:set log path: /var/log/oss-connector/connector_log_0
      /root/oss_dataloader.py:51: UserWarning: The given NumPy array is not writable, and PyTorch does not support non-writable tensors. This means writing to this tensor will result in undefined behavior. You may want to copy the array to protect its data or make it writable before converting it to a tensor. This type of warning will be suppressed for the rest of this program. (Triggered internally at /home/conda/feedstock_root/build_artifacts/libtorch_1733624414920/work/torch/csrc/utils/tensor_numpy.cpp:206.)
        return torch.from_numpy(numpy_array_from_binary)
      /root/oss_dataloader.py:51: UserWarning: The given NumPy array is not writable, and PyTorch does not support non-writable tensors. This means writing to this tensor will result in undefined behavior. You may want to copy the array to protect its data or make it writable before converting it to a tensor. This type of warning will be suppressed for the rest of this program. (Triggered internally at /home/conda/feedstock_root/build_artifacts/libtorch_1733624414920/work/torch/csrc/utils/tensor_numpy.cpp:206.)
        return torch.from_numpy(numpy_array_from_binary)
      /root/oss_dataloader.py:51: UserWarning: The given NumPy array is not writable, and PyTorch does not support non-writable tensors. This means writing to this tensor will result in undefined behavior. You may want to copy the array to protect its data or make it writable before converting it to a tensor. This type of warning will be suppressed for the rest of this program. (Triggered internally at /home/conda/feedstock_root/build_artifacts/libtorch_1733624414920/work/torch/csrc/utils/tensor_numpy.cpp:206.)
        return torch.from_numpy(numpy_array_from_binary)

4. 结果验证

   进入Bucket列表选择目标Bucket，单击文件管理 > 文件列表后查看checkpoints目录下的resnet18.pt文件。训练结束后checkpoint文件已成功上传至OSS。