BigData 复习笔记 03：推荐算法

发表于 2019-07-30 更新于 2021-06-18 分类于 Tech 阅读次数：

本文字数： 8.8k 阅读时长 ≈ 8 分钟

推荐系统

Chapter03. 推荐算法

推荐系统

个性化推荐系统，基于海量数据向用户建议商品

基于数据，利用算法，对用户和商品深度挖掘

常用推荐算法

基于人口的推荐

最容易实现

根据用户基本信息发现用户之间的相关程度，将相似用户喜欢的商品推荐给当前用户。

基本信息包括：（不包含用户行为信息）

性别
年龄
地域
……

基于内容的推荐

Content Based，初期最为广泛使用

根据商品的元信息，利用内容相关性，基于用户以往的喜好，给用户推荐相似的商品

引入 Item 属性的 CB 推荐

对当前 Item 进行内容分析，利用属性索引，在数据库中进行相关性计算，取出相关 Item 推荐给 User

优点
- 提升推荐结果相关性
- 结果可解释
- 推荐结果易被用户感知
缺点
- 无个性化
- 依赖对 Item 的深度分析

引入 User 属性的 CB 推荐

对当前 Item 进行内容分析，利用属性索引，结合 User 行为数据，在数据库中进行相关性计算，取出相关 Item 推荐给 User

优点
- 用户模型刻画了用户兴趣需求
- 推荐形式多样，具有个性化
- 结果可解释
缺点
- 推荐精度低
- 马太效应
  - 解决：试探，加入随机推荐位
- 用户行为稀疏导致覆盖率低

不同场景选取不同推荐方式

【Demo】倒排表

对 Item 进行挖掘，得出 Token -> ItemA:ScoreA, ItemB:ScoreB, … 倒排表

对 Item 中文分词，得出正排表

fenci.py

import os
import sys

os.system('tar xvzf jieba.tar.gz > /dev/null')
sys.path.append("./")

import jieba
import jieba.posseg
import jieba.analyse

for line in sys.stdin:
    ss = line.strip().split('\t')
    if len(ss) != 2:
        continue
    music_id, music_name = ss
    tmp_list = []
    for x, w in jieba.analyse.extract_tags(music_name, withWeight=True):
        tmp_list.append((x, float(w)))
    final_list = sorted(tmp_list, key=lambda x: x[1], reverse=True)

    print('\t'.join([music_id, music_name, ','.join([':'.join([t_w[0], str(t_w[1])]) for t_w in final_list])]))

转置

map_inverted.py

import sys

for line in sys.stdin:
    ss = line.strip().split('\t')
    if len(ss) != 3:
        continue
    music_id = ss[0].strip()
    music_name = ss[1].strip()
    music_token_list = ss[2].strip()

    for feature in music_token_list.split(','):
        ss = feature.strip().split(':')
        token = ss[0].strip()
        weight = ss[1].strip()
        print('\t'.join([token, music_name, weight]))

得到倒排表

red_inverted.py

import sys

cur_token = None
m_list = []

for line in sys.stdin:
    ss = line.strip().split('\t')
    if len(ss) != 3:
        continue
    token = ss[0].strip()
    name = ss[1].strip()
    weight = float(ss[2].strip())

    if cur_token is None:
        cur_token = token
    if cur_token != token:
        final_list = sorted(m_list, key=lambda x: x[1], reverse=True)
        print('\t'.join([cur_token, ','.join([':'.join([name_weight[0], str(name_weight[1])]) for name_weight in final_list])]))
        cur_token = token
        m_list = []

    m_list.append((name, weight))

final_list = sorted(m_list, key=lambda x: x[1], reverse=True)
print('\t'.join([cur_token.strip(), ','.join([':'.join([name_weight[0], str(name_weight[1])]) for name_weight in final_list])]))

【Hadoop Streaming 脚本】

#!/usr/bin/env bash

HADOOP_CMD="/usr/local/src/hadoop-2.8.5/bin/hadoop"
STREAM_JAR_PATH="/usr/local/src/hadoop-2.8.5/share/hadoop/tools/lib/hadoop-streaming-2.8.5.jar"

INPUT_FILE_PATH="/chapter03/01_inverted/music_meta.txt.small"
OUTPUT_PATH_FENCI="/chapter03/01_inverted/output/python3/fenci"
OUTPUT_PATH="/chapter03/01_inverted/output/python3/result"
LOCAL_FILE_PATH="/mnt/hgfs/Code/chapter03/01_inverted/music_meta.txt.small"
UPLOAD_PATH="/chapter03/01_inverted"


${HADOOP_CMD} fs -rm -r -skipTrash ${INPUT_FILE_PATH}
${HADOOP_CMD} fs -rm -r -skipTrash ${OUTPUT_PATH} ${OUTPUT_PATH_FENCI}
${HADOOP_CMD} fs -mkdir -p ${UPLOAD_PATH}
${HADOOP_CMD} fs -put ${LOCAL_FILE_PATH} ${UPLOAD_PATH}

# Step 1.
${HADOOP_CMD} jar ${STREAM_JAR_PATH} \
    -D mapreduce.job.reduces=0 \
    -D mapreduce.map.memory.mb=4096 \
    -D mapreduce.job.name=jieba_fenci_demo \
    -files map.py,jieba.tar.gz \
    -input ${INPUT_FILE_PATH} \
    -output ${OUTPUT_PATH_FENCI} \
    -mapper "python map.py" \


# Step 2.
${HADOOP_CMD} jar ${STREAM_JAR_PATH} \
    -D mapreduce.job.reduces=2 \
    -D mapreduce.job.name=demo_inverted \
    -files map_inverted.py,red_inverted.py \
    -input ${OUTPUT_PATH_FENCI} \
    -output ${OUTPUT_PATH} \
    -mapper "python map_inverted.py" \
    -reducer "python red_inverted.py" \

基于协同的推荐

从 User 行为日志中挖掘，User 与 Item 关联构成 UI 矩阵 or IU 矩阵

可使用 UI * IU 得到 UU 矩阵，使用 IU * UI 得到 II 矩阵

优点
- 充分利用群体智慧
- 推荐精度高于 CB
- 利于挖掘隐含的相关性
缺点
- 推荐结果解释性较差
- 对时效性强的 Item 不适用
- 冷启动问题

User-Based CF

假设：

用户喜欢那些跟他有相似爱好的用户喜欢的东西

具有相似兴趣的用户在未来也具有相似兴趣

Item-Based CF

假设：

用户喜欢跟他过去喜欢的物品相似的物品

历史上相似的物品未来也相似

	User-Based	Item-Based
性能	适用于用户较少场合，如果用户多，计算 UU 矩阵代价太大	适用于物品数明显小于用户数的场合，物品多。计算 II 矩阵代价太大
领域	时效性强，用户个性化兴趣不太明显的领域	长尾物品丰富，用户个性化需求强烈的领域
实时性	用户有新行为，不一定造成推荐结果立即变化	用户有新行为，一定导致推荐结果实时变化
冷启动	在新用户对很少物品产生行为后，不能立即进行个性化推荐，UU 矩阵需要更新。新物品上线后，一旦有用户对物品产生行为，就可以将新物品推荐给兴趣相似的其他用户	新用户只要对一个物品产生行为，就可以给他推荐和该物品相关的其他物品。在没有更新 II 矩阵的情况下无法将新物品推荐给用户
推荐理由	很难提供令用户信服的推荐解释	利用用户历史行为给用户做推荐解释，可以令用户比较信服

协同推荐实现方案（以 Item-Based 为例）

倒排式

相似度计算公式
处理数据

格式 {user, item, score}

构建 II 矩阵

归一化矩阵

转置 UI 矩阵 -> IU 矩阵

1_gen_ui_map.py

import sys

for line in sys.stdin:
    ss = line.strip().split('\t')
    if len(ss) != 3:
        continue
    user, item, score = ss
    print('%s\t%s\t%s' % (item, user, score))

归一化

1_gen_ui_red.py

import sys
import math

cur_item = None
user_score_list = []

for line in sys.stdin:
    item, user, score = line.strip().split('\t')
    if cur_item is None:
        cur_item = item
    if cur_item != item:
        sum = 0.0
        for kv in user_score_list:
            (u, s) = kv
            sum += pow(s, 2)
        sum = math.sqrt(sum)
        for kv in user_score_list:
            (u, s) = kv
            print('%s\t%s\t%s' % (u, cur_item, float(s / sum)))
        user_score_list = []
        cur_item = item

    user_score_list.append((user, float(score)))
sum = 0.0
for kv in user_score_list:
    (u, s) = kv
    sum += pow(s, 2)
sum = math.sqrt(sum)
for kv in user_score_list:
    (u, s) = kv
    print('%s\t%s\t%s' % (u, cur_item, float(s / sum)))

生成 II-pair 对

标准输入输出

2_gen_ii_pair_map.py

import sys

for line in sys.stdin:
    u, i, s = line.strip().split('\t')
    print("%s\t%s\t%s" % (u, i, s))

生成 {item, item, score}

2_gen_ii_pair_red.py

import sys

cur_user = None
item_score_list = []

for line in sys.stdin:
    user, item, score = line.strip().split('\t')
    if cur_user is None:
        cur_user = user
    if cur_user != user:
        for i in range(1, len(item_score_list) - 1):
            for j in range(i + 1, len(item_score_list)):
                item_a, score_a = item_score_list[i]
                item_b, score_b = item_score_list[j]
                print('%s\t%s\t%s' % (item_a, item_b, score_a * score_b))
                print('%s\t%s\t%s' % (item_b, item_a, score_a * score_b))
        item_score_list = []
        cur_user = user
    item_score_list.append((item, float(score)))

for i in range(1, len(item_score_list) - 1):
    for j in range(i + 1, len(item_score_list)):
        item_a, score_a = item_score_list[i]
        item_b, score_b = item_score_list[j]
        print('%s\t%s\t%s' % (item_a, item_b, score_a * score_b))
        print('%s\t%s\t%s' % (item_b, item_a, score_a * score_b))

求和

以 ii 为 key 做 partition

3_sum_map.py

import sys

for line in sys.stdin:
    item_a, item_b, score = line.strip().split('\t')
    print("%s\t%s" % (item_a + "," + item_b, score))

求和

3_sum_red.py

import sys

cur_ii_pair = None
sum = 0.0

for line in sys.stdin:
    ii_pair, score = line.strip().split('\t')
    if cur_ii_pair is None:
        cur_ii_pair = ii_pair
    if cur_ii_pair != ii_pair:
        ii = ii_pair.strip().split(',')
        if len(ii) != 2:
            continue
        item_a, item_b = ii
        print('%s\t%s\t%s' % (item_a, item_b, sum))
        cur_ii_pair = ii_pair
        sum = 0.0
    sum += float(score)
ii = cur_ii_pair.strip().split(',')
if len(ii) != 2:
    sys.exit()
item_a, item_b = ii
print('%s\t%s\t%s' % (item_a, item_b, sum))

【Hadoop Streaming 脚本】

#!/usr/bin/env bash

HADOOP_CMD="/usr/local/src/hadoop-2.8.5/bin/hadoop"
STREAM_JAR_PATH="/usr/local/src/hadoop-2.8.5/share/hadoop/tools/lib/hadoop-streaming-2.8.5.jar"

INPUT_FILE_PATH="/chapter03/02_cf/music_uis.data"
OUTPUT_PATH_1="/chapter03/02_cf/item_based/output/python3/step1"
OUTPUT_PATH_2="/chapter03/02_cf/item_based/output/python3/step2"
OUTPUT_PATH_3="/chapter03/02_cf/item_based/output/python3/step3"
LOCAL_FILE_PATH="/mnt/hgfs/Code/chapter03/02_cf/music_uis.data"
UPLOAD_PATH="/chapter03/02_cf"


${HADOOP_CMD} fs -rm -r -skipTrash ${INPUT_FILE_PATH}
${HADOOP_CMD} fs -rm -r -skipTrash ${OUTPUT_PATH_1} ${OUTPUT_PATH_2} ${OUTPUT_PATH_3}
${HADOOP_CMD} fs -mkdir -p ${UPLOAD_PATH}
${HADOOP_CMD} fs -put ${LOCAL_FILE_PATH} ${UPLOAD_PATH}

# Step 1.
${HADOOP_CMD} jar ${STREAM_JAR_PATH} \
    -D mapreduce.job.name=step1 \
    -files 1_gen_ui_map.py,1_gen_ui_red.py \
    -input ${INPUT_FILE_PATH} \
    -output ${OUTPUT_PATH_1} \
    -mapper "python 1_gen_ui_map.py" \
    -reducer "python 1_gen_ui_red.py"

# Step 2.
${HADOOP_CMD} jar ${STREAM_JAR_PATH} \
    -D mapreduce.job.name=step2 \
    -files 2_gen_ii_pair_map.py,2_gen_ii_pair_red.py \
    -input ${OUTPUT_PATH_1} \
    -output ${OUTPUT_PATH_2} \
    -mapper "python 2_gen_ii_pair_map.py" \
    -reducer "python 2_gen_ii_pair_red.py" \

# Step 3.
${HADOOP_CMD} jar ${STREAM_JAR_PATH} \
    -D mapreduce.job.name=step3 \
    -files 3_sum_map.py,3_sum_red.py \
    -input ${OUTPUT_PATH_2} \
    -output ${OUTPUT_PATH_3} \
    -mapper "python 3_sum_map.py" \
    -reducer "python 3_sum_red.py" \

冷启动问题

用户冷启动

提供热门排行榜，等用户数据收集到一定程度再切换到个性化推荐
利用用户注册时提供的年龄、性别等数据做粗粒度的个性化
利用用户社交网络账号，导入用户在社交网站上的好友信息，然后给用户推荐其好友喜欢的物品
在用户新登录时要求其对一些物品进行反馈，收集这些兴趣信息，然后给用户推荐相似的物品

物品冷启动

把新物品推荐给可能对他感兴趣的用户，利用内容信息，将物品推荐给喜欢过和与其相似的物品的用户
物品必须能够在第一时间展现给用户，否则一段时间过后，物品的价值就大大降低了
User-based 和 Item-based 都行不通时，只能利用 Content-Based 解决该问题，频繁更新相关性数据

系统冷启动

引入专家知识，通过一定的高效方式迅速建立起物品的相关性矩阵