Fleet

class paddle.distributed.fleet. Fleet [源代码]

Fleet是飞桨分布式训练统一API, 只需要import fleet并简单初始化后即可快速开始使用飞桨大规模分布式训练

init ( role_maker=None, is_collective=False, strategy=None )

使用RoleMaker或其他配置初始化fleet。

参数：

  • role_maker (RoleMakerBase) 已初始化好的PaddleCloudRoleMaker或UserDefineRoleMaker

  • is_collective (bool) 在未指定role_maker的情况下,可由init方法自行初始化RoleMaker, is_collective为True则按照collective模式进行创建， is_collective=False则按照ParameterServer模式进行创建

  • strategy (DistributedStrategy): 分布式训练的额外属性。详情请参阅paddle.distributed.fleet.DistributedStrategy。默认值：None。

返回：None

代码示例1

  1. import paddle.distributed.fleet as fleet
  2. fleet.init()

代码示例2

  1. import paddle.distributed.fleet as fleet
  2. fleet.init(is_collective=True)

代码示例3

  1. import paddle.distributed.fleet as fleet
  2. role = fleet.PaddleCloudRoleMaker()
  3. fleet.init(role)

代码示例4

  1. import paddle.distributed.fleet as fleet
  2. strategy = fleet.DistributedStrategy()
  3. fleet.init(strategy=strategy)

is_first_worker ( )

返回当前节点是否为第一个worker节点, 判断当前worker_index是否为0， 如果为0则返回True，否则返回False

返回：True/False

代码示例

  1. import paddle.distributed.fleet as fleet
  2. fleet.init()
  3. fleet.is_first_worker()

worker_index ( )

返回当前节点的编号, 每个worker节点被分配[0, worker_num-1]内的唯一的编码ID

返回：int

代码示例

  1. import paddle.distributed.fleet as fleet
  2. fleet.init()
  3. fleet.worker_index()

worker_num ( )

返回当前全部训练节点中workjer节点的个数

返回：int

代码示例

  1. import paddle.distributed.fleet as fleet
  2. fleet.init()
  3. fleet.worker_num()

is_worker ( )

返回当前节点是否为worker节点

返回：True/False

代码示例

  1. import paddle.distributed.fleet as fleet
  2. fleet.init()
  3. fleet.is_worker()

worker_endpoints ( to_string=False )

返回全部worker节点的ip及端口信息

返回：list/string

代码示例

  1. import paddle.distributed.fleet as fleet
  2. fleet.init()
  3. fleet.worker_endpoints()

server_num ( )

注意：

该参数只在ParameterServer模式下生效

返回当前全部Server节点的个数

返回：int

代码示例

  1. import paddle.distributed.fleet as fleet
  2. fleet.init()
  3. fleet.server_num()

server_index ( )

注意：

该参数只在ParameterServer模式下生效

返回当前节点的编号, 每个server节点被分配[0, server_num-1]内的唯一的编码ID

返回：int

代码示例

  1. import paddle.distributed.fleet as fleet
  2. fleet.init()
  3. fleet.server_index()

server_endpoints ( to_string=False )

注意：

该参数只在ParameterServer模式下生效

返回全部server节点的ip及端口信息

返回：list/string

代码示例

  1. import paddle.distributed.fleet as fleet
  2. fleet.init()
  3. fleet.server_endpoints()

is_server ( )

注意：

该参数只在ParameterServer模式下生效

返回当前节点是否为server节点

返回：True/False

代码示例

  1. import paddle.distributed.fleet as fleet
  2. fleet.init()
  3. fleet.is_server()

barrier_worker ( )

调用集合通信功能，强制要求所有的worker在此处相互等待一次

返回：无

代码示例

  1. import paddle.distributed.fleet as fleet
  2. fleet.init()
  3. fleet.barrier_worker()

init_worker ( )

worker节点在训练前的初始化, 包括通信模块， 参数同步等

返回：无

代码示例

  1. import paddle.distributed.fleet as fleet
  2. fleet.init()
  3. fleet.init_worker()

init_server ( args, kwargs* )

server节点的初始化, 包括server端参数初始化，模型加载等

返回：无

代码示例

  1. import paddle.distributed.fleet as fleet
  2. fleet.init()
  3. fleet.init_server()

run_server ( )

server节点的运行, 此命令会将ParameterServer的进程启动并常驻直至训练结束

返回：无

代码示例

  1. import paddle.distributed.fleet as fleet
  2. fleet.init()
  3. fleet.init_server()
  4. fleet.run_server()

stop_worker ( )

停止当前正在运行的worker节点

返回：无

代码示例

  1. import paddle.distributed.fleet as fleet
  2. fleet.init()
  3. fleet.init_worker()
  4. "..."
  5. fleet.stop_worker()

save_inference_model ( executor, dirname, feeded_var_names, target_vars, main_program=None, export_for_deployment=True )

修剪指定的 main_program 以构建一个专门用于预测的 Inference Program （ Program 含义详见 基础概念 ）。 所得到的 Inference Program 及其对应的所>有相关参数均被保存到 dirname 指定的目录中。

参数:

  • executor (Executor) – 用于保存预测模型的 executor ，详见 执行引擎 。

  • dirname (str) – 指定保存预测模型结构和参数的文件目录。

  • feeded_var_names (list[str]) – 字符串列表，包含着Inference Program预测时所需提供数据的所有变量名称（即所有输入变量的名称）。

  • target_vars (list[Tensor]) – Tensor （详见 基础概念 ）类型列表，包含着模型的所有输出变量。通过这些输出变量即可得到模型的预测结果。

  • main_program (Program，可选) – 通过该参数指定的 main_program 可构建一个专门用于预测的 Inference Program 。 若为None, 则使用全局默认的 _main_program_ 。>默认值为None。

  • export_for_deployment (bool，可选) – 若为True，则 main_program 指定的Program将被修改为只支持直接预测部署的Program。否则，将存储更多的信息，方便优化和再训练。目前

只支持设置为True，且默认值为True。

返回：无

代码示例

  1. import paddle
  2. paddle.enable_static()
  3. import paddle.distributed.fleet as fleet
  5. fleet.init()
  7. # build net
  8. # loss = Net()
  9. # fleet.distributed_optimizer(...)
  11. exe = paddle.static.Executor(paddle.CPUPlace())
  12. fleet.save_inference_model(exe, "dirname", ["feed_varname"], [loss], paddle.static.default_main_program())

save_persistables ( executor, dirname, main_program=None )

保存全量模型参数

参数:

  • executor (Executor) – 用于保存持久性变量的 executor ，详见 执行引擎 。

  • dirname (str) – 用于储存持久性变量的文件目录。

  • main_program (Program，可选) – 需要保存持久性变量的Program（ Program 含义详见 基础概念 ）。如果为None，则使用default_main_Program 。默认值为None>。

返回：无

代码示例

  1. import paddle
  2. paddle.enable_static()
  3. import paddle.distributed.fleet as fleet
  5. fleet.init()
  7. # build net
  8. # fleet.distributed_optimizer(...)
  10. exe = paddle.static.Executor(paddle.CPUPlace())
  11. fleet.save_persistables(exe, "dirname", paddle.static.default_main_program())

distributed_optimizer ( optimizer, strategy=None )

基于分布式布式并行策略进行模型的拆分及优化。

参数:

  • optimizer (optimizer) – paddle定义的优化器。

  • strategy (DistributedStrategy) – 分布式优化器的额外属性。建议在fleet.init()创建。这里的仅仅是为了兼容性。如果这里的参数strategy不是None，则它将覆盖在fleet.init()创建的DistributedStrategy，并在后续的分布式训练中生效。

代码示例

  1. import paddle
  2. paddle.enable_static()
  3. import paddle.distributed.fleet as fleet
  4. fleet.init(is_collective=True)
  5. strategy = fleet.DistributedStrategy()
  6. optimizer = paddle.optimizer.SGD(learning_rate=0.001)
  7. optimizer = fleet.distributed_optimizer(optimizer, strategy=strategy)

distributed_model ( model )

注意：

1. 该API只在 Dygraph 模式下生效

返回分布式数据并行模型。

参数：

model (Layer) - 用户定义的模型，此处模型是指继承动态图Layer的网络。

返回：分布式数据并行模型，该模型同样继承动态图Layer。

代码示例

  1. # 这个示例需要由fleetrun启动, 用法为:
  2. # fleetrun --gpus=0,1 example.py
  3. # 脚本example.py中的代码是下面这个示例.
  5. import paddle
  6. import paddle.nn as nn
  7. from paddle.distributed import fleet
  9. class LinearNet(nn.Layer):
  10.     def __init__(self):
  11.         super(LinearNet, self).__init__()
  12.         self._linear1 = nn.Linear(10, 10)
  13.         self._linear2 = nn.Linear(10, 1)
  15.     def forward(self, x):
  16.         return self._linear2(self._linear1(x))
  18. # 1. initialize fleet environment
  19. fleet.init(is_collective=True)
  21. # 2. create layer & optimizer
  22. layer = LinearNet()
  23. loss_fn = nn.MSELoss()
  24. adam = paddle.optimizer.Adam(
  25.     learning_rate=0.001, parameters=layer.parameters())
  27. # 3. get data_parallel model using fleet
  28. adam = fleet.distributed_optimizer(adam)
  29. dp_layer = fleet.distributed_model(layer)
  31. # 4. run layer
  32. inputs = paddle.randn([10, 10], 'float32')
  33. outputs = dp_layer(inputs)
  34. labels = paddle.randn([10, 1], 'float32')
  35. loss = loss_fn(outputs, labels)
  37. print("loss:", loss.numpy())
  39. loss.backward()
  41. adam.step()
  42. adam.clear_grad()

state_dict ( )

注意：

1. 该API只在 Dygraph 模式下生效

以 dict 返回当前 optimizer 使用的所有Tensor 。比如对于Adam优化器，将返回 beta1, beta2, momentum 等Tensor。

返回：dict, 当前 optimizer 使用的所有Tensor。

代码示例

  1. # 这个示例需要由fleetrun启动, 用法为:
  2. # fleetrun --gpus=0,1 example.py
  3. # 脚本example.py中的代码是下面这个示例.
  5. import numpy as np
  6. import paddle
  7. from paddle.distributed import fleet
  9. fleet.init(is_collective=True)
  11. value = np.arange(26).reshape(2, 13).astype("float32")
  12. a = paddle.to_tensor(value)
  14. layer = paddle.nn.Linear(13, 5)
  15. adam = paddle.optimizer.Adam(learning_rate=0.01, parameters=layer.parameters())
  17. adam = fleet.distributed_optimizer(adam)
  18. dp_layer = fleet.distributed_model(layer)
  19. state_dict = adam.state_dict()

set_state_dict ( state_dict )

注意：

1. 该API只在 Dygraph 模式下生效

加载 optimizer 的Tensor字典给当前 optimizer 。

返回：None

代码示例

  1. # 这个示例需要由fleetrun启动, 用法为:
  2. # fleetrun --gpus=0,1 example.py
  3. # 脚本example.py中的代码是下面这个示例.
  5. import numpy as np
  6. import paddle
  7. from paddle.distributed import fleet
  9. fleet.init(is_collective=True)
  11. value = np.arange(26).reshape(2, 13).astype("float32")
  12. a = paddle.to_tensor(value)
  14. layer = paddle.nn.Linear(13, 5)
  15. adam = paddle.optimizer.Adam(learning_rate=0.01, parameters=layer.parameters())
  17. adam = fleet.distributed_optimizer(adam)
  18. dp_layer = fleet.distributed_model(layer)
  19. state_dict = adam.state_dict()
  20. paddle.save(state_dict, "paddle_dy")
  21. para_state_dict = paddle.load( "paddle_dy")
  22. adam.set_state_dict(para_state_dict)

set_lr ( value )

注意：

1. 该API只在 Dygraph 模式下生效

手动设置当前 optimizer 的学习率。

参数：

value (float) - 需要设置的学习率的值。

返回：None

代码示例

  1. # 这个示例需要由fleetrun启动, 用法为:
  2. # fleetrun --gpus=0,1 example.py
  3. # 脚本example.py中的代码是下面这个示例.
  5. import numpy as np
  6. import paddle
  7. from paddle.distributed import fleet
  9. fleet.init(is_collective=True)
  11. value = np.arange(26).reshape(2, 13).astype("float32")
  12. a = paddle.to_tensor(value)
  14. layer = paddle.nn.Linear(13, 5)
  15. adam = paddle.optimizer.Adam(learning_rate=0.01, parameters=layer.parameters())
  17. adam = fleet.distributed_optimizer(adam)
  18. dp_layer = fleet.distributed_model(layer)
  20. lr_list = [0.2, 0.3, 0.4, 0.5, 0.6]
  21. for i in range(5):
  22.     adam.set_lr(lr_list[i])
  23.     lr = adam.get_lr()
  24.     print("current lr is {}".format(lr))
  25. # Print:
  26. #    current lr is 0.2
  27. #    current lr is 0.3
  28. #    current lr is 0.4
  29. #    current lr is 0.5
  30. #    current lr is 0.6

get_lr ( )

注意：

1. 该API只在 Dygraph 模式下生效

获取当前步骤的学习率。

返回：float，当前步骤的学习率。

代码示例

  1. # 这个示例需要由fleetrun启动, 用法为:
  2. # fleetrun --gpus=0,1 example.py
  3. # 脚本example.py中的代码是下面这个示例.
  5. import numpy as np
  6. import paddle
  7. from paddle.distributed import fleet
  9. fleet.init(is_collective=True)
  11. value = np.arange(26).reshape(2, 13).astype("float32")
  12. a = paddle.to_tensor(value)
  14. layer = paddle.nn.Linear(13, 5)
  15. adam = paddle.optimizer.Adam(learning_rate=0.01, parameters=layer.parameters())
  17. adam = fleet.distributed_optimizer(adam)
  18. dp_layer = fleet.distributed_model(layer)
  20. lr = adam.get_lr()
  21. print(lr) # 0.01

step ( )

注意：

1. 该API只在 Dygraph 模式下生效

执行一次优化器并进行参数更新。

返回：None。

代码示例

  1. # 这个示例需要由fleetrun启动, 用法为:
  2. # fleetrun --gpus=0,1 example.py
  3. # 脚本example.py中的代码是下面这个示例.
  5. import paddle
  6. import paddle.nn as nn
  7. from paddle.distributed import fleet
  9. class LinearNet(nn.Layer):
  10.     def __init__(self):
  11.         super(LinearNet, self).__init__()
  12.         self._linear1 = nn.Linear(10, 10)
  13.         self._linear2 = nn.Linear(10, 1)
  15.     def forward(self, x):
  16.         return self._linear2(self._linear1(x))
  18. # 1. initialize fleet environment
  19. fleet.init(is_collective=True)
  21. # 2. create layer & optimizer
  22. layer = LinearNet()
  23. loss_fn = nn.MSELoss()
  24. adam = paddle.optimizer.Adam(
  25.     learning_rate=0.001, parameters=layer.parameters())
  27. # 3. get data_parallel model using fleet
  28. adam = fleet.distributed_optimizer(adam)
  29. dp_layer = fleet.distributed_model(layer)
  31. # 4. run layer
  32. inputs = paddle.randn([10, 10], 'float32')
  33. outputs = dp_layer(inputs)
  34. labels = paddle.randn([10, 1], 'float32')
  35. loss = loss_fn(outputs, labels)
  37. print("loss:", loss.numpy())
  39. loss.backward()
  41. adam.step()
  42. adam.clear_grad()

clear_grad ( )

注意：

1. 该API只在 Dygraph 模式下生效

清除需要优化的参数的梯度。

返回：None。

代码示例

  1. # 这个示例需要由fleetrun启动, 用法为:
  2. # fleetrun --gpus=0,1 example.py
  3. # 脚本example.py中的代码是下面这个示例.
  5. import paddle
  6. import paddle.nn as nn
  7. from paddle.distributed import fleet
  9. class LinearNet(nn.Layer):
  10.     def __init__(self):
  11.         super(LinearNet, self).__init__()
  12.         self._linear1 = nn.Linear(10, 10)
  13.         self._linear2 = nn.Linear(10, 1)
  15.     def forward(self, x):
  16.         return self._linear2(self._linear1(x))
  18. # 1. initialize fleet environment
  19. fleet.init(is_collective=True)
  21. # 2. create layer & optimizer
  22. layer = LinearNet()
  23. loss_fn = nn.MSELoss()
  24. adam = paddle.optimizer.Adam(
  25.     learning_rate=0.001, parameters=layer.parameters())
  27. # 3. get data_parallel model using fleet
  28. adam = fleet.distributed_optimizer(adam)
  29. dp_layer = fleet.distributed_model(layer)
  31. # 4. run layer
  32. inputs = paddle.randn([10, 10], 'float32')
  33. outputs = dp_layer(inputs)
  34. labels = paddle.randn([10, 1], 'float32')
  35. loss = loss_fn(outputs, labels)
  37. print("loss:", loss.numpy())
  39. loss.backward()
  41. adam.step()
  42. adam.clear_grad()

minimize ( loss, startup_program=None, parameter_list=None, no_grad_set=None )

util