• 使用FleetAPI进行分布式训练
    • FleetAPI 设计说明
    • FleetAPI 接口说明
    • FleetAPI 一般训练步骤
      • 通过import引入需要使用的模式
      • 初始化
      • 分布式策略及多机配置
      • 具体训练流程

    使用FleetAPI进行分布式训练

    FleetAPI 设计说明

    Fleet是PaddlePaddle Fluid最新优化的多机API版本， 统一了多机API的实现，兼容Transpiler/Collective两种模式。 可以在MPI环境及K8S环境下进行多机训练，以及自定义分布式训练配置。

    FleetAPI 接口说明

    接口说明
    initfleet初始化，需要在使用fleet其他接口前先调用，用于定义多机的环境配置
    distributedoptimizerfleet多机训练策略优化，接收一个标准Optimizer及相应的多机运行策略，fleet会根据优化策略进行优化
    init_serverfleet加载model_dir中保存的模型相关参数进行parameter server的初始化
    run_serverfleet启动parameter server服务
    init_workerfleet初始化当前worker运行环境
    is_worker判断当前节点是否是Worker节点，是则返回True，否则返回False
    is_server判断当前节点是否是Server节点，是则返回True，否则返回False
    save_inference_modelfleet保存预测相关的模型及参数，参数及用法参考 code:_fluid.io.save_inference_model
    savepersistablesfleet保存多机模型参数，参数及用法参考 code:_fluid.io.save_persistables

    FleetAPI 一般训练步骤

    通过import引入需要使用的模式

    使用parameter server方式的训练：

    1. from paddle.fluid.incubate.fleet.parameter_server.distribute_transpiler import fleet

    初始化

    Fleet使用 code:fleet.init(role_maker=None) 进行初始化

    当用户不指定role_maker, 则Fleet默认用户使用MPI环境，会采用MPISymetricRoleMaker.

    如果用户使用非MPI环境，则需要通过UserDefinedRoleMaker自行定义执行环境。例如：

    1. role = UserDefinedRoleMaker(current_id=0,
    2.                  role=Role.WORKER,
    3.                  worker_num=3,
    4.                  server_endpoints=["127.0.0.1:6001","127.0.0.1:6002"])
    5. fleet.init(role_maker=role)

    分布式策略及多机配置

    对于Transpiler模式，需要使用 DistributeTranspilerConfig 指定多机配置。Fleet需要在用户定义的optimizer之上装饰 code:fleet.distributed_optimizer 来完成多机分布式策略的配置。

    接口说明
    syncmodeFleet可以支持同步训练或异步训练， 默认会生成同步训练的分布式程序，通过指定 sync_mode=False 参数即可生成异步训练的程序
    split_method指定参数在parameter server上的分布方式, 默认使用 _RoundRobin, 也可选HashName
    slice_var_up指定是否将较大（大于8192个元素）的参数切分到多个parameter server以均衡计算负载，默认为开启

    例如：

    1. config = DistributeTranspilerConfig()
    2. config.sync_mode = True
    3.  
    4. # build network
    5. # ...
    6. avg_cost = model()
    7.  
    8. optimizer = fluid.optimizer.Adam(learning_rate=0.001)
    9. # 加入 fleet distributed_optimizer 加入分布式策略配置及多机优化
    10. optimizer = fleet.distributed_optimizer(optimizer, config)
    11. optimizer.minimize(avg_cost)

    具体训练流程

    1. # 启动server
    2. if fleet.is_server():
    3.     fleet.init_server()
    4.     fleet.run_server()
    5.  
    6. # 启动worker
    7. if fleet.is_worker():
    8.     # 初始化worker配置
    9.     fleet.init_worker()
    10.  
    11.     feeder = fluid.DataFeeder(place=place, feed_list=[x, y])
    12.     train_reader = paddle.batch(fake_reader(), batch_size=24)
    13.  
    14.     exe.run(fleet.startup_program)
    15.  
    16.     PASS_NUM = 10
    17.     for pass_id in range(PASS_NUM):
    18.         for batch_id, data in enumerate(train_reader()):
    19.             avg_loss_value, auc_value, auc_batch_value = \
    20.                 exe.run(fleet.main_program, feed=feeder.feed(data), fetch_list=[avg_cost, auc, auc_batch])
    21.             print("Pass %d, cost = %f, auc = %f, batch_auc = %f" % (pass_id, avg_loss_value, auc_value, auc_batch_value))
    22.     # 通知server，当前节点训练结束
    23.     fleet.stop_worker()