PaddlePaddle分佈式訓練及CTR預估模型應用

時間 2019-11-06

標籤 paddlepaddle 分佈式訓練 ctr 預估模型應用欄目系統架構简体版

原文原文鏈接

　　前言：我在github上建立了一個新的repo：PaddleAI, 準備用Paddle作的一系列有趣又實用的案例，全部的案例都會上傳數據代碼和預訓練模型，下載後能夠在30s內上手，跑demo出結果，讓你們儘快看到訓練結果，用小批量數據調試，再用全量數據跑模型，固然，也能夠基於我上傳的預訓練模型進行遷移學習，若是你們有須要的話。今天剛寫好第一個項目，用Paddle作廣告CTR預估，來源於Kaggle的比賽Display Advertising Challenge, 感興趣的讀者往下看～（也能夠留言期待的todo案例）

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

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

　　歡迎你們star、fork、提issue和貢獻新案例～github

數據準備bash

數聽說明

數據來源：Kaggle公司舉辦的展現廣告競賽中所使用的Criteo數據集。該數據包含數百萬展現廣告的特徵值和點擊反饋，目的是對點擊率（CTR）的預測作基準預測。服務器
數據背景：Criteo是在線效果類數字營銷廠商，於2005年在法國巴黎成立，目前的核心業務是重定向廣告（retargeting）。Criteo在全球範圍內共有31間辦事處，有6間位於歐洲，有5間位於北美，有1間在巴西，在亞太地區總共有5間辦事處。Criteo是在線效果類展現廣告廠商於2014年5月13日宣佈啓動在中國的業務和運營，並將北京設爲中國區總部所在地。Criteo的核心產品主要包括訪客廣告、流失客戶廣告、移動應用內效果型廣告和AD-X 移動廣告跟蹤分析產品等。Criteo擁有世界領先的自主學習式推薦引擎和預測引擎，可以經過其對於市場的洞察提供可評估的結果，於是可以在正確的時間經過推送廣告，將對的產品推薦給對的用戶。而且，隨着每一條廣告的交付，Criteo的引擎在預測和推薦方面的精確性也不斷提升。網絡
數據格式：app
- 格式：… 。共計39個特徵，13個數值特徵（int），26個類別特徵。若value爲空值，則爲空白
訓練數據：train.txt：Criteo 公司在七天內的部分流量。每行對應的是Critio的展現廣告，第一列表明該廣告是否被點擊。咱們對正樣本（已點擊）的和負樣本（未點擊）均作了子採樣來減小數據量。類別特徵的值已通過哈希處理爲64位來進行脫敏。特徵的語義沒有公開，而且有些特徵有缺失值。行按照時間排序。分佈式
- 示例（只顯示了前20列）

測試數據：test.txt：測試集於訓練集的計算方式相同，但對應的是訓練集時間段的後一天的事件。而且第一列（label）已被移除。
- 示例（只顯示了前20列）

數據處理

下載數據oop
```
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