Skip to content

P
PaddleRec

PaddlePaddle / PaddleRec

通知 68
Star 12
Fork 5
P
PaddleRec
提交 6e023ec8 编写于 作者: Y yinhaofeng
浏览文件
操作

picture

上级 d33274ea
隐藏空白更改
内联 并排
Showing with 128 addition and 220 deletion
models/rank/fm/config.yaml
浏览文件 @ 6e023ec8
......@@ -35,7 +35,7 @@ hyper_parameters:
# 用户自定义配置
optimizer:
class: Adam
learning_rate: 0.0001
learning_rate: 0.001
sparse_feature_number: 1086460
sparse_feature_dim: 9
is_sparse: False
......@@ -48,7 +48,7 @@ mode: [train_runner,infer_runner]
runner:
- name: train_runner
class: train
epochs: 1
epochs: 2
device: cpu
init_model_path: ""
save_checkpoint_interval: 1
......
models/rank/fm/readme.md
浏览文件 @ 6e023ec8
# 基于FM模型的点击率预估模型
## 介绍
以下是本例的简要目录结构及说明：
```
├── sample_data #样例数据
├── train
├── sample_train.txt #训练数据样例
├── preprocess.py #数据处理程序
├── run.sh #数据一键处理脚本
├── download_preprocess.py #数据下载脚本
├── get_slot_data.py #格式整理程序
├── __init__.py
├── README.md #文档
├── model.py #模型文件
├── config.yaml #配置文件
```
注：在阅读该示例前，建议您先了解以下内容：
[paddlerec入门教程](https://github.com/PaddlePaddle/PaddleRec/blob/master/README.md)
## 内容
- [模型简介](#模型简介)
- [数据准备](#数据准备)
- [运行环境](#运行环境)
- [快速开始](#快速开始)
- [模型组网](#模型组网)
- [效果复现](#效果复现)
- [进阶使用](#进阶使用)
- [FAQ](#FAQ)
## 模型简介
`CTR(Click Through Rate)`，即点击率，是“推荐系统/计算广告”等领域的重要指标，对其进行预估是商品推送/广告投放等决策的基础。简单来说，CTR预估对每次广告的点击情况做出预测，预测用户是点击还是不点击。CTR预估模型综合考虑各种因素、特征，在大量历史数据上训练，最终对商业决策提供帮助。本模型实现了下述论文中的FM模型：
```text
......@@ -20,244 +51,117 @@
```bash
<label> <integer feature 1> ... <integer feature 13> <categorical feature 1> ... <categorical feature 26>
```
其中```<label>```表示广告是否被点击，点击用1表示，未点击用0表示。```<integer feature>```代表数值特征（连续特征），共有13个连续特征。```<categorical feature>```代表分类特征（离散特征），共有26个离散特征。相邻两个特征用```\t```分隔，缺失特征用空格表示。测试集中```<label>```特征已被移除。
### 数据预处理
数据预处理共包括两步：
- 将原始训练集按9:1划分为训练集和验证集
- 数值特征（连续特征）需进行归一化处理，但需要注意的是，对每一个特征```<integer feature i>```，归一化时用到的最大值并不是用全局最大值，而是取排序后95%位置处的特征值作为最大值，同时保留极值。
其中```<label>```表示广告是否被点击，点击用1表示，未点击用0表示。```<integer feature>```代表数值特征（连续特征），共有13个连续特征。```<categorical feature>```代表分类特征（离散特征），共有26个离散特征。相邻两个特征用```\t```分隔，缺失特征用空格表示。测试集中```<label>```特征已被移除。
详细的数据解析过程请参考dnn模型下的readme文件：[基于DNN模型的点击率预估模型](https://github.com/PaddlePaddle/PaddleRec/blob/master/models/rank/dnn/README.md)
### 一键下载训练及测试数据
```bash
sh run.sh
```
进入models/rank/fm/data目录下，执行该脚本，会从国内源的服务器上下载Criteo数据集，并解压到指定文件夹，然后自动处理数据转化为可直接进行训练的格式。解压后全量训练数据放置于`./train_datal`，全量测试数据放置于`./test_data`，可以直接输入的训练数据放置于`./slot_train_datal`，可直接输入的测试数据放置于`./slot_test_datal`
执行该脚本的理想输出为：
```
download and extract starting...
Downloading dac.tar.gz
[==================================================] 100.00%
Uncompress dac.tar.gz
[==================================================] 100.00%
Downloading deepfm%2Ffeat_dict_10.pkl2
[==================================================] 100.00%
download and extract finished
preprocessing...
('generating feature dict', 0)
('generating feature dict', 0)
('generating feature dict', 0)
('generating feature dict', 0)
...
('generating feature dict', 1)
('generating feature dict', 1)
('args.num_feat ', 1086460)
Done!
preprocess done
done
```
## 数据读取
为了能高速运行CTR模型的训练，`PaddleRec`封装了`dataset`与`dataloader`API进行高性能的数据读取。
如何在我们的训练中引入dataset读取方式呢？无需变更数据格式，只需在我们的训练代码中加入以下内容，便可达到媲美二进制读取的高效率，以下是一个比较完整的流程：
### 引入dataset
1. 通过工厂类`fluid.DatasetFactory()`创建一个dataset对象。
2. 将我们定义好的数据输入格式传给dataset，通过`dataset.set_use_var(inputs)`实现。
3. 指定我们的数据读取方式，由`dataset_generator.py`实现数据读取的规则，后面将会介绍读取规则的实现。
4. 指定数据读取的batch_size。
5. 指定数据读取的线程数，该线程数和训练线程应保持一致，两者为耦合的关系。
6. 指定dataset读取的训练文件的列表。
```python
def get_dataset(inputs, args)
dataset = fluid.DatasetFactory().create_dataset()
dataset.set_use_var(inputs)
dataset.set_pipe_command("python dataset_generator.py")
dataset.set_batch_size(args.batch_size)
dataset.set_thread(int(args.cpu_num))
file_list = [
str(args.train_files_path) + "/%s" % x
for x in os.listdir(args.train_files_path)
]
logger.info("file list: {}".format(file_list))
return dataset, file_list
```
## 运行环境
PaddlePaddle>=1.7.2
python 2.7/3.5/3.6/3.7
PaddleRec >=0.1
os : windows/linux/macos
## 快速开始
本文提供了样例数据可以供您快速体验，在paddlerec目录下执行下面的命令即可快速启动训练：
### 如何指定数据读取规则
在上文我们提到了由`dataset_generator.py`实现具体的数据读取规则，那么，怎样为dataset创建数据读取的规则呢？
以下是`dataset_generator.py`的全部代码，具体流程如下：
1. 首先我们需要引入dataset的库，位于`paddle.fluid.incubate.data_generator`。
2. 声明一些在数据读取中会用到的变量，如示例代码中的`cont_min_`、`categorical_range_`等。
3. 创建一个子类，继承dataset的基类，基类有多种选择，如果是多种数据类型混合，并且需要转化为数值进行预处理的，建议使用`MultiSlotDataGenerator`；若已经完成了预处理并保存为数据文件，可以直接以`string`的方式进行读取，使用`MultiSlotStringDataGenerator`，能够进一步加速。在示例代码，我们继承并实现了名为`CriteoDataset`的dataset子类，使用`MultiSlotDataGenerator`方法。
4. 继承并实现基类中的`generate_sample`函数，逐行读取数据。该函数应返回一个可以迭代的reader方法(带有yield的函数不再是一个普通的函数，而是一个生成器generator，成为了可以迭代的对象，等价于一个数组、链表、文件、字符串etc.)
5. 在这个可以迭代的函数中，如示例代码中的`def reader()`，我们定义数据读取的逻辑。例如对以行为单位的数据进行截取，转换及预处理。
6. 最后，我们需要将数据整理为特定的格式，才能够被dataset正确读取，并灌入的训练的网络中。简单来说，数据的输出顺序与我们在网络中创建的`inputs`必须是严格一一对应的。在示例代码中，我们将数据整理成`click:value dense_feature:value ... dense_feature:value 1:value ... 26:value`的格式。用print输出是因为我们在run.sh中将结果重定向到slot_train_data等文件中，由模型直接读取。在用户自定义使用时，可以使用`zip`的方法将参数名与数值构成的元组组成了一个list，并将其yield输出，并在config.yaml中的data_converter参数指定reader的路径。
```python
class Reader(dg.MultiSlotDataGenerator):
def __init__(self, config):
dg.MultiSlotDataGenerator.__init__(self)
_config = envs.load_yaml(config)
def init(self):
self.cont_min_ = [0, -3, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
self.cont_max_ = [
5775, 257675, 65535, 969, 23159456, 431037, 56311, 6047, 29019, 46,
231, 4008, 7393
]
self.cont_diff_ = [
self.cont_max_[i] - self.cont_min_[i]
for i in range(len(self.cont_min_))
]
self.continuous_range_ = range(1, 14)
self.categorical_range_ = range(14, 40)
# load preprocessed feature dict
self.feat_dict_name = "sample_data/feat_dict_10.pkl2"
self.feat_dict_ = pickle.load(open(self.feat_dict_name, 'rb'))
def _process_line(self, line):
features = line.rstrip('\n').split('\t')
feat_idx = []
feat_value = []
for idx in self.continuous_range_:
if features[idx] == '':
feat_idx.append(0)
feat_value.append(0.0)
else:
feat_idx.append(self.feat_dict_[idx])
feat_value.append(
(float(features[idx]) - self.cont_min_[idx - 1]) /
self.cont_diff_[idx - 1])
for idx in self.categorical_range_:
if features[idx] == '' or features[idx] not in self.feat_dict_:
feat_idx.append(0)
feat_value.append(0.0)
else:
feat_idx.append(self.feat_dict_[features[idx]])
feat_value.append(1.0)
label = [int(features[0])]
return feat_idx, feat_value, label
def generate_sample(self, line):
"""
Read the data line by line and process it as a dictionary
"""
def data_iter():
feat_idx, feat_value, label = self._process_line(line)
s = ""
for i in [('feat_idx', feat_idx), ('feat_value', feat_value),
('label', label)]:
k = i[0]
v = i[1]
for j in v:
s += " " + k + ":" + str(j)
print(s.strip()) # add print for data preprocessing
return data_iter
reader = Reader(
"../config.yaml") # run this file in original folder to find config.yaml
reader.init()
reader.run_from_stdin()
```
### 快速调试Dataset
我们可以脱离组网架构，单独验证Dataset的输出是否符合我们预期。使用命令
`cat 数据文件 | python dataset读取python文件`进行dataset代码的调试：
```bash
cat train_data/part-0 | python python get_slot_data.py
python -m paddlerec.run -m models/rank/fm/config.yaml
```
输出的数据格式如下：
`label:value dense_input:value ... dense_input:value sparse_input:value ... sparse_input:value `
使用样例数据快速跑通的结果实例:
理想的输出为(截取了一个片段)：
```bash
...
click:0 dense_feature:0.05 dense_feature:0.00663349917081 dense_feature:0.05 dense_feature:0.0 dense_feature:0.02159375 dense_feature:0.008 dense_feature:0.15 dense_feature:0.04 dense_feature:0.362 dense_feature:0.1 dense_feature:0.2 dense_feature:0.0 dense_feature:0.04 1:715353 2:817085 3:851010 4:833725 5:286835 6:948614 7:881652 8:507110 9:27346 10:646986 11:643076 12:200960 13:18464 14:202774 15:532679 16:729573 17:342789 18:562805 19:880474 20:984402 21:666449 22:26235 23:700326 24:452909 25:884722 26:787527
```
PaddleRec: Runner train_runner Begin
Executor Mode: train
processor_register begin
Running SingleInstance.
Running SingleNetwork.
Warning:please make sure there are no hidden files in the dataset folder and check these hidden files:[]
Running SingleStartup.
Running SingleRunner.
I0925 03:07:10.455991 26128 parallel_executor.cc:440] The Program will be executed on CPU using ParallelExecutor, 1 cards are used, so 1 programs are executed in parallel.
I0925 03:07:10.460461 26128 build_strategy.cc:365] SeqOnlyAllReduceOps:0, num_trainers:1
I0925 03:07:10.476308 26128 parallel_executor.cc:307] Inplace strategy is enabled, when build_strategy.enable_inplace = True
I0925 03:07:10.477986 26128 parallel_executor.cc:375] Garbage collection strategy is enabled, when FLAGS_eager_delete_tensor_gb = 0
2020-09-25 03:07:10,761-INFO: [Train] batch: 1, time_each_interval: 0.31s, BATCH_AUC: [0.55555556], AUC: [0.55555556]
2020-09-25 03:07:10,816-INFO: [Train] batch: 2, time_each_interval: 0.06s, BATCH_AUC: [0.61111111], AUC: [0.61111111]
2020-09-25 03:07:10,870-INFO: [Train] batch: 3, time_each_interval: 0.05s, BATCH_AUC: [0.515625], AUC: [0.515625]
2020-09-25 03:07:10,921-INFO: [Train] batch: 4, time_each_interval: 0.05s, BATCH_AUC: [0.59], AUC: [0.59]
2020-09-25 03:07:10,975-INFO: [Train] batch: 5, time_each_interval: 0.05s, BATCH_AUC: [0.52380952], AUC: [0.52380952]
...
2020-09-25 03:07:13,307-INFO: [Train] batch: 17, time_each_interval: 0.06s, BATCH_AUC: [0.63376048], AUC: [0.63376048]
2020-09-25 03:07:13,363-INFO: [Train] batch: 18, time_each_interval: 0.06s, BATCH_AUC: [0.64298615], AUC: [0.64298615]
2020-09-25 03:07:13,417-INFO: [Train] batch: 19, time_each_interval: 0.05s, BATCH_AUC: [0.64117133], AUC: [0.64117133]
epoch 1 done, use time: 1.08840990067, global metrics: BATCH_AUC=0.64117133, AUC=[0.64117133]
PaddleRec Finish
```
### FM模型组网
## 模型组网
FM模型的组网本质是一个二分类任务，代码参考`model.py`。模型主要组成是一阶项部分，二阶项部分以及相应的分类任务的loss计算和auc计算。模型的组网可以较为方便的于公式对应，FM的表达式如下，可观察到，只是在线性表达式后面加入了新的交叉项特征及对应的权值。
$$Out=sigmoid(b + \sum^{N}_{i=1}W_iX_i + \sum^{N-1}_{i=1}\sum^{N}_{j=i+1}W_{ij}X_iX_j)$$
#### 一阶项部分
一阶项部分类似于我们rank下的logistic_regression模型。主要由embedding层和reduce_sum层组成
首先介绍Embedding层的搭建方式：`Embedding`层的输入是`feat_idx`，shape由超参的`sparse_feature_number`定义。需要特别解释的是`is_sparse`参数，当我们指定`is_sprase=True`后，计算图会将该参数视为稀疏参数，反向更新以及分布式通信时，都以稀疏的方式进行，会极大的提升运行效率，同时保证效果一致。
各个稀疏的输入通过Embedding层后，进行reshape操作，方便和连续值进行结合。
将离散数据通过embedding查表得到的值，与连续数据的输入进行相乘再累加的操作，合为一个一阶项的整体。
我们又构造了一个初始化为0，shape为1的变量，作为公式前零阶项的部分。
$$b + \sum^{N}_{i=1}W_iX_i$$
```python
first_weights_re = fluid.embedding(
input=feat_idx,
is_sparse=self.is_sparse,
is_distributed=self.is_distributed,
dtype='float32',
size=[self.sparse_feature_number + 1, 1],
padding_idx=0,
param_attr=fluid.ParamAttr(
initializer=fluid.initializer.TruncatedNormalInitializer(
loc=0.0, scale=init_value_),
regularizer=fluid.regularizer.L1DecayRegularizer(self.reg))) # (batch_size * num_field) * 1 * 1(embedding_size)
first_weights = fluid.layers.reshape(
first_weights_re,
shape=[-1, self.num_field, 1]) # batch_size * num_field * 1
y_first_order = fluid.layers.reduce_sum((first_weights * feat_value),1) # batch_size * 1
b_linear = fluid.layers.create_parameter(
shape=[1],
dtype='float32',
default_initializer=fluid.initializer.ConstantInitializer(value=0)) # 1
```
#### 二阶项部分
二阶项部分主要实现了公式中的交叉项部分，也就是特征的组合部分。Wij求解的思路是通过矩阵分解的方法。所有的二次项参数Wij可以组成一个对称阵W，那么这个矩阵就可以分解为 $W=V^TV$，V 的第 i 列便是第 i 维特征的隐向量。交叉项的展开式如下：
$$\sum^{N-1}_{i=1}\sum^{N}_{j=i+1}W_{ij}X_iX_j =1/2\sum^{k}_{j=1}((\sum^{N}_{i=1}W_iX_i)^2-\sum^{N}_{i=1}W_i^2X_i^2)$$
```python
feat_embeddings_re = fluid.embedding(
input=feat_idx,
is_sparse=self.is_sparse,
is_distributed=self.is_distributed,
dtype='float32',
size=[self.sparse_feature_number + 1, self.sparse_feature_dim],
padding_idx=0,
param_attr=fluid.ParamAttr(
initializer=fluid.initializer.TruncatedNormalInitializer(
loc=0.0,
scale=init_value_ /
math.sqrt(float(self.sparse_feature_dim))))) # (batch_size * num_field) * 1 * embedding_size
feat_embeddings = fluid.layers.reshape(feat_embeddings_re,shape=[-1, self.num_field, self.sparse_feature_dim]) # batch_size * num_field * embedding_size
feat_embeddings = feat_embeddings * feat_value
# sum_square part
summed_features_emb = fluid.layers.reduce_sum(feat_embeddings, 1) # batch_size * embedding_size
summed_features_emb_square = fluid.layers.square(summed_features_emb) # batch_size * embedding_size
# square_sum part
squared_features_emb = fluid.layers.square(feat_embeddings) # batch_size * num_field * embedding_size
squared_sum_features_emb = fluid.layers.reduce_sum(squared_features_emb, 1) # batch_size * embedding_size
y_FM = 0.5 * fluid.layers.reduce_sum(
summed_features_emb_square - squared_sum_features_emb,
dim=1,
keep_dim=True) # batch_size * 1
```
<img align="center" src="picture/1.jpg">
### 一阶项部分
一阶项部分类似于我们rank下的logistic_regression模型。主要由embedding层和reduce_sum层组成
首先介绍Embedding层的搭建方式：`Embedding`层的输入是`feat_idx`，shape由超参的`sparse_feature_number`定义。需要特别解释的是`is_sparse`参数，当我们指定`is_sprase=True`后，计算图会将该参数视为稀疏参数，反向更新以及分布式通信时，都以稀疏的方式进行，会极大的提升运行效率，同时保证效果一致。
各个稀疏的输入通过Embedding层后，进行reshape操作，方便和连续值进行结合。
将离散数据通过embedding查表得到的值，与连续数据的输入进行相乘再累加的操作，合为一个一阶项的整体。
我们又构造了一个初始化为0，shape为1的变量，作为公式前零阶项的部分。
#### Loss及Auc计算
<img align="center" src="picture/2.jpg">
### 二阶项部分
二阶项部分主要实现了公式中的交叉项部分，也就是特征的组合部分。Wij求解的思路是通过矩阵分解的方法。所有的二次项参数Wij可以组成一个对称阵W，那么这个矩阵就可以如下分解：
<img align="center" src="picture/3.jpg">
V 的第 i 列便是第 i 维特征的隐向量。特征分量Xi与Xj的交叉项系数就等于Xi对应的隐向量与Xj对应的隐向量的内积，即每个参数 wij=⟨vi,vj⟩
交叉项的展开式如下：
<img align="center" src="picture/4.jpg">
### Loss及Auc计算
- 预测的结果通过直接通过激活函数sigmoid给出，为了得到每条样本分属于正负样本的概率，我们将预测结果和`1-predict`合并起来得到predict_2d，以便接下来计算auc。
- 每条样本的损失为负对数损失值，label的数据类型将转化为float输入。
- 该batch的损失`avg_cost`是各条样本的损失之和
- 我们同时还会计算预测的auc，auc的结果由`fluid.layers.auc()`给出，该层的返回值有三个，分别是全局auc: `auc_var`，当前batch的auc: `batch_auc_var`，以及auc_states: `_`，auc_states包含了`batch_stat_pos, batch_stat_neg, stat_pos, stat_neg`信息。
- 该batch的损失`avg_cost`是各条样本的损失之和
- 我们同时还会计算预测的auc，auc的结果由`fluid.layers.auc()`给出，该层的返回值有三个，分别是全局auc: `auc_var`，当前batch的auc: `batch_auc_var`，以及auc_states: `_`，auc_states包含了`batch_stat_pos, batch_stat_neg, stat_pos, stat_neg`信息。
完成上述组网后，我们最终可以通过训练拿到`auc`指标。
## 效果复现
为了方便使用者能够快速的跑通每一个模型，我们在每个模型下都提供了样例数据。如果需要复现readme中的效果,请按如下步骤依次操作即可。
在全量数据下模型的指标如下：
| 模型 | auc | batch_size | thread_num| epoch_num| Time of each epoch |
| :------| :------ | :------| :------ | :------| :------ |
| FM | 0.7619 | 1024 | 10 | 2 | 约3.5小时 |
1. 确认您当前所在目录为PaddleRec/models/rank/fm
2. 在data目录下运行数据一键处理脚本，命令如下：
```
cd data
sh run.sh
cd ..
```
self.predict = fluid.layers.sigmoid(b_linear + y_first_order + y_FM)
cost = fluid.layers.log_loss(input=self.predict, label=fluid.layers.cast(self.label,"float32")) # log_loss
avg_cost = fluid.layers.reduce_sum(cost)
self._cost = avg_cost
predict_2d = fluid.layers.concat([1 - self.predict, self.predict], 1)
label_int = fluid.layers.cast(self.label, 'int64')
auc_var, batch_auc_var, _ = fluid.layers.auc(input=predict_2d,
label=label_int,
slide_steps=0)
3. 退回fm目录中，打开文件config.yaml,更改其中的参数
将workspace改为您当前的绝对路径。（可用pwd命令获取绝对路径）
将train_sample中的batch_size从5改为1024
将train_sample中的data_path改为{workspace}/data/slot_train_data
将infer_sample中的batch_size从5改为1024
将infer_sample中的data_path改为{workspace}/data/slot_test_data
根据自己的需求调整phase中的线程数
4. 运行命令，模型会进行两个epoch的训练，然后预测第二个epoch，并获得相应auc指标
```
完成上述组网后，我们最终可以通过训练拿到`auc`指标。
python -m paddlerec.run -m ./config.yaml
```
5. 经过全量数据训练后，执行预测的结果示例如下：
```
PaddleRec: Runner train_runner Begin
Executor Mode: train
......@@ -291,3 +195,7 @@ I0921 13:11:33.123886 20932 parallel_executor.cc:375] Garbage collection strateg
Infer infer_phase of epoch 1 done, use time: 1764.81796193, global metrics: AUC=0.76193726
PaddleRec Finish
```
## 进阶使用
## FAQ
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或