快速开始-参数服务器¶

import paddle
import paddle.distributed.fleet as fleet

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

# 模型定义参考 examples/wide_and_deep_dataset 中 model.py
from model import WideDeepModel
model = WideDeepModel()
model.net(is_train=True)

import paddle
# 导入所需要的 fleet 依赖
import paddle.distributed.fleet as fleet

# 需要继承 fleet.MultiSlotDataGenerator
class WideDeepDatasetReader(fleet.MultiSlotDataGenerator):
    def line_process(self, line):
        features = line.rstrip('\n').split('\t')
        # 省略数据处理过程，具体实现可参考单机 reader 的 line_process()方法
        # 返回值为一个 list，其中的每个元素均为一个 list，不需要转成 np.array 格式
        # 具体格式：[[dense_value1, dense_value2, ...], [sparse_value1], [sparse_value2], ..., [label]]
        return [dense_feature] + sparse_feature + [label]

    # 实现 generate_sample()函数
    # 该方法有一个名为 line 的参数，只需要逐行处理数据，不需要对数据文件进行操作
    def generate_sample(self, line):
        def wd_reader():
            # 按行处理数据
            input_data = self.line_process(line)

            # 构造特征名数组 feature_name
            feature_name = ["dense_input"]
            for idx in categorical_range_:
                feature_name.append("C" + str(idx - 13))
            feature_name.append("label")

            # 返回一个 list，其中的每个元素都是一个元组
            # 元组的第一个元素为特征名（string 类型），第二个元素为特征值（list 类型）
            # 具体格式：[('dense_input', [dense_value1, dense_value2, ...]), ('C1', [sparse_value1]), ('C2', [sparse_value2]), ..., ('label', [label])]
            yield zip(feature_name, input_data)

        # generate_sample()函数需要返回一个可以迭代的 reader 方法
        return wd_reader

if __name__ == "__main__":
    # 调用 run_from_stdin()方法，直接从标准输入流获取待处理数据
    my_data_generator = WideDeepDatasetReader()
    my_data_generator.run_from_stdin()

dataset = paddle.distributed.QueueDataset()
thread_num = 1

# use_var 指定网络中的输入数据，pipe_command 指定数据处理脚本
# 要求 use_var 中输入数据的顺序与数据处理脚本输出的特征顺序一一对应
dataset.init(use_var=model.inputs,
             pipe_command="python reader.py",
             batch_size=batch_size,
             thread_num=thread_num)

train_files_list = [os.path.join(train_data_path, x)
                      for x in os.listdir(train_data_path)]

# set_filelist 指定 dataset 读取的训练文件的列表
dataset.set_filelist(train_files_list)

# 定义异步训练
dist_strategy = fleet.DistributedStrategy()
dist_strategy.a_sync = True

# 定义单机优化器
optimizer = paddle.optimizer.SGD(learning_rate=0.0001)
# 单机优化器转换成分布式优化器
optimizer = fleet.distributed_optimizer(optimizer, dist_strategy)
# 使用分布式优化器最小化模型损失值 model.loss，model.loss 定义参见 model.py
optimizer.minimize(model.loss)

if fleet.is_server():
    fleet.init_server()
    fleet.run_server()
else:
    exe = paddle.static.Executor(paddle.CPUPlace())
    exe.run(paddle.static.default_startup_program())

    fleet.init_worker()

    for epoch_id in range(1):
        exe.train_from_dataset(paddle.static.default_main_program(),
                               dataset,
                               paddle.static.global_scope(),
                               debug=False,
                               fetch_list=[model.loss],
                               fetch_info=["loss"],
                               print_period=1)

    fleet.stop_worker()