前面我们完成了文章画像和用户画像的构建,画像数据主要是提供给召回阶段的各种召回算法使用。接下来,我们还要为排序阶段的各种排序模型做数据准备,通过特征工程将画像数据进一步加工为特征数据,以供排序模型直接使用。

我们可以将特征数据存储到 Hbase 中,这里我们先在 Hbase 中创建好 ctr_feature_article 表 和 ctr_feature_user 表,分别存储文章特征数据和用户特征数据

1
2
3
4
5
6
7
-- 文章特征表
create 'ctr_feature_article', 'article'
-- 如 article:13401 timestamp=1555635749357, value=[18.0,0.08196639249252607,0.11217275332895373,0.1353835167902181,0.16086650318453152,0.16356418791892943,0.16740082750337945,0.18091837445730974,0.1907214431716628,0.2........................-0.04634634410271921,-0.06451843378804649,-0.021564142420785692,0.10212902152136256]

-- 用户特征表
create 'ctr_feature_user', 'channel'
-- 如 4 column=channel:13, timestamp=1555647172980, value=[]

构建文章特征

文章特征包括文章关键词权重、文章频道以及文章向量,我们首先读取文章画像

1
2
spark.sql("use article")
article_profile = spark.sql("select * from article_profile")

在文章画像中筛选出权重最高的 K 个关键词的权重,作为文章关键词的权重向量

1
2
3
4
5
6
7
def article_profile_to_feature(row):
try:
article_weights = sorted(row.keywords.values())[:10]
except Exception as e:
article_weights = [0.0] * 10
return row.article_id, row.channel_id, article_weights
article_profile = article_profile.rdd.map(article_profile_to_feature).toDF(['article_id', 'channel_id', 'weights'])

article_profile 结果如下所示,weights 即为文章关键词的权重向量

接下来,读取文章向量信息,再将频道 ID 和文章向量加入进来,利用 article_id 将 article_profile 和 article_vector 进行内连接,并将 weights 和 articlevector 转为 vector 类型

1
2
3
4
5
6
7
8
9
10
article_vector = spark.sql("select * from article_vector")
article_feature = article_profile.join(article_vector, on=['article_id'], how='inner')

def feature_to_vector(row):

from pyspark.ml.linalg import Vectors

return row.article_id, row.channel_id, Vectors.dense(row.weights), Vectors.dense(row.articlevector)

article_feature = article_feature.rdd.map(feature_to_vector).toDF(['article_id', 'channel_id', 'weights', 'articlevector'])

最后,我们将 channel_id, weights, articlevector 合并为一列 features 即可(通常 channel_id 可以进行 one-hot 编码,我们这里先省略了)

1
2
3
4
from pyspark.ml.feature import VectorAssembler

columns = ['article_id', 'channel_id', 'weights', 'articlevector']
article_feature = VectorAssembler().setInputCols(columns[1:4]).setOutputCol("features").transform(article_feature)

article_feature 结果如下所示,features 就是我们准备好的文章特征

最后,将文章特征结果保存到 Hbase 中

1
2
3
4
5
6
7
8
9
10
def save_article_feature_to_hbase(partition):
import happybase
pool = happybase.ConnectionPool(size=10, host='hadoop-master')
with pool.connection() as conn:
table = conn.table('ctr_feature_article')
for row in partition:
table.put('{}'.format(row.article_id).encode(),
{'article:{}'.format(row.article_id).encode(): str(row.features).encode()})

article_feature.foreachPartition(save_article_feature_to_hbase)

构建用户特征

由于用户在不同频道的偏好差异较大,所以我们要计算用户在每个频道的特征。首先读取用户画像,将空值列删除

1
2
3
spark.sql("use profile")

user_profile_hbase = spark.sql("select user_id, information.birthday, information.gender, article_partial, env from user_profile_hbase")

user_profile_hbase 结果如下所示,其中 article_partial 为用户标签及权重,如 ([‘18:vars’: 0.2, ‘18: python’:0.2, …], [‘19:java’: 0.2, ‘19: javascript’:0.2, …], …) 表示某个用户在 18 号频道的标签包括 var、python 等,在 19 号频道的标签包括 java、javascript等。

由于 gender 和 birthday 两列空值较多,我们将这两列去除(实际场景中也可以根据数据情况选择填充)

1
2
# 去除空值列
user_profile_hbase = user_profile_hbase.drop('env', 'birthday', 'gender')

提取用户 ID,获取 user_id 列的内容中 : 后面的数值即为用户 ID

1
2
3
4
def get_user_id(row):
return int(row.user_id.split(":")[1]), row.article_partial

user_profile_hbase = user_profile_hbase.rdd.map(get_user_id)

user_profile_hbase 转为 DataFrame 类型

1
2
3
4
5
6
7
8
from pyspark.sql.types import *

_schema = StructType([
StructField("user_id", LongType()),
StructField("weights", MapType(StringType(), DoubleType()))
])

user_profile_hbase = spark.createDataFrame(user_profile_hbase, schema=_schema)

接着,将每个频道内权重最高的 K 个标签的权重作为用户标签权重向量

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def frature_preprocess(row):

from pyspark.ml.linalg import Vectors

user_weights = []
for i in range(1, 26):
try:
channel_weights = sorted([row.weights[key] for key in row.weights.keys() if key.split(':')[0] == str(i)])[:10]
user_weights.append(channel_weights)
except:
user_weights.append([0.0] * 10)
return row.user_id, user_weights

user_features = user_profile_hbase.rdd.map(frature_preprocess).collect()

user_features 就是我们计算好的用户特征,数据结构类似 (10, [[0.2, 2.1, …], [0.2, 2.1, …]], …),其中元组第一个元素 10 即为用户 ID,第二个元素是长度为 25 的用户频道标签权重列表,列表中每个元素是长度为 K 的用户标签权重列表,代表用户在某个频道下的标签权重向量。

最后,将用户特征结果保存到 Hbase,利用 Spark 的 batch() 方法,按频道批量存储用户特征

1
2
3
4
5
6
7
8
9
10
11
import happybase

# 批量插入Hbase数据库中
pool = happybase.ConnectionPool(size=10, host='hadoop-master', port=9090)
with pool.connection() as conn:
ctr_feature = conn.table('ctr_feature_user')
with ctr_feature.batch(transaction=True) as b:
for i in range(len(user_features)):
for j in range(25):
b.put("{}".format(res[i][0]).encode(),{"channel:{}".format(j+1).encode(): str(res[i][1][j]).encode()})
conn.close()

Apscheduler 定时更新

定义文章特征和用户特征的离线更新方法

1
2
3
4
5
6
7
8
def update_ctr_feature():
"""
更新文章特征和用户特征
:return:
"""
fp = FeaturePlatform()
fp.update_user_ctr_feature_to_hbase()
fp.update_article_ctr_feature_to_hbase()

在 Apscheduler 中添加定时更新文章特征和用户特征的任务,每隔 4 小时运行一次

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.executors.pool import ProcessPoolExecutor

# 创建scheduler,多进程执行
executors = {
'default': ProcessPoolExecutor(3)
}

scheduler = BlockingScheduler(executors=executors)

# 添加一个定时运行文章画像更新的任务, 每隔1个小时运行一次
scheduler.add_job(update_article_profile, trigger='interval', hours=1)
# 添加一个定时运行用户画像更新的任务, 每隔2个小时运行一次
scheduler.add_job(update_user_profile, trigger='interval', hours=2)
# 添加一个定时运行特征中心平台的任务,每隔4小时更新一次
scheduler.add_job(update_ctr_feature, trigger='interval', hours=4)

scheduler.start()

参考

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


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


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