用户画像往往是大型网站的重要模块，基于用户画像不仅可以实现个性化推荐，还可以实现用户分群、精准推送、精准营销以及用户行为预测、商业化转化分析等，为商业决策提供数据支持。通常用户画像包括用户属性信息（性别、年龄、出生日期等）、用户行为信息（浏览、收藏、点赞等）以及环境信息（时间、地理位置等）。
在数据准备阶段，我们通过 Flume 已经可以将用户行为数据收集到 Hive 的 user_action 表的 HDFS 路径中，先来看一下这些数据长什么样子，我们读取当天的用户行为数据，注意读取之前要先关联分区
_day = time.strftime("%Y-%m-%d", time.localtime())
_localions = '/user/hive/warehouse/profile.db/user_action/' + _dayif fs.exists(_localions): # 如果有该文件直接关联，捕获关联重复异常 try:
self.spark.sql("alter table user_action add partition (dt='%s') location '%s'" % (_day, _localions))
except Exception as e:
pass self.spark.sql("use profile")
user_action = self.spark.sql("select actionTime, readTime, channelId, param.articleId, param.algorithmCombine, param.action, param.userId from user_action where dt>=" + _day)
user_action 结果如下所示
useraction
可以发现，上面的一条记录代表用户对文章的一次行为，但通常我们需要查询某个用户对某篇文章的所有行为，所以，我们要将这里用户对文章的多条行为数据合并为一条，其中包括用户对文章的所有行为。我们需要新建一个 Hive 表 user_article_basic，这张表包括了用户 ID、文章 ID、是否曝光、是否点击、阅读时间等等，随后我们将处理好的用户行为数据存储到此表中
create table user_article_basic
( user_id BIGINT comment "userID",
action_time STRING comment "user actions time",
article_id BIGINT comment "articleid",
channel_id INT comment "channel_id",
shared BOOLEAN comment "is shared",
clicked BOOLEAN comment "is clicked",
collected BOOLEAN comment "is collected",
exposure BOOLEAN comment "is exposured",
read_time STRING comment "reading time"
) COMMENT "user_article_basic"
CLUSTERED by (user_id) into 2 buckets
STORED as textfile
LOCATION '/user/hive/warehouse/profile.db/user_article_basic';
遍历每一条原始用户行为数据，判断用户对文章的行为，在 user_action_basic 中将该用户与该文章对应的行为设置为 True
if user_action.collect():
def _generate(row): _list = [] if row.action == 'exposure':
for article_id in eval(row.articleId):
# ["user_id", "action_time","article_id", "channel_id", "shared", "clicked", "collected", "exposure", "read_time"]
_list.Append( [row.userId, row.actionTime, article_id, row.channelId, False, False, False, True, row.readTime]) return _list
else:
class Temp(object):
shared = False clicked = False collected = False read_time = ""
_tp = Temp() if row.action == 'click':
_tp.clicked = True elif row.action == 'share':
_tp.shared = True elif row.action == 'collect':
_tp.collected = True elif row.action == 'read':
_tp.clicked = True _list.append( [row.userId, row.actionTime, int(row.articleId), row.channelId, _tp.shared, _tp.clicked, _tp.collected, True, row.readTime]) return _list
user_action_basic = user_action.rdd.flatMap(_generate) user_action_basic = user_action_basic.toDF( ["user_id", "action_time", "article_id", "channel_id", "shared", "clicked", "collected", "exposure",
"read_time"])
user_action_basic 结果如下所示，这里的一条记录包括了某个用户对某篇文章的所有行为
user_action_basic
由于 Hive 目前还不支持 pyspark 的原子性操作，所以 user_article_basic 表的用户行为数据只能全量更新（实际场景中可以选择其他语言或数据库实现）。这里，我们需要将当天的用户行为与 user_action_basic 的历史用户行为进行合并
old_data = uup.spark.sql("select * from user_article_basic")
new_data = old_data.unionAll(user_action_basic)
合并后又会产生一个新的问题，那就是用户 ID 和文章 ID 可能重复，因为今天某个用户对某篇文章的记录可能在历史数据中也存在，而 unionAll() 方法并没有去重，这里我们可以按照用户 ID 和文章 ID 进行分组，利用 max() 方法得到 action_time, channel_id, shared, clicked, collected, exposure, read_time 即可，去重后直接存储到 user_article_basic 表中
new_data.registerTempTable("temptable")
self.spark.sql('''insert overwrite table user_article_basic select user_id, max(action_time) as action_time,
article_id, max(channel_id) as channel_id, max(shared) as shared, max(clicked) as clicked,
max(collected) as collected, max(exposure) as exposure, max(read_time) as read_time from temptable
group by user_id, article_id''')
表 user_article_basic 结果如下所示
user_article_basic
我们选择将用户画像存储在 Hbase 中，因为 Hbase 支持原子性操作和快速读取，并且 Hive 也可以通过创建外部表关联到 Hbase，进行离线分析，如果要删除 Hive 外部表的话，对 Hbase 也没有影响。首先，在 Hbase 中创建用户画像表
create 'user_profile', 'basic','partial','env'
在 Hive 中创建 Hbase 外部表，注意字段类型设置为 map
create external table user_profile_hbase
( user_id STRING comment "userID",
information MAP<STRING, DOUBLE> comment "user basic information",
article_partial MAP<STRING, DOUBLE> comment "article partial",
env MAP<STRING, INT> comment "user env"
) COMMENT "user profile table"
STORED BY 'org.Apache.hadoop.hive.hbase.HBaseStorageHandler'
WITH SERDEPROPERTIES ("hbase.columns.mapping" = ":key,basic:,partial:,env:")
TBLPROPERTIES ("hbase.table.name" = "user_profile");
创建外部表之后，还需要导入一些依赖包
cp -r /root/bigdata/hbase/lib/hbase-*.jar /root/bigdata/spark/jars/
cp -r /root/bigdata/hive/lib/h*.jar /root/bigdata/spark/jars/
接下来，读取处理好的用户行为数据，由于日志中的 channel_id 有可能是来自于推荐频道（0），而不是文章真实的频道，所以这里要将 channel_id 列删除
spark.sql("use profile")
user_article_basic = spark.sql("select * from user_article_basic").drop('channel_id')
通过文章 ID，将用户行为数据与文章画像数据进行连接，从而得到文章频道 ID 和文章主题词
spark.sql('use article')
article_topic = spark.sql("select article_id, channel_id, topics from article_profile")
user_article_topic = user_article_basic.join(article_topic, how='left', on=['article_id'])
user_article_topic 结果如下图所示，其中 topics 列即为文章主题词列表，如 ['补码', '字符串', '李白', ...]
user_article_topic
接下来，我们需要计算每一个主题词对于用户的权重，所以需要将 topics 列中的每个主题词都拆分为单独的一条记录。可以利用 Spark 的 explode() 方法，达到类似“爆炸”的效果
import pyspark.sql.functions as F
user_article_topic = user_topic.withColumn('topic', F.explode('topics')).drop('topics')
user_article_topic 如下图所示
user_article_topic
我们通过用户对哪些文章发生了行为以及该文章有哪些主题词，计算出了用户对哪些主题词发生了行为。这样，我们就可以根据用户对主题词的行为来计算主题词对用户的权重，并且将这些主题词作为用户的标签。那么，用户标签权重的计算公式为：用户标签权重 =（用户行为分值之和）x 时间衰减。其中，时间衰减公式为：时间衰减系数 = 1 / (log(t) + 1)，其中 t 为发生行为的时间距离当前时间的大小
不同的用户行为对应不同的权重，如下所示
用户行为分值阅读时间(<1000)1阅读时间(>=1000)2收藏2分享3点击5
计算用户标签及权重，并存储到 Hbase 中 user_profile 表的 partial 列族中。注意，这里我们将频道 ID 和标签一起作为 partial 列族的键存储，这样我们就方便查询不同频道的标签及权重了
def compute_user_label_weights(partitions):
""" 计算用户标签权重
"""
action_weight = { "read_min": 1,
"read_middle": 2,
"collect": 2,
"share": 3,
"click": 5
} from datetime import datetime import numpy as np # 循环处理每个用户对应的每个主题词 for row in partitions:
# 计算时间衰减系数 t = datetime.now() - datetime.strptime(row.action_time, '%Y-%m-%d %H:%M:%S')
alpha = 1 / (np.log(t.days + 1) + 1)
if row.read_time == '':
read_t = 0
else:
read_t = int(row.read_time) # 计算阅读时间的行为分数 read_score = action_weight['read_middle'] if read_t > 1000 else action_weight['read_min']
# 计算各种行为的权重和并乘以时间衰减系数 weights = alpha * (row.shared * action_weight['share'] + row.clicked * action_weight['click'] +
row.collected * action_weight['collect'] + read_score)
# 更新到user_profilehbase表 with pool.connection() as conn: table = conn.table('user_profile')
table.put('user:{}'.format(row.user_id).encode(),
{'partial:{}:{}'.format(row.channel_id, row.topic).encode(): json.dumps(
weights).encode()}) conn.close()
user_topic.foreachPartition(compute_user_label_weights)
在 Hive 中查询用户标签及权重
hive> select * from user_profile_hbase limit 1;
OKuser:1 {"birthday":0.0,"gender":null} {"18:##":0.25704484358604845,"18:&#":0.25704484358604845,"18:+++":0.23934588700996243,"18:+++++":0.23934588700996243,"18:AAA":0.2747964402379244,"18:Animal":0.2747964402379244,"18:Author":0.2747964402379244,"18:BASE":0.23934588700996243,"18:BBQ":0.23934588700996243,"18:Blueprint":1.6487786414275463,"18:Code":0.23934588700996243,"18:DIR......
接下来，要将用户属性信息加入到用户画像中。读取用户基础信息，存储到用户画像表的 basic 列族即可
def update_user_info():
"""
更新用户画像的属性信息
:return:
"""
spark.sql("use toutiao")
user_basic = spark.sql("select user_id, gender, birthday from user_profile")
def udapte_user_basic(partition): import happybase # 用于读取hbase缓存结果配置 pool = happybase.ConnectionPool(size=10, host='172.17.0.134', port=9090)
for row in partition:
from datetime import date
age = 0
if row.birthday != 'null':
born = datetime.strptime(row.birthday, '%Y-%m-%d')
today = date.today()
age = today.year - born.year - ((today.month, today.day) < (born.month, born.day)) with pool.connection() as conn: table = conn.table('user_profile')
table.put('user:{}'.format(row.user_id).encode(),
{'basic:gender'.encode(): json.dumps(row.gender).encode()})
table.put('user:{}'.format(row.user_id).encode(),
{'basic:birthday'.encode(): json.dumps(age).encode()})
conn.close()
user_basic.foreachPartition(udapte_user_basic)
到这里，我们的用户画像就计算完成了。
定义更新用户画像方法，首先处理用户行为日志，拆分文章主题词，接着计算用户标签的权重，最后再将用户属性信息加入到用户画像中
def update_user_profile():
"""
定时更新用户画像的逻辑
:return:
"""
up = UpdateUserProfile() if up.update_user_action_basic():
up.update_user_label() up.update_user_info()
在 Apscheduler 中添加定时更新用户画像任务，设定每隔 2 个小时更新一次
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)
scheduler.start()
另外说一下，在实际场景中，用户画像往往是非常复杂的，下面是电商场景的用户画像，可以了解一下。
用户画像
京公网安备 11010802035086号