分布式训练快速开始

使用Fleet API进行分布式训练

从PaddlePaddle Release 1.5.1 开始，官方推荐使用Fleet API进行分布式训练。

准备条件

  • [x] 成功安装PaddlePaddle，如果尚未安装，请参考 快速开始

  • [x] 学会最基本的单机训练方法，请参考 单机训练 中描述的单机训练，进行学习

点击率预估任务

本文使用一个简单的示例，点击率预估任务，来说明如何使用Fleet API进行分布式训练的配置方法，并利用单机环境模拟分布式环境给出运行示例。

为了方便学习，这里给出的示例是单机与多机混合的代码，用户可以通过不同的启动命令进行单机或多机任务的启动。

  1. from __future__ import print_function
  2. from args import parse_args
  3. import os
  4. import sys
  6. import paddle
  7. import paddle.distributed.fleet.base.role_maker as role_maker
  8. import paddle.distributed.fleet as fleet
  10. from network_conf import ctr_dnn_model_dataset
  12. dense_feature_dim = 13
  13. sparse_feature_dim = 10000001
  15. batch_size = 100
  16. thread_num = 10
  17. embedding_size = 10
  19. args = parse_args()
  21. def main_function(is_local):
  23.     # common code for local training and distributed training
  24.     dense_input = paddle.static.data(
  25.       name="dense_input", shape=[dense_feature_dim], dtype='float32')
  26.     sparse_input_ids = [
  27.           paddle.static.data(name="C" + str(i), shape=[1], lod_level=1,
  28.                             dtype="int64") for i in range(1, 27)]
  30.     label = paddle.static.data(name="label", shape=[1], dtype="int64")
  32.     dataset = paddle.distributed.QueueDataset()
  33.     dataset.init(
  34.           batch_size=batch_size,
  35.           thread_num=thread_num,
  36.           input_type=0,
  37.           pipe_command=python criteo_reader.py %d" % sparse_feature_dim,
  38.           use_var=[dense_input] + sparse_input_ids + [label])
  40.     whole_filelist = ["raw_data/part-%d" % x
  41.                        for x in range(len(os.listdir("raw_data")))]
  42.     dataset.set_filelist(whole_filelist)
  44.     loss, auc_var, batch_auc_var = ctr_dnn_model_dataset(
  45.         dense_input, sparse_input_ids, label, embedding_size,
  46.         sparse_feature_dim)
  48.     exe = paddle.static.Executor(paddle.CPUPlace())
  50.     def train_loop(epoch=20):
  51.         for i in range(epoch):
  52.             exe.train_from_dataset(program=paddle.static.default_main_program(),
  53.                                    dataset=dataset,
  54.                                    fetch_list=[auc_var],
  55.                                    fetch_info=["auc"],
  56.                                    debug=False)
  58.     # local training
  59.     def local_train():
  60.         optimizer = paddle.optimizer.SGD(learning_rate=1e-4)
  61.         optimizer.minimize(loss)
  62.         exe.run(paddle.static.default_startup_program())
  63.         train_loop()
  65.   # distributed training
  66.     def dist_train():
  67.         role = role_maker.PaddleCloudRoleMaker()
  68.         fleet.init(role)
  70.         strategy = paddle.distributed.fleet.DistributedStrategy()
  71.         strategy.a_sync = True
  72.         optimizer = paddle.optimizer.SGD(learning_rate=1e-4)
  73.         optimizer = fleet.distributed_optimizer(optimizer, strategy)
  74.         optimizer.minimize(loss)
  76.         if fleet.is_server():
  77.             fleet.init_server()
  78.             fleet.run_server()
  80.         elif fleet.is_worker():
  81.             fleet.init_worker()
  82.             exe.run(paddle.static.default_startup_program())
  83.             train_loop()
  85.     if is_local:
  86.         local_train()
  87.     else:
  88.         dist_train()
  90. if __name__ == '__main__':
  91.     main_function(args.is_local)
  • 说明：示例中使用的IO方法是dataset，想了解具体的文档和用法请参考 Dataset API 。

单机训练启动命令

  1. python train.py --is_local 1

单机模拟分布式训练的启动命令

在单机模拟多机训练的启动命令，这里我们用到了paddle内置的一个启动器launch_ps，用户可以指定worker和server的数量进行参数服务器任务的启动

  1. python -m paddle.distributed.launch_ps --worker_num 2 --server_num 2 train.py

任务运行的日志在工作目录的logs目录下可以查看，当您能够使用单机模拟分布式训练。