CPUPS 流式训练示例¶

# 该方法定义在 tools/utils/static_ps/flow_helper.py 中
def get_online_pass_interval(split_interval, split_per_pass,
                            is_data_hourly_placed):
    split_interval = int(split_interval)
    split_per_pass = int(split_per_pass)
    splits_per_day = 24 * 60 // split_interval
    pass_per_day = splits_per_day // split_per_pass
    left_train_hour = 0
    right_train_hour = 23

    start = 0
    split_path = []
    for i in range(splits_per_day):
        h = start // 60
        m = start % 60
        if h < left_train_hour or h > right_train_hour:
            start += split_interval
            continue
        if is_data_hourly_placed:
            split_path.append("%02d" % h)
        else:
            split_path.append("%02d%02d" % (h, m))
        start += split_interval

    start = 0
    online_pass_interval = []
    for i in range(pass_per_day):
        online_pass_interval.append([])
        for j in range(start, start + split_per_pass):
            online_pass_interval[i].append(split_path[j])
        start += split_per_pass

    return online_pass_interval

# 根据 split_interval 和 split_per_pass，在训练之前生成每个 Pass 所需要的数据分片列表
self.online_intervals = get_online_pass_interval(
          self.split_interval, self.split_per_pass, False)

# 该方法定义在 tools/utils/static_ps/flow_helper.py 中
def file_ls(path_array, client):
    # 获取 path 数组下的所有文件
    # 如果数据存在 hdfs/afs 上，需要使用 hadoop_client
    result = []
    for path in path_array:
        if is_local(path):
            cur_path = os.listdir(path)
        else:
            cur_path = client.ls_dir(path)[1]
        if len(cur_path) > 0:
            result += [os.path.join(path, i) for i in cur_path]
    logger.info("file ls result = {}".format(result))
    return result

cur_path = []
for i in self.online_intervals[pass_index - 1]:
    # p 为一个具体的数据分片目录，例如："data/20190720/0000"
    p = os.path.join(train_data_path, day, str(i))
    if self.data_donefile:
      # 数据等待策略生效，如果目录下无 data_donefile 文件，需等待 data_sleep_second 后再探测
      cur_donefile = os.path.join(p, self.data_donefile)
      data_ready(cur_donefile, self.data_sleep_second,
                self.hadoop_client)
    # cur_path 存储当前 Pass 下的所有数据目录，对应一个或多个数据分片文件夹
    # 例如：["data/20190720/0000", "data/20190720/0005"]
    cur_path.append(p)

# 获取当前数据分片下的所有数据文件
global_file_list = file_ls(cur_path, self.hadoop_client)
# 将数据文件拆分到每个 Worker 上
my_file_list = fleet.util.get_file_shard(global_file_list)

# 创建 InMemoryDataset
dataset = paddle.distributed.InMemoryDataset()

# InMemoryDataset 初始化
dataset.init(use_var=self.input_data,
             pipe_command=self.pipe_command,
             batch_size=batch_size,
             thread_num=thread_num)

# 设置文件列表为拆分到当前 Worker 的 file_list
dataset.set_filelist(my_file_list)

# 将训练数据加载到内存
dataset.load_into_memory()
# 数据全局打散
dataset.global_shuffle(fleet, shuffle_thread_num)
# 获取当前 Worker 在全局打散之后的训练数据样例数
shuffle_data_size = dataset.get_shuffle_data_size(fleet)

# 省略具体的训练过程

# 在当前 Pass 训练结束后，InMemoryDataset 需调用 release_memory()方法释放内存
dataset.release_memory()

# 该方法定义在 tools/utils/static_ps/metric_helper.py 中
def set_zero(var_name,
             scope=fluid.global_scope(),
             place=fluid.CPUPlace(),
             param_type="int64"):
    # 对变量进行清零操作
    param = scope.var(var_name).get_tensor()
    param_array = np.zeros(param._get_dims()).astype(param_type)
    param.set(param_array, place)

# 组网阶段，AUC 算子在计算 auc 指标同时，返回正负样例中间统计结果（stat_pos, stat_neg）
auc, batch_auc, [batch_stat_pos, batch_stat_neg, stat_pos, stat_neg] = \
    paddle.static.auc(input=pred, label=label)

strategy = fleet.DistributedStrategy()
strategy.a_sync = True

# 获取计算指标所需的中间变量的 name 列表，并将其配置到 strategy 的 stat_var_names 选项中
stat_var_names = [stat_pos.name, stat_neg.name]
strategy.trainer_desc_configs = {"stat_var_names": stat_var_names}

# 省略具体训练过程

# 训练结束后，利用 AUC 算子返回的中间计算结果，以及 fleet 提供的分布式指标计算接口，完成全局 AUC 计算。
global_auc = fleet.metrics.auc(stat_pos, stat_neg)

# 指标计算所需的中间变量清零
set_zero(stat_pos.name)
set_zero(stat_neg.name)

# 该方法定义在 tools/utils/static_ps/program_helper.py 中
def set_dump_config(program, dump_config):
    # 配置 dump 相关信息
    if dump_config.get("dump_fields_path") is not None:
        # 打印出的中间结果存放路径
        program._fleet_opt["dump_fields_path"] = dump_config.get(
            "dump_fields_path")
    if dump_config.get("dump_fields") is not None:
        # 需要打印的中间层变量名
        program._fleet_opt["dump_fields"] = dump_config.get("dump_fields")
    if dump_config.get("dump_param") is not None:
        # 需要打印的参数名
        program._fleet_opt["dump_param"] = dump_config.get("dump_param")

# dataset 需要设置 parse_ins_id 和 parse_content 为 True
# 同时，输入数据也需要在最前面增加 ins_id 和 content 两个字段，用来标识具体的样例
dataset.set_parse_ins_id(True)
dataset.set_parse_content(True)

# 在训练或者预测前配置 dump 信息
dump_fields_dir = "dump_data"
# dump 出的中间结果存放路径
dump_fields_path = "{}/{}/{}".format(dump_fields_dir, day, pass_index)
# 需要 dump 的中间变量，具体定义参考 static_model.py 和 net.py
dump_fields = [var.name for var in self.infer_dump_fields]
# 调用 set_dump_config 配置 dump 信息
set_dump_config(paddle.static.default_main_program(), {
    "dump_fields_path": dump_fields_path,
    "dump_fields": dump_fields
})

# 预测
self.exe.infer_from_dataset(
    program=paddle.static.default_main_program(),
    dataset=cur_dataset,
    fetch_list=fetch_vars,
    fetch_info=fetch_info,
    print_period=print_step,
    debug=debug)

# 该方法定义在 tools/utils/static_ps/flow_helper.py 中
def save_model(exe, output_path, day, pass_id, mode=0):
    # 保存明文模型，具体目录为 output_path/day/pass_id，例如：output_path/20190720/6
    day = str(day)
    pass_id = str(pass_id)
    suffix_name = "/%s/%s/" % (day, pass_id)
    model_path = output_path + suffix_name
    fleet.save_persistables(exe, model_path, None, mode=mode)

# 该方法定义在 tools/utils/static_ps/flow_helper.py 中
def save_batch_model(exe, output_path, day):
    # 保存 batch_model，具体目录为 output_path/day/0，例如：output_path/20190721/0
    day = str(day)
    suffix_name = "/%s/0/" % day
    model_path = output_path + suffix_name
    fleet.save_persistables(exe, model_path, mode=3)

for pass_id in range(1, 1 + len(self.online_intervals)):
    # 分 Pass 训练，省略具体训练过程

    if pass_id % self.checkpoint_per_pass == 0:
        # 在到达配置的 Pass 时，调用 save_model 保存明文模型
        save_model(self.exe, self.save_model_path, day, pass_id)

# 一天数据训练完成
# 调用 shrink 删除某些稀疏参数
fleet.shrink()

next_day = get_next_day(day)
# 调用 save_batch_model 保存 batch_model
save_batch_model(self.exe, self.save_model_path, next_day)

# 该方法定义在 tools/utils/static_ps/flow_helper.py 中
def save_inference_model(output_path, day, pass_id, exe, feed_vars, target_vars, client):
    if pass_id != -1:
        # mode=1，保存 delta 模型
        mode = 1
        suffix_name = "/%s/delta-%s/" % (day, pass_id)
        model_path = output_path.rstrip("/") + suffix_name
    else:
        # mode=2，保存 base 模型
        mode = 2
        suffix_name = "/%s/base/" % day
        model_path = output_path.rstrip("/") + suffix_name
    fleet.save_inference_model(
        exe,
        model_path, [feed.name for feed in feed_vars],
        target_vars,
        mode=mode)
    if not is_local(model_path) and fleet.is_first_worker():
        client.upload_dir("./dnn_plugin", model_path)
    fleet.barrier_worker()

# 定义推理裁剪网络的输入和输出，具体定义参考 static_model.py 和 net.py
self.inference_feed_vars = model.inference_feed_vars
self.inference_target_var = model.inference_target_var
for pass_id in range(1, 1 + len(self.online_intervals)):
    # 分 Pass 训练，省略具体训练过程

    if pass_id % self.save_delta_frequency == 0:
        # 在到达配置的 Pass 时，调用 save_xbox_model 保存 delta 推理模型
        save_inference_model(self.save_model_path, day, pass_id,
                             self.exe, self.inference_feed_vars,
                             self.inference_target_var,
                             self.hadoop_client)

# 一天数据训练完成
# 调用 shrink 删除某些稀疏参数
fleet.shrink()

next_day = get_next_day(day)
# 由 0 号节点调用 save_xbox_model 保存 base 推理模型
save_inference_model(self.save_model_path, next_day, -1,
                     self.exe, self.inference_feed_vars,
                     self.inference_target_var,
                     self.hadoop_client)