构造 TFRecord 训练集

和前面的 LR 离线模型一样,FTRL 模型首先也是要完成训练集的构建。在上篇文章中,我们已经知道,可以通过读取用户历史行为数据,及文章特征和用户特征,构建出训练集 train,其中包括 features 和 label 两列数据,features 是文章特征和用户特征的组合。在 TensorFlow 通常使用 TFRecord 文件进行数据的存取。接下来,我们就要将 train 保存到 TFRecord 文件中。首先开启会话,将 train 中的特征和标签分别传入 write_to_tfrecords() 方法,并利用多线程执行

1
2
3
4
5
6
7
8
9
10
11
12
import tensorflow as tf

with tf.Session() as sess:
# 创建线程协调器
coord = tf.train.Coordinator()
# 开启子线程去读取数据
threads = tf.train.start_queue_runners(sess=sess, coord=coord)
# 存入数据
write_to_tfrecords(train.iloc[:, 0], train.iloc[:, 1])
# 关闭子线程,回收
coord.request_stop()
coord.join(threads)

接着,在 write_to_tfrecords() 方法中,遍历训练集数据,将每个样本构造为 tf.train.Example,其中 feature 为 BytesList 类型,label 为 Int64List 类型,并保存到 TFRecords 文件中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
def write_to_tfrecords(feature_batch, click_batch):
"""将用户与文章的点击日志构造的样本写入TFRecords文件
"""
# 1、构造tfrecords的存储实例
writer = tf.python_io.TFRecordWriter("./train_ctr_20190605.tfrecords")

# 2、循环将所有样本一个个封装成example,写入文件
for i in range(len(click_batch)):
# 取出第i个样本的特征值和目标值,格式转换
click = click_batch[i]
feature = feature_batch[i].tostring()
# 构造example
example = tf.train.Example(features=tf.train.Features(feature={
"feature": tf.train.Feature(bytes_list=tf.train.BytesList(value=[feature])),
"label": tf.train.Feature(int64_list=tf.train.Int64List(value=[click]))
}))
# 序列化example,写入文件
writer.write(example.SerializeToString())

writer.close()

离线训练

FTRL(Follow The Regularized Leader)模型是一种获得稀疏模型的优化方法,我们利用构建好的 TFRecord 样本数据对 FTRL 模型进行离线训练。首先,定义 read_ctr_records() 方法来读取 TFRecord 文件,并通过调用 parse_tfrecords() 方法遍历解析每个样本,并设置了批大小和迭代次数

1
2
3
4
5
def read_ctr_records():
train = tf.data.TFRecordDataset(["./train_ctr_20190605.tfrecords"])
train = train.map(parse_tfrecords)
train = train.batch(64)
train = train.repeat(10000)

解析每个样本,将 TFRecord 中序列化的 feature 列,解析成 channel_id (1), article_vector (100), user_weights (10), article_weights (10)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
FEATURE_COLUMNS = ['channel_id', 'article_vector', 'user_weigths', 'article_weights']

def parse_tfrecords(example):
features = {
"feature": tf.FixedLenFeature([], tf.string),
"label": tf.FixedLenFeature([], tf.int64)
}
parsed_features = tf.parse_single_example(example, features)
feature = tf.decode_raw(parsed_features['feature'], tf.float64)
feature = tf.reshape(tf.cast(feature, tf.float32), [1, 121])

channel_id = tf.cast(tf.slice(feature, [0, 0], [1, 1]), tf.int32)
article_vector = tf.reduce_sum(tf.slice(feature, [0, 1], [1, 100]), axis=1)
user_weights = tf.reduce_sum(tf.slice(feature, [0, 101], [1, 10]), axis=1)
article_weights = tf.reduce_sum(tf.slice(feature, [0, 111], [1, 10]), axis=1)

label = tf.cast(parsed_features['label'], tf.float32)

# 构造字典 名称-tensor
tensor_list = [channel_id, article_vector, user_weights, article_weights]
feature_dict = dict(zip(FEATURE_COLUMNS, tensor_list))

return feature_dict, label

指定输入特征的数据类型,并定义 FTRL 模型 model

1
2
3
4
5
6
7
8
9
10
11
12
13
# 定义离散类型特征
article_id = tf.feature_column.categorical_column_with_identity('channel_id', num_buckets=25)
# 定义连续类型特征
article_vector = tf.feature_column.numeric_column('article_vector')
user_weigths = tf.feature_column.numeric_column('user_weigths')
article_weights = tf.feature_column.numeric_column('article_weights')

feature_columns = [article_id, article_vector, user_weigths, article_weights]

model = tf.estimator.LinearClassifier(feature_columns=feature_columns,
optimizer=tf.train.FtrlOptimizer(learning_rate=0.1,
l1_regularization_strength=10,
l2_regularization_strength=10))

通过调用 read_ctr_records() 方法,来读取 TFRecod 文件中的训练数据,并设置训练步长,对定义好的 FTRL 模型进行训练及预估

1
2
model.train(read_ctr_records, steps=1000)
result = model.evaluate(read_ctr_records)

通常需要编写离线任务,定时读取用户行为数据作为训练集和验证集,对训练集及验证集进行 CTR 预估,并根据离线指标对结果进行分析,决定是否更新模型。

在线排序

通常在线排序是根据用户实时的推荐请求,对召回结果进行 CTR 预估,进而计算出排序结果并返回。我们需要根据召回结果构造测试集,其中每个测试样本包括用户特征和文章特征。首先,根据用户 ID 和频道 ID 读取用户特征(用户在每个频道的特征不同,所以是分频道存储的)

1
2
3
4
5
6
try:
user_feature = eval(hbu.get_table_row('ctr_feature_user',
'{}'.format(temp.user_id).encode(),
'channel:{}'.format(temp.channel_id).encode()))
except Exception as e:
user_feature = []

再根据用户 ID 读取召回结果

1
2
3
recall_set = read_hbase_recall('cb_recall', 
'recall:user:{}'.format(temp.user_id).encode(),
'als:{}'.format(temp.channel_id).encode())

接着,遍历召回结果,获取文章特征,并将用户特征合并,作为测试样本

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
test = []
for article_id in recall_set:
try:
article_feature = eval(hbu.get_table_row('ctr_feature_article',
'{}'.format(article_id).encode(),
'article:{}'.format(article_id).encode()))
except Exception as e:
article_feature = []

if not article_feature:
article_feature = [0.0] * 111
feature = []
feature.extend(user_feature)
feature.extend(article_feature)

test.append(f)

加载本地 FTRL 模型并对测试样本进行 CTR 预估

1
2
3
4
5
6
test_array = np.array(test)
model.load_weights('/root/toutiao_project/reco_sys/offline/models/ckpt/ctr_lr_ftrl.h5')
init = tf.global_variables_initializer()
with tf.Session() as sess:
sess.run(init)
predictions = self.model.predict(sess.run(tf.constant(test_array)))

对结果进行排序并提取 CTR 最高的前 K 个文章,这样就得到了 FTRL 模型在线排序的结果。

1
2
3
4
5
6
7
8
9
res = pd.DataFrame(np.concatenate((np.array(recall_set).reshape(len(recall_set), 1), predictions),
axis=1), columns=['article_id', 'prob'])

res_sort = res.sort_values(by=['prob'], ascending=True)

# 排序后,只将排名在前100个文章ID作为推荐结果返回给用户
if len(res_sort) > 100:
recall_set = list(res_sort.iloc[:100, 0])
recall_set = list(res_sort.iloc[:, 0])

参考

https://www.bilibili.com/video/av68356229
https://pan.baidu.com/s/1-uvGJ-mEskjhtaial0Xmgw(学习资源已保存至网盘, 提取码:eakp)


【技术服务】,详情点击查看: https://mp.weixin.qq.com/s/PtX9ukKRBmazAWARprGIAg


扫一扫 关注微信公众号!号主 专注于搜索和推荐系统,尝试使用算法去更好的服务于用户,包括但不局限于机器学习,深度学习,强化学习,自然语言理解,知识图谱,还不定时分享技术,资料,思考等文章!