构造 TFRecord 训练集

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

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 文件中

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() 方法遍历解析每个样本，并设置了批大小和迭代次数

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)

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

# 定义离散类型特征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 模型进行训练及预估

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

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

在线排序

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

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 读取召回结果

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

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

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 预估

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 模型在线排序的结果。

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])

list对oracle结果集排序了_文章推荐系统系列之基于 FTRL模型的在线排序

构造 TFRecord 训练集

离线训练

在线排序

猜你喜欢

list对oracle结果集排序了_文章推荐系统系列之基于 FTRL模型的在线排序相关推荐

最新文章

热门文章