第一篇:使用Spark探索经典数据集MovieLens
前言
MovieLens数据集包含多个用户对多部电影的评级数据,也包括电影元数据信息和用户属性信息。
这个数据集经常用来做推荐系统,机器学习算法的测试数据集。尤其在推荐系统领域,很多著名论文都是基于这个数据集的。(PS: 它是某次具有历史意义的推荐系统竞赛所用的数据集)。
下载地址为:http://files.grouplens.org/datasets/movielens/,有好几种版本,对应不同数据量,可任君选用。
本文下载数据量最小的100k版本,对该数据集进行探索:
环境
本人机器所用的操作系统为号称国产操作系统的Ubuntu Kylin 14.04,美化后的界面还是蛮酷炫的:
spark版本为:v1.5.2,下面是集成了Ipython,pylab的python-shell:
初步预览
1. 首先是用户信息:
其中各列数据分别为:
用户id | 用户年龄 | 用户性别 | 用户职业 | 用户邮政编码
2. 然后是影片信息:
其中前几列数据分别为:
影片id | 影片名 | 影片发行日期 | 影片链接 | (后面几列先不去管)
3. 最后是评分数据:
用户id | 影片id | 评分值 | 时间戳(timestamp格式)
探索用户数据
1. 打开Spark的python-shell,执行以下代码载入数据集并打印首行记录:
# 载入数据集
user_data = sc.textFile("/home/kylin/ml-100k/u.user")
# 展示首行记录
user_data.first()
结果如下:
2. 分别统计用户,性别,职业的个数:
# 以' | '切分每列,返回新的用户RDD
user_fields = user_data.map(lambda line: line.split("|"))
# 统计用户数
num_users = user_fields.map(lambda fields: fields[0]).count()
# 统计性别数
num_genders = user_fields.map(lambda fields: fields[2]).distinct().count()
# 统计职业数
num_occupations = user_fields.map(lambda fields: fields[3]).distinct().count()
# 统计邮编数
num_zipcodes = user_fields.map(lambda fields: fields[4]).distinct().count()
# 返回结果
print "用户数: %d, 性别数: %d, 职业数: %d, 邮编数: %d" % (num_users, num_genders, num_occupations, num_zipcodes)
结果如下:
3. 查看年龄分布情况:
# 获取用户年龄RDD,并将其落地到驱动程序
ages = user_fields.map(lambda x: int(x[1])).collect()
# 绘制用户年龄直方图
hist(ages, bins=20, color='lightblue', normed=True)
结果如下:
18岁以上观看者人数激增,估计是“高考”完了时间多了?
20多岁的年轻人,我猜主要是大学生和刚工作不久的年轻人?观看者最多。
然后50岁的观看者也蛮多的,估计是快退休了,开始享受生活了。
4. 查看职业分布情况:
# 并行统计各职业人数的个数,返回职业统计RDD后落地
count_by_occupation = user_fields.map(lambda fields: (fields[3], 1)).reduceByKey(lambda x, y: x + y).collect() # 生成x/y坐标轴
x_axis1 = np.array([c[0] for c in count_by_occupation])
y_axis1 = np.array([c[1] for c in count_by_occupation])
x_axis = x_axis1[np.argsort(y_axis1)]
y_axis = y_axis1[np.argsort(y_axis1)] # 生成x轴标签
pos = np.arange(len(x_axis))
width = 1.0
ax = plt.axes()
ax.set_xticks(pos + (width / 2))
ax.set_xticklabels(x_axis) # 绘制职业人数条状图
plt.xticks(rotation=30)
plt.bar(pos, y_axis, width, color='lightblue')
值得注意的是,统计各职业人数的时候,是将不同职业名记录搜集到不同节点,然后开始并行统计。
结果如下:
果然,是学生,教育工作者观看影片的多。
不过程序猿观众也不少...... 医生是最少看电影的。
这里再给出一种统计各职业人数的解决方案:
count_by_occupation2 = user_fields.map(lambda fields: fields[3]).countByValue()
countByValue是Spark提供的便捷函数,它会自动统计每个Key下面的Value个数,并以字典的格式返回。
探索电影数据
1. 打开Spark的python-shell,执行以下代码载入数据集并打印首行记录:
# 载入数据集
movie_data = sc.textFile("/home/kylin/ml-100k/u.item")
# 展示首行记录
print movie_data.first()
结果如下:
2. 查看下有多少部电影吧:
num_movies = movie_data.count()
print num_movies
结果为:
3. 过滤掉没有发行时间信息的记录:
# 输入影片的发行时间字段,若非法则返回1900
def convert_year(x):
try:
return int(x[-4:])
except:
return 1900 # 以' | '切分每列,返回影片RDD
movie_fields = movie_data.map(lambda lines: lines.split("|"))
# 生成新的影片发行年份RDD,并将空/异常的年份置为1900,
years = movie_fields.map(lambda fields: fields[2]).map(lambda x: convert_year(x))
# 过滤掉影片发行年份RDD中空/异常的记录
years_filtered = years.filter(lambda x: x != 1900)
4. 统计影片的年龄分布:
# 生成影片年龄RDD,然后统计不同年龄的影片数并落地
movie_ages = years_filtered.map(lambda yr: 1998-yr).countByValue()
# 获得影片数
values = movie_ages.values()
# 获得年龄
bins = movie_ages.keys()
# 绘制电影年龄分布图
hist(values, bins=bins, color='lightblue', normed=True)
因为这份数据集比较老,1998年提供的,所以就按当时的年龄来统计吧。另外这次使用了countByValue来统计个数,而它是执行函数,不需要再collect了。
结果为:
看得出电影库中的电影大都还是比较新的。
探索评级数据
1. 打开Spark的python-shell,执行以下代码载入数据集并打印首行记录:
# 载入数据集
rating_data_raw = sc.textFile("/home/kylin/ml-100k/u.data")
# 展示首行记录
print rating_data_raw.first()
结果为:
2. 先看看有多少评分记录吧:
num_ratings = rating_data .count()
print num_ratings
结果为:。果然共有10万条记录,没下载错版本。
3. 统计最高评分,最低评分,平均评分,中位评分,平均每个用户的评分次数,平均每部影片被评分次数:
# 获取评分RDD
rating_data = rating_data_raw.map(lambda line: line.split("\t"))
ratings = rating_data.map(lambda fields: int(fields[2]))
# 计算最大/最小评分
max_rating = ratings.reduce(lambda x, y: max(x, y))
min_rating = ratings.reduce(lambda x, y: min(x, y))
# 计算平均/中位评分
mean_rating = ratings.reduce(lambda x, y: x + y) / float(num_ratings)
median_rating = np.median(ratings.collect())
# 计算每个观众/每部电影平均打分/被打分次数
ratings_per_user = num_ratings / num_users
ratings_per_movie = num_ratings / num_movies
# 输出结果
print "最低评分: %d" % min_rating
print "最高评分: %d" % max_rating
print "平均评分: %2.2f" % mean_rating
print "中位评分: %d" % median_rating
print "平均每个用户打分(次数): %2.2f" % ratings_per_user
print "平均每部电影评分(次数): %2.2f" % ratings_per_movie
结果为:
另外Spark有个挺实用的统计函数stats可直接获取常用的统计信息,类似R语言的summary函数:
ratings.stats()
结果为:
4. 统计评分分布:
# 生成评分统计RDD,并落地
count_by_rating = ratings.countByValue()
# 生成x/y坐标轴
x_axis = np.array(count_by_rating.keys())
y_axis = np.array([float(c) for c in count_by_rating.values()])
# 对人数做标准化
y_axis_normed = y_axis / y_axis.sum() # 生成x轴标签
pos = np.arange(len(x_axis))
width = 1.0
ax = plt.axes()
ax.set_xticks(pos + (width / 2))
ax.set_xticklabels(x_axis) # 绘制评分分布柱状图
plt.bar(pos, y_axis_normed, width, color='lightblue')
plt.xticks(rotation=30)
结果为:
评分分布看来也应该挺满足正态分布的。
5. 统计不同用户的评分次数:
# 首先将数据以用户id为Key分发到各个节点
user_ratings_grouped = rating_data.map(lambda fields: (int(fields[0]), int(fields[2]))).groupByKey()
# 然后统计每个节点元素的个数,也即每个用户的评论次数
user_ratings_byuser = user_ratings_grouped.map(lambda (k, v): (k, len(v)))
# 输出前5条记录
user_ratings_byuser.take(5)
注意到这次使用了groupyByKey方法,这个方法和reduceByKey功能有点相似,但是有区别。请读者自行百度。
结果为:
最新文章
- Vmware玩mac os x分享
- vim的一些配置
- [deviceone开发]-土地销售App开源
- AR , VR, GVR ...
- vIDC v2.0 强大的端口转发神器使用总结-开放内网tfs代码服务
- 优化后的二次测试Miller_Rabin素性测试算法
- 【WEB API项目实战干货系列】- API登录与身份验证(三)
- ssh连接慢的问题的解决?
- poj 2749 Building roads (二分+拆点+2-sat)
- Java Swing 日期控件(转载)
- leetcode【sql】 Delete Duplicate Emails
- linux链接
- 基于开源CA系统ejbca community 6.3.1.1构建私有CA管理数字证书
- Win8 &; WP8.1 轻型数据库
- Maven 通过maven对项目进行拆分、聚合(重点)
- oracle 定时 job
- Wireless Penetration Testing(7-11 chapter)
- P1996 约瑟夫问题
- network error:software caused connection abort
- 使用PXE+Kickstart无人值守安装系统