#####带时间衰减因子#####应用实战: 如何利用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[0in item_dict and items[1in 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

相关文章

C语言中位操作的实际应用举例

《C语言中位操作的实际应用举例》:本文主要介绍C语言中位操作的实际应用,总结了位操作的使用场景,并指出了需要注意的问题,如可读性、平台依赖性和溢出风险,文中通过代码介绍的非常详细,需要的朋友可以参... 目录1. 嵌入式系统与硬件寄存器操作2. 网络协议解析3. 图像处理与颜色编码4. 高效处理布尔标志集合

Python的time模块一些常用功能(各种与时间相关的函数)

《Python的time模块一些常用功能(各种与时间相关的函数)》Python的time模块提供了各种与时间相关的函数,包括获取当前时间、处理时间间隔、执行时间测量等,:本文主要介绍Python的... 目录1. 获取当前时间2. 时间格式化3. 延时执行4. 时间戳运算5. 计算代码执行时间6. 转换为指

Java中字符串转时间与时间转字符串的操作详解

《Java中字符串转时间与时间转字符串的操作详解》Java的java.time包提供了强大的日期和时间处理功能,通过DateTimeFormatter可以轻松地在日期时间对象和字符串之间进行转换,下面... 目录一、字符串转时间(一)使用预定义格式(二)自定义格式二、时间转字符串(一)使用预定义格式(二)自

Java中的Lambda表达式及其应用小结

《Java中的Lambda表达式及其应用小结》Java中的Lambda表达式是一项极具创新性的特性,它使得Java代码更加简洁和高效,尤其是在集合操作和并行处理方面,:本文主要介绍Java中的La... 目录前言1. 什么是Lambda表达式?2. Lambda表达式的基本语法例子1:最简单的Lambda表

Python结合PyWebView库打造跨平台桌面应用

《Python结合PyWebView库打造跨平台桌面应用》随着Web技术的发展,将HTML/CSS/JavaScript与Python结合构建桌面应用成为可能,本文将系统讲解如何使用PyWebView... 目录一、技术原理与优势分析1.1 架构原理1.2 核心优势二、开发环境搭建2.1 安装依赖2.2 验

Java字符串操作技巧之语法、示例与应用场景分析

《Java字符串操作技巧之语法、示例与应用场景分析》在Java算法题和日常开发中,字符串处理是必备的核心技能,本文全面梳理Java中字符串的常用操作语法,结合代码示例、应用场景和避坑指南,可快速掌握字... 目录引言1. 基础操作1.1 创建字符串1.2 获取长度1.3 访问字符2. 字符串处理2.1 子字

Python列表去重的4种核心方法与实战指南详解

《Python列表去重的4种核心方法与实战指南详解》在Python开发中,处理列表数据时经常需要去除重复元素,本文将详细介绍4种最实用的列表去重方法,有需要的小伙伴可以根据自己的需要进行选择... 目录方法1:集合(set)去重法(最快速)方法2:顺序遍历法(保持顺序)方法3:副本删除法(原地修改)方法4:

在Spring Boot中浅尝内存泄漏的实战记录

《在SpringBoot中浅尝内存泄漏的实战记录》本文给大家分享在SpringBoot中浅尝内存泄漏的实战记录,结合实例代码给大家介绍的非常详细,感兴趣的朋友一起看看吧... 目录使用静态集合持有对象引用,阻止GC回收关键点:可执行代码:验证:1,运行程序(启动时添加JVM参数限制堆大小):2,访问 htt

SpringShell命令行之交互式Shell应用开发方式

《SpringShell命令行之交互式Shell应用开发方式》本文将深入探讨SpringShell的核心特性、实现方式及应用场景,帮助开发者掌握这一强大工具,具有很好的参考价值,希望对大家有所帮助,如... 目录引言一、Spring Shell概述二、创建命令类三、命令参数处理四、命令分组与帮助系统五、自定

SpringBoot应用中出现的Full GC问题的场景与解决

《SpringBoot应用中出现的FullGC问题的场景与解决》这篇文章主要为大家详细介绍了SpringBoot应用中出现的FullGC问题的场景与解决方法,文中的示例代码讲解详细,感兴趣的小伙伴可... 目录Full GC的原理与触发条件原理触发条件对Spring Boot应用的影响示例代码优化建议结论F