使用OSS加速器提升模型训练速度_对象存储(OSS)-阿里云帮助中心

使用OSS加速器可以显著提升数据集加载速度，从而提升整体的模型训练速度。本文基于大量性能测试对比了使用和未使用OSS加速器的加速效果。说明在GPU利用率尚未达到瓶颈的情况下，数据加载效率至关重要。此外，本文以预训练模型ResNet-18在Imagenet ILSVRC数据集上进行微调的训练任务为例，指导您如何在GPU云服务器中使用OSS加速器来提升模型训练速度。

加速效果

对比标准OSS Bucket，OSS加速器展现出了显著的性能优势。凭借其低延迟特性，OSS加速器在使用较少的worker数量时便能实现更高的吞吐量表现。在多组实验中，OSS加速器提升训练效率40%～400%，大幅降低了计算资源的消耗与使用成本，为用户提供了更高效的解决方案。

性能测试实验结果

重要

以下性能测试结果仅供参考，实际训练加速效果可能受到多种因素的影响，例如数据集大小、硬件配置、模型复杂性以及其他超参数设置等。

针对使用了加速器的OSS Bucket和未使用加速器的标准OSS Bucket，开展了模型训练过程中的性能测试工作。此次测试所采用的数据集由128万张图片的训练集以及5万张图片的验证集构成，根据机器规格（4c15g+1*Tesla T4）设计了多组并发参数，并且分别使用标准OSS和加速器进行数据集加载完成多组实验。具体测试结果如下所示。

batch size	worker数量	平均每epoch耗时（min）
batch size	worker数量	标准OSS	加速器
64	6	63.18	34.70
	4	54.96	34.68
	2	146.05	34.66
32	6	82.19	37.11
	4	108.33	37.13
	2	137.87	37.30
16	6	68.93	41.58
	4	132.97	41.69
	2	206.32	41.69

方案概览

在GPU云服务器上使用OSS加速器加速加载训练数据的过程如下：

要实现以上加速效果只需三步：

创建GPU云服务器：构建一个契合模型训练工作需求的GPU云服务器。
创建OSS Bucket并开通OSS加速器：创建一个OSS Bucket存储空间并开通OSS加速器，同时获取Bucket内网域名以及OSS加速器域名用于后续训练任务。
训练模型：在完成上述准备工作后，先对原始数据集进行预处理，再将其上传至OSS。随后，在训练过程中使用OSS加速器把数据集加载到本地，以此开展模型的训练工作。

操作步骤

步骤一：创建GPU云服务器

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

1. 创建GPU云服务器

前往实例创建页。
选择自定义购买页签。
按需选择付费类型、地域、网络及可用区、实例规格、镜像等配置。完成创建。有关各配置项详细说明，请参见配置项说明。
- 本次实践ECS实例所采用规格为ecs.gn6i-c4g1.xlarge，仅供参考。
- 选择操作系统镜像以Ubuntu 22.04版本为例，并勾选安装GPU驱动选择CUDA版本为CUDA 版本 12.4.1，届时服务器启动时会自动安装CUDA环境，无需额外手动配置。

2. 连接GPU云服务器

在云服务器ECS控制台的实例列表页面，根据地域、实例ID找到创建好的ECS实例，单击操作列下的远程连接。
在远程连接对话框中，单击通过Workbench远程连接对应的立即登录。
在登录实例对话框中，选择认证方式为SSH密钥认证，用户名为ecs-user，输入或上传创建密钥对时下载的私钥文件，单击确定，即可登录ECS实例。
说明
私钥文件在创建密钥对时自动下载到本地，请您关注浏览器的下载记录，查找.pem格式的私钥文件。
显示如下页面后，说明您已成功登录ECS实例且CUDA驱动已开始自动安装，只需等待安装完成即可。

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

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

创建Bucket并获取内网域名
重要
请确保您所创建的Bucket与目标GPU云服务器位于同一地域，本此实践以杭州地域为例。
1. 进入对象存储OSS控制台的Bucket列表页面，单击创建Bucket。
2. 在创建Bucket面板，依据面板所给出的提示信息来完成Bucket的创建工作。
3. 在目标Bucket概览页面，访问端口区域，复制ECS的VPC网络访问（内网）的Endpoint留作备用，以便在后续训练时上传数据集和checkpoint到目标Bucket。
开通OSS加速器并获取加速器域名
1. 进入对象存储OSS控制台的Bucket列表页面，选择目标Bucket后在左侧导航栏，选择Bucket配置 > OSS加速器进入OSS加速器页面
2. 单击设置。选中我已阅读服务试用条款，单击下一步。
3. 在创建OSS加速器面板设置加速器容量，本次实验以配置500 GB为例。单击下一步。
4. 选择指定路径加速，并配置加速路径为数据集所在目录后单击确定，然后根据页面指引完成加速器的创建。
5. 在OSS加速器界面复制加速器域名留作备用，以便在后续训练时用于从目标Bucket下载数据集。

步骤三：训练模型

以下步骤旨在GPU云服务器上，完成模型训练环境的配置、数据集的上传，以及使用OSS加速器域名来加速模型训练。

说明

完整代码工程参见demo.tar.gz。

配置训练环境
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作为访问凭证的配置示例，示例中的<Access-key-id>、<Access-key-secret>请分别替换为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
    }
}
```

准备数据

上传训练数据集和验证数据集到目标Bucket。

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

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

执行以下命令解压下载好的数据集，并在当前路径下创建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/

执行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)))

下载图像数据集标签文件，用于构建数据集的分类映射关系。

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

训练流程

构建用于处理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-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

构建预训练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

构建用于训练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)

构建用于整合模型训练流程的主脚本文件。该文件整合以上多个工具类开启模型训练。

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)

执行python3 main.py命令来开始模型训练，如图所示，已成功启动训练流程。

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