分布式训练快速开始

FleetX 是飞桨分布式训练扩展包，为了可以让用户更快速了解和使用飞桨分布式训练特性，提供了大量分布式训练例子，可以查阅 https://github.com/PaddlePaddle/FleetX/tree/develop/examples，以下章节的例子都可以在这找到，用户也可以直接将仓库下载到本地直接。

一、Collective 训练快速开始

本节将采用CV领域非常经典的模型ResNet50为例，介绍如何使用Fleet API（paddle.distributed.fleet）完成Collective训练任务。 数据方面我们采用Paddle内置的flowers数据集，优化器使用Momentum方法。循环迭代多个epoch，每轮打印当前网络具体的损失值和acc值。 具体代码保存在FleetX/examples/resnet下面， 其中包含动态图和静态图两种执行方式。resnet_dygraph.py为动态图模型相关代码，train_fleet_dygraph.py为动态图训练脚本。 resnet_static.py为静态图模型相关代码，而train_fleet_static.py为静态图训练脚本。

1.1 版本要求

在编写分布式训练程序之前，用户需要确保已经安装paddlepaddle-2.0.0-rc-cpu或paddlepaddle-2.0.0-rc-gpu及以上版本的飞桨开源框架。

1.2 操作方法

与单机单卡的普通模型训练相比，无论静态图还是动态图，Collective训练的代码都只需要补充三个部分代码：

  1. 导入分布式训练需要的依赖包。

  2. 初始化Fleet环境。

  3. 设置分布式训练需要的优化器。

下面将逐一进行讲解。

1.2.1 导入依赖

导入必要的依赖，例如分布式训练专用的Fleet API(paddle.distributed.fleet)。

  1. from paddle.distributed import fleet

1.2.2 初始化fleet环境

包括定义缺省的分布式策略，然后通过将参数is_collective设置为True，使训练架构设定为Collective架构。

  1. strategy = fleet.DistributedStrategy()
  2. fleet.init(is_collective=True, strategy=strategy)

1.2.3 设置分布式训练使用的优化器

使用distributed_optimizer设置分布式训练优化器。

  1. optimizer = fleet.distributed_optimizer(optimizer)

1.3 动态图完整代码

train_fleet_dygraph.py的完整训练代码如下所示。

  1. # -*- coding: UTF-8 -*-
  2. import numpy as np
  3. import argparse
  4. import ast
  5. import paddle
  6. # 导入必要分布式训练的依赖包
  7. from paddle.distributed import fleet
  8. # 导入模型文件
  9. from resnet_dygraph import ResNet
  11. base_lr = 0.1   # 学习率
  12. momentum_rate = 0.9 # 冲量
  13. l2_decay = 1e-4 # 权重衰减
  15. epoch = 10  #训练迭代次数
  16. batch_size = 32 #训练批次大小
  17. class_dim = 102
  19. # 设置数据读取器
  20. def reader_decorator(reader):
  21.     def __reader__():
  22.         for item in reader():
  23.             img = np.array(item[0]).astype('float32').reshape(3, 224, 224)
  24.             label = np.array(item[1]).astype('int64').reshape(1)
  25.             yield img, label
  27.     return __reader__
  29. # 设置优化器
  30. def optimizer_setting(parameter_list=None):
  31.     optimizer = paddle.optimizer.Momentum(
  32.         learning_rate=base_lr,
  33.         momentum=momentum_rate,
  34.         weight_decay=paddle.regularizer.L2Decay(l2_decay),
  35.         parameters=parameter_list)
  36.     return optimizer
  38. # 设置训练函数
  39. def train_resnet():
  40.     # 初始化Fleet环境
  41.     fleet.init(is_collective=True)
  43.     resnet = ResNet(class_dim=class_dim, layers=50)
  45.     optimizer = optimizer_setting(parameter_list=resnet.parameters())
  46.     optimizer = fleet.distributed_optimizer(optimizer)
  47.     # 通过Fleet API获取分布式model，用于支持分布式训练
  48.     resnet = fleet.distributed_model(resnet)
  50.     train_reader = paddle.batch(
  51.             reader_decorator(paddle.dataset.flowers.train(use_xmap=True)),
  52.             batch_size=batch_size,
  53.             drop_last=True)
  55.     train_loader = paddle.io.DataLoader.from_generator(
  56.         capacity=32,
  57.         use_double_buffer=True,
  58.         iterable=True,
  59.         return_list=True,
  60.         use_multiprocess=True)
  61.     train_loader.set_sample_list_generator(train_reader)
  63.     for eop in range(epoch):
  64.         resnet.train()
  66.         for batch_id, data in enumerate(train_loader()):
  67.             img, label = data
  68.             label.stop_gradient = True
  70.             out = resnet(img)
  71.             loss = paddle.nn.functional.cross_entropy(input=out, label=label)
  72.             avg_loss = paddle.mean(x=loss)
  73.             acc_top1 = paddle.metric.accuracy(input=out, label=label, k=1)
  74.             acc_top5 = paddle.metric.accuracy(input=out, label=label, k=5)
  76.             dy_out = avg_loss.numpy()
  78.             avg_loss.backward()
  80.             optimizer.minimize(avg_loss)
  81.             resnet.clear_gradients()
  82.             if batch_id % 5 == 0:
  83.                 print("[Epoch %d, batch %d] loss: %.5f, acc1: %.5f, acc5: %.5f" % (eop, batch_id, dy_out, acc_top1, acc_top5))
  84. # 启动训练
  85. if __name__ == '__main__':
  86.     train_resnet()

1.4 静态图完整代码

train_fleet_static.py的完整训练代码如下所示。

  1. # -*- coding: UTF-8 -*-
  2. import numpy as np
  3. import argparse
  4. import ast
  5. import paddle
  6. # 导入必要分布式训练的依赖包
  7. import paddle.distributed.fleet as fleet
  8. # 导入模型文件
  9. import resnet_static as resnet
  10. import os
  12. base_lr = 0.1   # 学习率
  13. momentum_rate = 0.9 # 冲量
  14. l2_decay = 1e-4 # 权重衰减
  16. epoch = 10  #训练迭代次数
  17. batch_size = 32 #训练批次大小
  18. class_dim = 10
  20. # 设置优化器
  21. def optimizer_setting(parameter_list=None):
  22.     optimizer = paddle.optimizer.Momentum(
  23.         learning_rate=base_lr,
  24.         momentum=momentum_rate,
  25.         weight_decay=paddle.regularizer.L2Decay(l2_decay),
  26.         parameters=parameter_list)
  27.     return optimizer
  28. # 设置数据读取器
  29. def get_train_loader(feed_list, place):
  30.     def reader_decorator(reader):
  31.         def __reader__():
  32.             for item in reader():
  33.                 img = np.array(item[0]).astype('float32').reshape(3, 224, 224)
  34.                 label = np.array(item[1]).astype('int64').reshape(1)
  35.                 yield img, label
  37.         return __reader__
  38.     train_reader = paddle.batch(
  39.             reader_decorator(paddle.dataset.flowers.train(use_xmap=True)),
  40.             batch_size=batch_size,
  41.             drop_last=True)
  42.     train_loader = paddle.io.DataLoader.from_generator(
  43.         capacity=32,
  44.         use_double_buffer=True,
  45.         feed_list=feed_list,
  46.         iterable=True)
  47.     train_loader.set_sample_list_generator(train_reader, place)
  48.     return train_loader
  49. # 设置训练函数
  50. def train_resnet():
  51.     print("Start collective training example:")
  52.     paddle.enable_static() # 使能静态图功能
  53.     paddle.vision.set_image_backend('cv2')
  55.     image = paddle.static.data(name="x", shape=[None, 3, 224, 224], dtype='float32')
  56.     label= paddle.static.data(name="y", shape=[None, 1], dtype='int64')
  57.     # 调用ResNet50模型
  58.     model = resnet.ResNet(layers=50)
  59.     out = model.net(input=image, class_dim=class_dim)
  60.     avg_cost = paddle.nn.functional.cross_entropy(input=out, label=label)
  61.     acc_top1 = paddle.metric.accuracy(input=out, label=label, k=1)
  62.     acc_top5 = paddle.metric.accuracy(input=out, label=label, k=5)
  63.     # 设置训练资源，本例使用GPU资源
  64.     place = paddle.CUDAPlace(int(os.environ.get('FLAGS_selected_gpus', 0)))
  65.     print("Run on {}.".format(place))
  67.     train_loader = get_train_loader([image, label], place)
  68.     #初始化Fleet环境
  69.     strategy = fleet.DistributedStrategy()
  70.     fleet.init(is_collective=True, strategy=strategy)
  71.     optimizer = optimizer_setting()
  73.     # 通过Fleet API获取分布式优化器，将参数传入飞桨的基础优化器
  74.     optimizer = fleet.distributed_optimizer(optimizer)
  75.     optimizer.minimize(avg_cost)
  77.     exe = paddle.static.Executor(place)
  78.     print("Execute startup program.")
  79.     exe.run(paddle.static.default_startup_program())
  81.     epoch = 10
  82.     step = 0
  83.     for eop in range(epoch):
  84.         for batch_id, data in enumerate(train_loader()):
  85.             loss, acc1, acc5 = exe.run(paddle.static.default_main_program(), feed=data, fetch_list=[avg_cost.name, acc_top1.name, acc_top5.name])
  86.             if batch_id % 5 == 0:
  87.                 print("[Epoch %d, batch %d] loss: %.5f, acc1: %.5f, acc5: %.5f" % (eop, batch_id, loss, acc1, acc5))
  88. # 启动训练
  89. if __name__ == '__main__':
  90.     train_resnet()

1.5 运行示例

假设要运行2卡的任务，那么只需在命令行中执行:

动态图：

  1. python3 -m paddle.distributed.launch --gpus=0,1 train_fleet_dygraph.py

您将看到显示如下日志信息：

  1. -----------  Configuration Arguments -----------
  2. gpus: 0,1
  3. heter_worker_num: None
  4. heter_workers:
  5. http_port: None
  6. ips: 127.0.0.1
  7. log_dir: log
  8. nproc_per_node: None
  9. server_num: None
  10. servers:
  11. training_script: train_fleet_dygraph.py
  12. training_script_args: []
  13. worker_num: None
  14. workers:
  15. ------------------------------------------------
  16. WARNING 2021-05-06 11:32:50,804 launch.py:316] Not found distinct arguments and compiled with cuda. Default use collective mode
  17. launch train in GPU mode
  18. INFO 2021-05-06 11:32:50,806 launch_utils.py:472] Local start 2 processes. First process distributed environment info (Only For Debug):
  19.     +=======================================================================================+
  20.     |                        Distributed Envs                      Value                    |
  21.     +---------------------------------------------------------------------------------------+
  22.     |                PADDLE_TRAINER_ENDPOINTS         127.0.0.1:20923,127.0.0.1:10037       |
  23.     |                     FLAGS_selected_gpus                        0                      |
  24.     |                       PADDLE_TRAINER_ID                        0                      |
  25.     |                     PADDLE_TRAINERS_NUM                        2                      |
  26.     |                 PADDLE_CURRENT_ENDPOINT                 127.0.0.1:20923               |
  27.     +=======================================================================================+
  29. INFO 2021-05-06 11:32:50,806 launch_utils.py:475] details abouts PADDLE_TRAINER_ENDPOINTS can be found in log/endpoints.log, and detail running logs maybe found in log/workerlog.0
  30. grep: warning: GREP_OPTIONS is deprecated; please use an alias or script
  31. I0506 11:32:51.828132  6427 nccl_context.cc:189] init nccl context nranks: 2 local rank: 0 gpu id: 0 ring id: 0
  32. W0506 11:32:52.365190  6427 device_context.cc:362] Please NOTE: device: 0, GPU Compute Capability: 7.0, Driver API Version: 11.0, Runtime API Version: 11.0
  33. W0506 11:32:52.368203  6427 device_context.cc:372] device: 0, cuDNN Version: 8.0.
  34. [Epoch 0, batch 0] loss: 4.98047, acc1: 0.00000, acc5: 0.00000
  35. [Epoch 0, batch 5] loss: 39.06348, acc1: 0.03125, acc5: 0.09375
  36. ...

静态图：

  1. python3 -m paddle.distributed.launch --gpus=0,1 train_fleet_static.py

您将看到显示如下日志信息：

  1. -----------  Configuration Arguments -----------
  2. gpus: 0,1
  3. heter_worker_num: None
  4. heter_workers:
  5. http_port: None
  6. ips: 127.0.0.1
  7. log_dir: log
  8. nproc_per_node: None
  9. server_num: None
  10. servers:
  11. training_script: train_fleet_static.py
  12. training_script_args: []
  13. worker_num: None
  14. workers:
  15. ------------------------------------------------
  16. WARNING 2021-05-06 11:36:30,019 launch.py:316] Not found distinct arguments and compiled with cuda. Default use collective mode
  17. launch train in GPU mode
  18. INFO 2021-05-06 11:36:30,021 launch_utils.py:472] Local start 2 processes. First process distributed environment info (Only For Debug):
  19.     +=======================================================================================+
  20.     |                        Distributed Envs                      Value                    |
  21.     +---------------------------------------------------------------------------------------+
  22.     |                       PADDLE_TRAINER_ID                        0                      |
  23.     |                 PADDLE_CURRENT_ENDPOINT                 127.0.0.1:10039               |
  24.     |                PADDLE_TRAINER_ENDPOINTS         127.0.0.1:10039,127.0.0.1:31719       |
  25.     |                     PADDLE_TRAINERS_NUM                        2                      |
  26.     |                     FLAGS_selected_gpus                        0                      |
  27.     +=======================================================================================+
  29. INFO 2021-05-06 11:36:30,021 launch_utils.py:475] details abouts PADDLE_TRAINER_ENDPOINTS can be found in log/endpoints.log, and detail running logs maybe found in log/workerlog.0
  30. grep: warning: GREP_OPTIONS is deprecated; please use an alias or script
  31. Start collective training example:
  32. Run on CUDAPlace(0).
  33. server not ready, wait 3 sec to retry...
  34. not ready endpoints:['127.0.0.1:31719']
  35. Execute startup program.
  36. W0506 11:36:35.667778  6697 device_context.cc:362] Please NOTE: device: 0, GPU Compute Capability: 7.0, Driver API Version: 11.0, Runtime API Version: 11.0
  37. W0506 11:36:35.671609  6697 device_context.cc:372] device: 0, cuDNN Version: 8.0.
  38. Start training:
  39. W0506 11:36:39.900507  6697 fuse_all_reduce_op_pass.cc:79] Find all_reduce operators: 161. To make the speed faster, some all_reduce ops are fused during training, after fusion, the number of all_reduce ops is 5.
  40. [Epoch 0, batch 0] loss: 4.67622, acc1: 0.00000, acc5: 0.09375
  41. [Epoch 0, batch 5] loss: 30.24010, acc1: 0.00000, acc5: 0.06250
  42. ...

从单机多卡到多机多卡训练，在代码上不需要做任何改动，只需再额外指定ips参数即可。其内容为多机的ip列表，命令如下所示：

  1. # 动态图
  2. python3 -m paddle.distributed.launch --ips="xx.xx.xx.xx,yy.yy.yy.yy" --gpus 0,1,2,3,4,5,6,7 train_fleet_dygraph.py
  4. # 静态图
  5. python3 -m paddle.distributed.launch --ips="xx.xx.xx.xx,yy.yy.yy.yy" --gpus 0,1,2,3,4,5,6,7 train_fleet_static.py

二、ParameterServer训练快速开始

本节将采用推荐领域非常经典的模型wide_and_deep为例，介绍如何使用Fleet API（paddle.distributed.fleet）完成参数服务器训练任务，本次快速开始的完整示例代码位于 https://github.com/PaddlePaddle/FleetX/tree/develop/examples/wide_and_deep。

2.1 版本要求

在编写分布式训练程序之前，用户需要确保已经安装paddlepaddle-2.0.0-rc-cpu或paddlepaddle-2.0.0-rc-gpu及以上版本的飞桨开源框架。

2.2 操作方法

参数服务器训练的基本代码主要包括如下几个部分：

  1. 导入分布式训练需要的依赖包。

  2. 定义分布式模式并初始化分布式训练环境。

  3. 加载模型及数据。

  4. 定义参数更新策略及优化器。

  5. 开始训练。

下面将逐一进行讲解。

2.2.1 导入依赖

导入必要的依赖，例如分布式训练专用的Fleet API(paddle.distributed.fleet)。

  1. import paddle
  2. import paddle.distributed.fleet as fleet

2.2.2 定义分布式模式并初始化分布式训练环境

通过 fleet.init() 接口，用户可以定义训练相关的环境，注意此环境是用户预先在环境变量中配置好的，包括：训练节点个数，服务节点个数，当前节点的序号，服务节点完整的IP:PORT列表等。

  1. # 当前参数服务器模式只支持静态图模式， 因此训练前必须指定 ``paddle.enable_static()``
  2. paddle.enable_static()
  3. fleet.init(is_collective=False)

2.2.3 加载模型及数据

  1. # 模型定义参考 examples/wide_and_deep 中 model.py
  2. from model import WideDeepModel
  3. from reader import WideDeepDataset
  5. model = WideDeepModel()
  6. model.net(is_train=True)
  8. def distributed_training(exe, train_model, train_data_path="./data", batch_size=10, epoch_num=1):
  9.     train_data = WideDeepDataset(data_path=train_data_path)
  10.     reader = train_model.loader.set_sample_generator(
  11.         train_data, batch_size=batch_size, drop_last=True, places=paddle.CPUPlace())
  13.     for epoch_id in range(epoch_num):
  14.         reader.start()
  15.         try:
  16.             while True:
  17.                 loss_val = exe.run(program=paddle.static.default_main_program(),
  18.                                 fetch_list=[train_model.cost.name])
  19.                 loss_val = np.mean(loss_val)
  20.                 print("TRAIN ---> pass: {} loss: {}n".format(epoch_id, loss_val))
  21.         except paddle.common_ops_import.core.EOFException:
  22.             reader.reset()

2.2.4 定义同步训练 Strategy 及 Optimizer

在Fleet API中，用户可以使用 fleet.DistributedStrategy() 接口定义自己想要使用的分布式策略。

其中 a_sync 选项用于定义参数服务器相关的策略，当其被设定为 False 时，分布式训练将在同步的模式下进行。反之，当其被设定成 True 时，分布式训练将在异步的模式下进行。

  1. # 定义异步训练
  2. dist_strategy = fleet.DistributedStrategy()
  3. dist_strategy.a_sync = True
  5. # 定义同步训练
  6. dist_strategy = fleet.DistributedStrategy()
  7. dist_strategy.a_sync = False
  9. # 定义Geo异步训练, Geo异步目前只支持SGD优化算法
  10. dist_strategy = fleet.DistributedStrategy()
  11. dist_strategy.a_sync = True
  12. dist_strategy.a_sync_configs = {"k_steps": 100}
  14. optimizer = paddle.optimizer.SGD(learning_rate=0.0001)
  15. optimizer = fleet.distributed_optimizer(optimizer, dist_strategy)
  16. optimizer.minimize(model.loss)

2.2.5 开始训练

完成模型及训练策略以后，我们就可以开始训练模型了。因为在参数服务器模式下会有不同的角色，所以根据不同节点分配不同的任务。

对于服务器节点，首先用 init_server() 接口对其进行初始化，然后启动服务并开始监听由训练节点传来的梯度。

同样对于训练节点，用 init_worker() 接口进行初始化后， 开始执行训练任务。运行 exe.run() 接口开始训练，并得到训练中每一步的损失值。

  1. if fleet.is_server():
  2.     fleet.init_server()
  3.     fleet.run_server()
  4. else:
  5.     exe = paddle.static.Executor(paddle.CPUPlace())
  6.     exe.run(paddle.static.default_startup_program())
  8.     fleet.init_worker()
  10.     distributed_training(exe, model)
  12.     fleet.stop_worker()

2.3 运行训练脚本

定义完训练脚本后，我们就可以用 python3 -m paddle.distributed.launch 指令运行分布式任务了。其中 server_num , worker_num 分别为服务节点和训练节点的数量。在本例中，服务节点有1个，训练节点有2个。

  1. python3 -m paddle.distributed.launch --server_num=1 --worker_num=2 --gpus=0,1 train.py

您将看到显示如下日志信息：

  1. -----------  Configuration Arguments -----------
  2. gpus: 0,1
  3. heter_worker_num: None
  4. heter_workers:
  5. http_port: None
  6. ips: 127.0.0.1
  7. log_dir: log
  8. nproc_per_node: None
  9. server_num: 1
  10. servers:
  11. training_script: train.py
  12. training_script_args: []
  13. worker_num: 2
  14. workers:
  15. ------------------------------------------------
  16. INFO 2021-05-06 12:14:26,890 launch.py:298] Run parameter-sever mode. pserver arguments:['--worker_num', '--server_num'], cuda count:8
  17. INFO 2021-05-06 12:14:26,892 launch_utils.py:973] Local server start 1 processes. First process distributed environment info (Only For Debug):
  18.     +=======================================================================================+
  19.     |                        Distributed Envs                      Value                    |
  20.     +---------------------------------------------------------------------------------------+
  21.     |                     PADDLE_TRAINERS_NUM                        2                      |
  22.     |                           TRAINING_ROLE                     PSERVER                   |
  23.     |                                  POD_IP                    127.0.0.1                  |
  24.     |                  PADDLE_GLOO_RENDEZVOUS                        3                      |
  25.     |            PADDLE_PSERVERS_IP_PORT_LIST                 127.0.0.1:34008               |
  26.     |                             PADDLE_PORT                      34008                    |
  27.     |                        PADDLE_WITH_GLOO                        0                      |
  28.     |       PADDLE_HETER_TRAINER_IP_PORT_LIST                                               |
  29.     |                PADDLE_TRAINER_ENDPOINTS         127.0.0.1:18913,127.0.0.1:10025       |
  30.     |               PADDLE_GLOO_HTTP_ENDPOINT                 127.0.0.1:23053               |
  31.     |                     PADDLE_GLOO_FS_PATH                /tmp/tmp8vqb8arq               |
  32.     +=======================================================================================+
  34. INFO 2021-05-06 12:14:26,902 launch_utils.py:1041] Local worker start 2 processes. First process distributed environment info (Only For Debug):
  35.     +=======================================================================================+
  36.     |                        Distributed Envs                      Value                    |
  37.     +---------------------------------------------------------------------------------------+
  38.     |               PADDLE_GLOO_HTTP_ENDPOINT                 127.0.0.1:23053               |
  39.     |                  PADDLE_GLOO_RENDEZVOUS                        3                      |
  40.     |            PADDLE_PSERVERS_IP_PORT_LIST                 127.0.0.1:34008               |
  41.     |                        PADDLE_WITH_GLOO                        0                      |
  42.     |                PADDLE_TRAINER_ENDPOINTS         127.0.0.1:18913,127.0.0.1:10025       |
  43.     |                     FLAGS_selected_gpus                        0                      |
  44.     |                     PADDLE_GLOO_FS_PATH                /tmp/tmp8vqb8arq               |
  45.     |                     PADDLE_TRAINERS_NUM                        2                      |
  46.     |                           TRAINING_ROLE                     TRAINER                   |
  47.     |                     XPU_VISIBLE_DEVICES                        0                      |
  48.     |       PADDLE_HETER_TRAINER_IP_PORT_LIST                                               |
  49.     |                       PADDLE_TRAINER_ID                        0                      |
  50.     |                    CUDA_VISIBLE_DEVICES                        0                      |
  51.     |                     FLAGS_selected_xpus                        0                      |
  52.     +=======================================================================================+
  54. INFO 2021-05-06 12:14:26,921 launch_utils.py:903] Please check servers, workers and heter_worker logs in log/workerlog.*, log/serverlog.* and log/heterlog.*
  55. INFO 2021-05-06 12:14:33,446 launch_utils.py:914] all workers exit, going to finish parameter server and heter_worker.
  56. INFO 2021-05-06 12:14:33,446 launch_utils.py:926] all parameter server are killed