#####带时间衰减因子#####应用实战: 如何利用Spark集群计算物品相似度

2024-05-07 15:08
文章标签 实战 集群 计算 应用 时间 物品 因子 衰减 spark 相似 #####

本文主要是介绍#####带时间衰减因子#####应用实战: 如何利用Spark集群计算物品相似度，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

本文是Spark调研笔记的最后一篇，以代码实例说明如何借助Spark平台高效地实现推荐系统CF算法中的物品相似度计算。

在推荐系统中，最经典的推荐算法无疑是协同过滤（Collaborative Filtering, CF），而item-cf又是CF算法中一个实现简单且效果不错的算法。
在item-cf算法中，最关键的步骤是计算物品之间的相似度。本文以代码实例来说明如何利用Spark平台快速计算物品间的余弦相似度。
Cosine Similarity是相似度的一种常用度量，根据《推荐系统实践》一书第2.4.2节关于Item-CF算法部分的说明，其计算公式如下：

举个例子，若对item1有过行为的用户集合为{u1, u2, u3}，对item2有过行为的用户集合为{u1, u3, u4, u5}，则根据上面的式子，item1和item2间的相似度为2/(3*4)，其中分子的2是因为item1的user_list与item2的user_list的交集长度为2，即item1和item2的共现（co-occurence）次数是2。

在工程实现上，根据论文"Empirical Analysis of Predictive Algorithms for Collaborative Filtering"的分析，为对活跃用户做惩罚，引入了IUF (Inverse User Frequency)的概念（与TF-IDF算法引入IDF的思路类似：活跃用户对物品相似度的贡献应该小于不活跃的用户），因此，对余弦相似度做改进后相似度计算公式如下：

可以看到，上式分子部分的1/log(1 + N(u))体现了对活跃用户的惩罚。

此外，通常认为用户在相隔很短的时间内喜欢的物品具有更高相似度。因此，工程实现上，还会考虑时间衰减效应。一种典型的时间衰减函数如下所示：

最终，时间上下文相关的Item-CF算法中的相似度计算公式如下：

上式中，分母部分与标准的相似度公式分母保持一致；分子部分参与运算的是item_i和item_j的共现用户集合，其中，f(t)是时间衰减效应的体现，N(u)对活跃用户做了惩罚。

下面的Python代码是计算物品相似度的Spark任务的代码片段（从HDFS加载用户历史行为日志，计算物品相似度，相似列表取TopN，将相似度计算结果写会HDFS），供大家参考：

[python]  view plain  copy
  1. #!/bin/env/python  
  2.   
  3.   
  4. import pyspark as ps  
  5. import math  
  6. import datetime as dt  
  7. import util  
  8.   
  9.   
  10. def generate_item_pair(x):  
  11.     """ 
  12.         Find co-occurence items of every given user  
  13.         Return a tuple in the format of ((item_0, item_1), cooccurrence_factor). 
  14.     """  
  15.     items = x[1]  
  16.     item_cnt = len(items)  
  17.     alpha = 1  
  18.     for i in items:  
  19.         item1 = i[0]  
  20.         ts1   = i[1]  
  21.         for j in items:  
  22.             item2 = j[0]  
  23.             ts2   = j[1]  
  24.             if item1 != item2:  
  25.                 ## introduce time decay and penalize active users  
  26.                 ft = 1.0 / (1 + alpha * abs(ts1 - ts2))  
  27.                 yield ((item1, item2), (ft / math.log(1 + item_cnt)))  
  28.   
  29.   
  30. def compute_item_similarity(x):  
  31.     items = x[0]  
  32.     cooccurrence = float(x[1])  
  33.     item_dict = g_item_freq_d   
  34.     norm_factor = 5  
  35.     if items[0] in item_dict and items[1] in item_dict:  
  36.         freq_0 = item_dict[items[0]]  
  37.         freq_1 = item_dict[items[1]]  
  38.         ## calculate similarity between the item pair  
  39.         sim = cooccurrence / math.sqrt(freq_0 * freq_1)  
  40.         ## normalize similarity  
  41.         norm_sim = (cooccurrence / (cooccurrence + norm_factor)) * sim  
  42.         yield (items[0], (items[1], norm_sim))  
  43.   
  44.   
  45. def sort_items(x):  
  46.     """ 
  47.         For a given item, sort all items similar to it as descent (using similarity scores), take topN similar items, and return as the following format: 
  48.         given_item \t sorted_item_0$sorted_score_0,sorted_item_1$sorted_score_1,... 
  49.     """  
  50.     similar_items = list(x[1])  
  51.     if len(similar_items) > 0:  
  52.         ## sort list of (item, score) tuple by score from high to low  
  53.         similar_items.sort(key=lambda x: x[1], reverse=True)  
  54.         ## format the list of sorted items as a string  
  55.         similar_items_str = ",".join(["$".join(map(str,item)) for item in similar_items[0:50]])  
  56.         yield "\t".join([str(x[0]), similar_items_str])  
  57.   
  58.   
  59. def main():  
  60.     base_hdfs_uri = "hdfs://to/user/behavior/log"  
  61.     today = dt.date.today()  
  62.     knn_similarity_file = '%s/%s/knn_sim' % (base_hdfs_uri, today.strftime('%Y%m%d'))  
  63.   
  64.     sc = ps.SparkContext()  
  65.   
  66.     ## load user behavior from hdfs log  
  67.     ## each element in user_item is a tuple: (user, (item, timestamp))  
  68.     history_s = (today - dt.timedelta(8)).strftime('%Y%m%d')  
  69.     history_e = (today - dt.timedelta(2)).strftime('%Y%m%d')  
  70.     input_files = util.get_input_files(action='play', start=history_s, end=history_e)  
  71.     user_item = sc.textFile(",".join(input_files))\  
  72.         .mapPartitions(util.parse_user_item) \  
  73.         .map(lambda x: (x[0], (x[1], x[2]))) \  
  74.         .distinct() \  
  75.         .cache()  
  76.   
  77.     ## compute item frequency and store as a global dict  
  78.     item_freq = user_item.map(lambda x: (x[1][0], 1)) \  
  79.         .reduceByKey(lambda x, y: x + y) \  
  80.         .collect()  
  81.     global g_item_freq_d  
  82.     g_item_freq_d = dict()  
  83.     for x in item_freq:  
  84.         g_item_freq_d[x[0]] = x[1]  
  85.      
  86.     ## compute item similarity and find top n most similar items    
  87.     item_pair_sim = user_item.groupByKey() \  
  88.         .flatMap(generate_item_pair) \  
  89.         .reduceByKey(lambda x, y: x + y) \  
  90.         .flatMap(compute_item_similarity) \  
  91.         .groupByKey() \  
  92.         .flatMap(sort_items) \  
  93.         .cache()  
  94.   
  95.     ## dump to hdfs  
  96.     item_pair_sim.repartition(1).saveAsTextFile(knn_similarity_file)  
  97.   
  98.   
  99. if __name__ == '__main__':  
  100.     main()  

上面的代码中，import util中引入的util只是负责从HDFS获取用户历史日志的文件名列表，非常简单，实现细节这里不赘述。

【参考资料】
1. wikipedia: Collaborative filtering
2. 推荐系统实践（项亮著）第2.4.2节: 基于物品的协同过滤算法
3. Paper: Empirical Analysis of Predictive Algorithms for Collaborative Filtering
4. 推荐系统实践（项亮著）第5.1.6节: 时间上下文相关的ItemCF算法
 5.  Spark Programming Guide

========================== EOF ===========================

这篇关于#####带时间衰减因子#####应用实战: 如何利用Spark集群计算物品相似度的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/967700。

相关文章

Python在二进制文件中进行数据搜索的实战指南

Python在二进制文件中进行数据搜索的实战指南

《Python在二进制文件中进行数据搜索的实战指南》在二进制文件中搜索特定数据是编程中常见的任务,尤其在日志分析、程序调试和二进制数据处理中尤为重要,下面我们就来看看如何使用Python实现这一功能吧... 目录简介1. 二进制文件搜索概述2. python二进制模式文件读取（rb）2.1 二进制模式与文本
阅读更多...
Django调用外部Python程序的完整项目实战

Django调用外部Python程序的完整项目实战

《Django调用外部Python程序的完整项目实战》Django是一个强大的PythonWeb框架,它的设计理念简洁优雅,：本文主要介绍Django调用外部Python程序的完整项目实战,文中通... 目录一、为什么 Django 需要调用外部 python 程序二、三种常见的调用方式方式 1：直接 im
阅读更多...
线程池ThreadPoolExecutor应用过程

线程池ThreadPoolExecutor应用过程

《线程池ThreadPoolExecutor应用过程》：本文主要介绍如何使用ThreadPoolExecutor创建线程池,包括其构造方法、常用方法、参数校验以及如何选择合适的拒绝策略,文章还讨论... 目录ThreadPoolExecutor构造说明及常用方法为什么强制要求使用ThreadPoolExec
阅读更多...
SpringBoot整合 Quartz实现定时推送实战指南

SpringBoot整合 Quartz实现定时推送实战指南

《SpringBoot整合Quartz实现定时推送实战指南》文章介绍了SpringBoot中使用Quartz动态定时任务和任务持久化实现多条不确定结束时间并提前N分钟推送的方案,本文结合实例代码给大... 目录前言一、Quartz 是什么？1、核心定位：解决什么问题？2、Quartz 核心组件二、使用步骤1
阅读更多...
mysql_mcp_server部署及应用实践案例

mysql_mcp_server部署及应用实践案例

《mysql_mcp_server部署及应用实践案例》文章介绍了在CentOS7.5环境下部署MySQL_mcp_server的步骤,包括服务安装、配置和启动,还提供了一个基于Dify工作流的应用案例... 目录mysql_mcp_server部署及应用案例1. 服务安装1.1. 下载源码1.2. 创建独立
阅读更多...
SpringBoot整合AOP及使用案例实战

SpringBoot整合AOP及使用案例实战

《SpringBoot整合AOP及使用案例实战》本文详细介绍了SpringAOP中的切入点表达式,重点讲解了execution表达式的语法和用法,通过案例实战,展示了AOP的基本使用、结合自定义注解以... 目录一、 引入依赖二、切入点表达式详解三、案例实战1. AOP基本使用2. AOP结合自定义注解3.
阅读更多...
Java 队列Queue从原理到实战指南

Java 队列Queue从原理到实战指南

《Java队列Queue从原理到实战指南》本文介绍了Java中队列（Queue）的底层实现、常见方法及其区别,通过LinkedList和ArrayDeque的实现,以及循环队列的概念,展示了如何高效... 目录一、队列的认识队列的底层与集合框架常见的队列方法插入元素方法对比（add和offer）移除元素方法
阅读更多...
MySQL MHA集群详解(数据库高可用)

MySQL MHA集群详解(数据库高可用)

《MySQLMHA集群详解(数据库高可用)》MHA（MasterHighAvailability）是开源MySQL高可用管理工具,用于自动故障检测与转移,支持异步或半同步复制的MySQL主从架构,本... 目录mysql 高可用方案：MHA 详解与实战1. MHA 简介2. MHA 的组件组成（1）MHA
阅读更多...
Spring Boot基于 JWT 优化 Spring Security 无状态登录实战指南

Spring Boot基于 JWT 优化 Spring Security 无状态登录实战指南

《SpringBoot基于JWT优化SpringSecurity无状态登录实战指南》本文介绍如何使用JWT优化SpringSecurity实现无状态登录,提高接口安全性,并通过实际操作步骤... 目录Spring Boot 实战：基于 JWT 优化 Spring Security 无状态登录一、先搞懂：为什
阅读更多...
C++11中的包装器实战案例

C++11中的包装器实战案例

《C++11中的包装器实战案例》本文给大家介绍C++11中的包装器实战案例,本文结合实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧... 目录引言1.std::function1.1.什么是std::function1.2.核心用法1.2.1.包装普通函数1.2.
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2