PaddlePaddle分布式训练及CTR预估模型应用

　　前言：我在github上建立了一个新的repo：PaddleAI, 准备用Paddle作的一系列有趣又实用的案例，全部的案例都会上传数据代码和预训练模型，下载后能够在30s内上手，跑demo出结果，让你们尽快看到训练结果，用小批量数据调试，再用全量数据跑模型，固然，也能够基于我上传的预训练模型进行迁移学习，若是你们有须要的话。今天刚写好第一个项目，用Paddle作广告CTR预估，来源于Kaggle的比赛Display Advertising Challenge, 感兴趣的读者往下看～（也能够留言期待的todo案例）

 　　github：https://github.com/huxiaoman7/PaddleAI  

　　Paddle models：https://github.com/PaddlePaddle/models

　　欢迎你们star、fork、提issue和贡献新案例～

 

---

数据准备

数听说明

• 数据来源：Kaggle公司举办的展现广告竞赛中所使用的Criteo数据集。该数据包含数百万展现广告的特征值和点击反馈，目的是对点击率（CTR）的预测作基准预测。

• 数据背景：Criteo是在线效果类数字营销厂商，于2005年在法国巴黎成立，目前的核心业务是重定向广告（retargeting）。Criteo在全球范围内共有31间办事处，有6间位于欧洲，有5间位于北美，有1间在巴西，在亚太地区总共有5间办事处。Criteo是在线效果类展现广告厂商于2014年5月13日宣布启动在中国的业务和运营，并将北京设为中国区总部所在地。Criteo的核心产品主要包括访客广告、流失客户广告、移动应用内效果型广告和AD-X 移动广告跟踪分析产品等。Criteo拥有世界领先的自主学习式推荐引擎和预测引擎，可以经过其对于市场的洞察提供可评估的结果，于是可以在正确的时间经过推送广告，将对的产品推荐给对的用户。而且，随着每一条广告的交付，Criteo的引擎在预测和推荐方面的精确性也不断提升。

• 数据格式：

  • 格式：… 。共计39个特征，13个数值特征（int），26个类别特征。若value为空值，则为空白

• 训练数据：train.txt：Criteo 公司在七天内的部分流量。每行对应的是Critio的展现广告，第一列表明该广告是否被点击。咱们对正样本（已点击）的和负样本（未点击）均作了子采样来减小数据量。类别特征的值已通过哈希处理为64位来进行脱敏。特征的语义没有公开，而且有些特征有缺失值。行按照时间排序。

  • 示例（只显示了前20列）

• 测试数据：test.txt：测试集于训练集的计算方式相同，但对应的是训练集时间段的后一天的事件。而且第一列（label）已被移除。
  • 示例（只显示了前20列）

数据处理

• 下载数据

  cd data && ./download.sh && cd .. 

• 数据读取

• code：reader.py

• 原始数据中前13个feature为int型，经过reader.py将其作了数据归一化处理为float型，避免过大和太小的数据在模型训练中的影响。

  .── CriteoDataset
  │
  ├── train
  │
  ├── test
  │
  ├── infer

• 具体代码：

 1 class CriteoDataset(Dataset):  2     def __init__(self, sparse_feature_dim):  3         self.cont_min_ = [0, -3, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]  4         self.cont_max_ = [20, 600, 100, 50, 64000, 500, 100, 50, 500, 10, 10, 10, 50]  5         self.cont_diff_ = [20, 603, 100, 50, 64000, 500, 100, 50, 500, 10, 10, 10, 50]  6         self.hash_dim_ = sparse_feature_dim  7         # here, training data are lines with line_index < train_idx_
 8         self.train_idx_ = 41256555
 9         self.continuous_range_ = range(1, 14) 10         self.categorical_range_ = range(14, 40) 11 
12     def _reader_creator(self, file_list, is_train, trainer_num, trainer_id): 13         def reader(): 14             for file in file_list: 15                 with open(file, 'r') as f: 16                     line_idx = 0 17                     for line in f: 18                         line_idx += 1
19                         if is_train and line_idx > self.train_idx_: 20                             break
21                         elif not is_train and line_idx <= self.train_idx_: 22                             continue
23                         if line_idx % trainer_num != trainer_id: 24                             continue
25                         features = line.rstrip('\n').split('\t') 26                         dense_feature = [] 27                         sparse_feature = [] 28                         for idx in self.continuous_range_: 29                             if features[idx] == '': 30                                 dense_feature.append(0.0) 31                             else: 32                                 dense_feature.append((float(features[idx]) - self.cont_min_[idx - 1]) / self.cont_diff_[idx - 1]) 33                         for idx in self.categorical_range_: 34                             sparse_feature.append([hash(str(idx) + features[idx]) % self.hash_dim_]) 35 
36                         label = [int(features[0])] 37                         yield [dense_feature] + sparse_feature + [label] 38 
39         return reader

---

模型训练

网络结构

• code: network_conf.py (只用到ctr_dnn_model) 详细讲解

训练方式

单机训练

python train.py \ --train_data_path data/raw/train.txt \ 2>&1 | tee train.log

分布式训练

运行方式

sh cluster_train.sh

调用接口

 1 pe = fluid.ParallelExecutor(  2         use_cuda=False,  3         loss_name=loss.name,  4         main_program=train_program,  5         build_strategy=build_strategy,  6         exec_strategy=exec_strategy)  7 logger.info("run dist training")  8         t = fluid.DistributeTranspiler()  9         t.transpile(args.trainer_id, pservers=args.endpoints, trainers=args.trainers) 10         if args.role == "pserver" or args.role == "PSERVER": 11             logger.info("run pserver") 12             prog = t.get_pserver_program(args.current_endpoint) 13             startup = t.get_startup_program(args.current_endpoint, pserver_program=prog) 14             exe = fluid.Executor(fluid.CPUPlace()) 15  exe.run(startup) 16  exe.run(prog) 17         elif args.role == "trainer" or args.role == "TRAINER": 18             logger.info("run trainer") 19             train_prog = t.get_trainer_program() 20  train_loop(args, train_prog, py_reader, loss, auc_var, batch_auc_var, 21                        args.trainers, args.trainer_id)

注：batch_size由默认的1000修改成64，可提升auc

训练结果

• 单机训练

  • 速度太慢，迭代到第1轮batch= 4919时就停住了

• 分布式训练

  • 设置：2pserver、2trainer

  • 训练日志：alldata/log/trainer0.log 、alldata/log/trainer1.log 

  • 训练结果：　　

2019-05-11 08:34:19,678-INFO: TRAIN --> pass: 9 batch: 2577 loss: 0.467225006104 auc: 0.787909292672, batch_auc: 0.797377570934 pass_id: 0, pass_time_cost: 3150.447569 pass_id: 1, pass_time_cost: 3177.322331 pass_id: 2, pass_time_cost: 3174.676812 pass_id: 3, pass_time_cost: 3209.558880 pass_id: 4, pass_time_cost: 3134.910369 pass_id: 5, pass_time_cost: 3202.956675 pass_id: 6, pass_time_cost: 3169.575809 pass_id: 7, pass_time_cost: 3210.294044 pass_id: 8, pass_time_cost: 3039.102302 pass_id: 9, pass_time_cost: 3036.933163

模型预测

预测方式

python infer.py \ --model_path models/pass-0/ \ --data_path data/raw/valid.txt

预测结果：

• log：alldata/log/infer.txt

2019-05-13 09:35:49,177-INFO: TEST --> batch: 4500 loss: [0.46127334] auc: [0.78797872]

---

实验对比

原始数据状况

	label	数量	比例
负样本	0	34095179	0.74377662
正样本	1	11745438	0.25622338

实验数据：

• mini-demo：1%的全量数据
• demo：10%的全量数据
• raw：全量数据

BaseLine实验

• 数听说明： mini-data(1%全量数据)

• 网络配置：一层网络，batch_size =1000

• 修改方式：

  • 在network_conf.py 第151行修改,以下如所示，将input=fc3 修改成input=fc1

  • 在cluster_train.sh 里第35行和第47行将batch_size=64 修改成batch_size=1000

• 运行方式：

  • 修改完以上两个文件后，执行：sh cluster_train.sh

• 输出日志：

  • pserver：pserver0.log、pserver1.log 参数服务器的输出日志
  • trainer：trainer0.log、trainer1.log 每一个trainer的输出日志 能够经过 tail -f trainer0.log -n 999 查看输出结果

• 实验效果

  • 训练时间：33s（一轮迭代）
  • auc：0.50234167

优化实验（一）：优化网络层

• 数听说明： mini-data(1%全量数据)

• 网络配置：三层网络，batch_size =1000

• 修改方式：

  • 在network_conf.py 第151行修改,以下如所示，将input=fc1 修改成input=fc3

  • 在cluster_train.sh 里第35行和第47行将batch_size=64 修改成batch_size=1000

• 运行方式：

  • 修改完以上两个文件后，执行：sh cluster_train.sh

• 输出日志：

  • pserver：pserver0.log、pserver1.log 参数服务器的输出日志
  • trainer：trainer0.log、trainer1.log 每一个trainer的输出日志 能够经过 tail -f trainer0.log -n 999 查看输出结果

• 实验效果

  • 训练时间：35s（一轮迭代）
  • auc：0.54893279

优化实验（二）：调整batch_size

• 数听说明： mini-data(1%全量数据)

• 网络配置：三层网络，batch_size =64

• 修改方式：

  • 在network_conf.py 第151行修改,以下如所示，将input=fc1 修改成input=fc3

  • 在cluster_train.sh 里第35行和第47行将batch_size=1000 修改成batch_size=64

• 运行方式：

  • 修改完以上两个文件后，执行：sh cluster_train.sh

• 输出日志：

  • pserver：pserver0.log、pserver1.log 参数服务器的输出日志
  • trainer：trainer0.log、trainer1.log 每一个trainer的输出日志 能够经过 tail -f trainer0.log -n 999 查看输出结果

• 实验效果

  • 训练时间：103s（一轮迭代）
  • auc：0.74322927

优化实验（三）：增长数据集

• 数听说明： 全量数据

• 网络配置：三层网络，batch_size =64，数据量由10%数据（demo_data）扩充到全量数据

• 修改方式：

  • 在network_conf.py 第151行修改,以下如所示，将input=fc1 修改成input=fc3

  • 在cluster_train.sh 里第35行和第47行将batch_size=1000 修改成batch_size=64

  • 在cluster_train.sh 里将连个pserver和两个trainer的train_data_path地址修改成raw_data的地址，以下图所示，注意：一共须要修改四个地址

• 运行方式：

  • 修改完以上两个文件后，执行：sh cluster_train.sh

• 输出日志：

  • pserver：pserver0.log、pserver1.log 参数服务器的输出日志
  • trainer：trainer0.log、trainer1.log 每一个trainer的输出日志 能够经过 tail -f trainer0.log -n 999 查看输出结果

• 实验效果

  • 训练时间：3150s（一轮迭代）
  • auc：0.81093872

优化实验对比结果

• 表格1:对mini_demo(1%全量数据)作训练，采起不一样的优化方式，获得最后的优化方案的实验结果

评估	batch_size	batch_1000	batch_1000	batch_64	batch_64
优化方式	评估	一层网络	三层网络	一层网络	三层网络
mini_demo	time	33s	35s	97s	103s
	auc	0.50234167	0.54893279	0.721332392	0.74322927

• 表格2: 增长数据集对模型精度的提高

	batch_size	time	auc
demo	64	1133s	0.73777626
全量	64	3150s	0.81093872

 

优化方案总结

由以上两个表格可知：

• batch 大小的改变：在数据集和其余参数不变的状况下，batch_size由1000变为64.能够提高20%的模型精度
• 网络结构的改变：在数据集和batch_size等参数不变的状况，由一层网络变为三层网络结构，大约可提高2～4%的模型精度
• 数据集的改变：由demo数据（10%全量数据）扩充到全量数据，采用一样的batch_size，一样的迭代次数和其余超参数，大约可提高7%的精度