快速提交PAI-DLC训练任务

准备环境及资源

10

开始教程前，请按以下步骤准备环境和资源：

【重要】PAI-DLC免费资源包只适用于本教程中的PAI-DLC产品。如果您领取了PAI-DLC资源包后，使用了PAI-DLC及PAI的其他产品功能（如PAI-DSW、PAI-EAS等），PAI-DLC产品产生的费用由资源包抵扣，其他产品功能产生的费用无法抵扣，会产生对应的费用账单。

准备训练脚本

5

1. 在本地创建torch_ddp_sample_code.py（Python）代码文件，文件内容如下。

   import datetime
   import logging
   import os
   import argparse
   
   from math import ceil
   from random import Random
   
   import torch
   import torch.distributed as dist
   import torch.nn as nn 
   import torch.nn.functional as F
   import torch.optim as optim 
   import torch.utils.data
   import torch.utils.data.distributed
   from torch._utils import _flatten_dense_tensors, _unflatten_dense_tensors
   from torch.autograd import Variable
   from torch.nn.modules import Module
   from torchvision import datasets, transforms
   
   gbatch_size = 128
   epochs = 10
   world_size = os.environ.get("WORLD_SIZE", "{}")
   rank = os.environ.get("RANK", "{}")
   
   class DistributedDataParallel(Module):
     def __init__(self, module):
       super(DistributedDataParallel, self).__init__()
       self.module = module
       self.first_call = True
   
       def allreduce_params():
         if self.needs_reduction:
           self.needs_reduction = False 
           buckets = {}
           for param in self.module.parameters():
             if param.requires_grad and param.grad is not None:
               tp = type(param.data)
               if tp not in buckets:
                 buckets[tp] = []
               buckets[tp].append(param)
           for tp in buckets:
             bucket = buckets[tp]
             grads = [param.grad.data for param in bucket]
             coalesced = _flatten_dense_tensors(grads)
             dist.all_reduce(coalesced)
             coalesced /= dist.get_world_size()
             
             for buf, synced in zip(grads, _unflatten_dense_tensors(coalesced, grads)):
               buf.copy_(synced)
   
       for param in list(self.module.parameters()):
         def allreduce_hook(*unused): 
           Variable._execution_engine.queue_callback(allreduce_params)  
   
         if param.requires_grad:
           param.register_hook(allreduce_hook)
   
     def weight_broadcast(self):
       for param in self.module.parameters():
         dist.broadcast(param.data, 0)
   
     def forward(self, *inputs, **kwargs):  
       if self.first_call:
         logging.info("first broadcast start")
         self.weight_broadcast()
         self.first_call = False
         logging.info("first broadcast done")
       self.needs_reduction = True  
       return self.module(*inputs, **kwargs)
   
   class Net(nn.Module):
     def __init__(self):
       super(Net, self).__init__()
       self.conv1 = nn.Conv2d(1, 10, kernel_size=5)
       self.conv2 = nn.Conv2d(10, 20, kernel_size=5)
       self.conv2_drop = nn.Dropout2d()
       self.fc1 = nn.Linear(320, 50)
       self.fc2 = nn.Linear(50, 10)
   
     def forward(self, x): 
       x = F.relu(F.max_pool2d(self.conv1(x), 2))
       x = F.relu(F.max_pool2d(self.conv2_drop(self.conv2(x)), 2))
       x = x.view(-1, 320)
       x = F.relu(self.fc1(x))
       x = F.dropout(x, training=self.training)
       x = self.fc2(x)
       return F.log_softmax(x)
   
   
   def partition_dataset(rank):
     dataset = datasets.MNIST(
       './data{}'.format(rank),
       train=True,
       download=True,
       transform=transforms.Compose([
         transforms.ToTensor(),
         transforms.Normalize((0.1307,), (0.3081,))
       ]))
     size = dist.get_world_size()
     bsz = int(gbatch_size / float(size))
     train_sampler = torch.utils.data.distributed.DistributedSampler(dataset)
     train_set = torch.utils.data.DataLoader(
       dataset, batch_size=bsz, shuffle=(train_sampler is None), sampler=train_sampler)
     return train_set, bsz
   
   
   def average_gradients(model):
     size = float(dist.get_world_size())
     group = dist.new_group([0])
     for param in model.parameters():
       dist.all_reduce(param.grad.data, op=dist.ReduceOp.SUM, group=group)
       
       param.grad.data /= size
   
   
   def run(gpu):
     rank = dist.get_rank()
     torch.manual_seed(1234)
     train_set, bsz = partition_dataset(rank)
     model = Net()
     if gpu:
       model = model.cuda()
       model = torch.nn.parallel.DistributedDataParallel(model)
     else:
       model = DistributedDataParallel(model)
     optimizer = optim.SGD(model.parameters(), lr=0.01, momentum=0.5)
   
     num_batches = ceil(len(train_set.dataset) / float(bsz))
     logging.info("num_batches = %s", num_batches)
     time_start = datetime.datetime.now()
     for epoch in range(epochs):
       epoch_loss = 0.0
       for data, target in train_set:
         if gpu:
           data, target = Variable(data).cuda(), Variable(target).cuda()
         else:
           data, target = Variable(data), Variable(target)
         optimizer.zero_grad()
         output = model(data)
         loss = F.nll_loss(output, target)
         epoch_loss += loss.item()
         loss.backward()
         average_gradients(model)
         optimizer.step()
       logging.info('Epoch {} Loss {:.6f} Global batch size {} on {} ranks'.format(
         epoch, epoch_loss / num_batches, gbatch_size, dist.get_world_size()))
     if gpu:
       logging.info("GPU training time= {}".format(  
         str(datetime.datetime.now() - time_start)))  
     else:
       logging.info("CPU training time= {}".format(  
         str(datetime.datetime.now() - time_start))) 
   
   
   if __name__ == "__main__":
     logging.basicConfig(level=logging.INFO,
                         format=('%(levelname)s|%(asctime)s'
                                 '|%(pathname)s|%(lineno)d| %(message)s'),
                         datefmt='%Y-%m-%dT%H:%M:%S',
                         )
     logging.getLogger().setLevel(logging.INFO)
   
     parser = argparse.ArgumentParser(description='Train Pytorch model using DDP')
     parser.add_argument('--gpu', action='store_true',
                         help='Use GPU and CUDA')
     parser.set_defaults(gpu=False)
     args = parser.parse_args()
     if args.gpu:
       logging.info("\n======= CUDA INFO =======")
       logging.info("CUDA Availability: %s", torch.cuda.is_available())
       if torch.cuda.is_available():
         logging.info("CUDA Device Name: %s", torch.cuda.get_device_name(0))
         logging.info("CUDA Version: %s", torch.version.cuda)
       logging.info("=========================\n")
     dist.init_process_group(backend='gloo', init_method='env://',world_size=int(world_size),rank=int(rank))
     run(gpu=False)
     dist.destroy_process_group()

2. 支持使用以下两种方式保存代码数据，供后续模型训练使用。本教程以创建数据集为例。

创建数据集

创建阿里云对象存储（OSS）类型数据集，其中关键参数配置如下，其他参数取默认配置即可。更多详细内容，请参见创建及管理数据集。

参数	描述
数据集名称	自定义数据集名称，本教程配置为：dataset-oss。
属性	选择文件夹。
从阿里云云存储创建	按照以下操作步骤选择代码文件所在的Bucket路径。 单击，选择已创建的Bucket。 单击新建目录，输入目录名称，单击确定。 进入新创建的目录中，单击上传文件，按照界面操作指引上传步骤1中在本地创建的torch_ddp_sample_code.py代码文件。 选中步骤b创建的目录，单击确定。
默认挂载路径	保持默认配置：/mnt/data/。

创建代码集

1. 将步骤1创建的代码文件保存到您的Git地址。

   【说明】您可以将代码文件保存到您自己的Git地址；您也可以使用阿里云云效基础版保存代码文件，该版本免费使用，但有代码库容量限制，如果您使用云效其他版本，会收取费用，具体计费详情请参见计费说明。

2. 创建代码集，其中关键参数配置如下，其他参数取默认配置即可。更多详细内容，请参见代码配置。

   参数	描述
   名称	代码集名称。
   Git地址	您在步骤a中配置的Git地址。

提交训练任务

15

1. 进入新建任务页面。

   1. 登录PAI控制台。

   2. 在页面左上方，选择工作空间所在地域。本教程选择：华南1（深圳）。

   3. 在左侧导航栏单击工作空间列表，在工作空间列表页面中单击默认工作空间名称，进入对应工作空间内。

   4. 在工作空间页面的左侧导航栏选择模型开发与训练>分布式训练（DLC），在分布式训练（DLC）页面，单击新建任务。

2. 配置训练任务的参数。

   1. 在基本信息区域配置以下关键参数，其他参数取默认配置即可，更多详细内容，请参见创建训练任务。

      参数	描述
      任务名称	自定义任务名称，本教程配置为：doc_test_for_dlc。
      节点镜像	目前PAI-DLC支持使用PAI官方镜像和用户自定义镜像。 本教程选择PAI官方镜像：pytorch-training:1.8PAI-gpu-py36-cu101-ubuntu18.04。
      数据集配置	单击添加按钮，配置类型选择按数据集： 数据集：选择已创建的数据集dataset-oss。 挂载路径：配置为/mnt/data/。
      代码配置	本教程无需配置该参数。 【说明】如果您将代码文件保存为代码集，需要配置该参数。
      执行命令	本教程配置为：python /mnt/data/torch_ddp_sample_code.py。 【说明】如果您将代码文件保存为代码集，需要根据实际情况来配置执行命令。

   2. 在资源配置区域，配置以下关键参数。

      参数	描述
      资源配额	本教程选择：公共资源（后付费）。
      框架	本教程选择：PyTorch。
      任务资源	单击，配置资源规格。本教程选择FreeTier > ecs.gn6v-c8g1.2xlarge。阿里云免费试用提供的节点规格包括以下几种规格： ecs.gn7i-c8g1.2xlarge ecs.g6.xlarge ecs.gn6v-c8g1.2xlarge

3. 单击确定，大约需要持续10分钟，任务运行成功。

完成

5

清理及后续

5

清理

• 领取免费资源包后，请在免费额度和有效试用期内使用。如果免费额度用尽或试用期结束后，继续使用计算资源，会产生后付费账单。

  您可前往资源实例管理页面，查看免费额度使用量和过期时间。

• 如果无需继续使用DLC训练任务，您可以删除DLC任务。

  在分布式训练（DLC）页面，当目标任务状态为已成功、已停止或已失败，您可以单击操作列下的删除，来删除目标任务。当目标任务状态为其他状态时，您可以先单击操作列下的停止，再单击操作列下的删除，即可删除任务。

• 对象存储OSS资源清理

  删除对象存储空间，详情请参见删除存储空间。

后续

在试用有效期期间，您还可自行提交DLC训练任务进行模型训练。