下面用“|”字元來分隔各行資料。這將生成一個RDD,其中每一個記錄對應一個Python列表,各列表由使用者ID(user ID)、年齡(age)、性別(gender)、職業(occupation)和郵編(ZIP code)五個屬性構成。4之後再統計使用者、性別、職業和郵編的數目。這可通過如下程式碼實現。該資料集不大,故這裡並未快取它。

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 "Users: %d, genders: 
 
%d, occupations: %d, ZIP codes: %d"%(num_users,num_genders,num_occupations,num_zipcodes)

輸出結果：Users: 943, genders: 2, occupations: 21, ZIP codes: 795

畫出使用者的年齡分佈圖：

%matplotlib inline
import matplotlib.pyplot as plt
from matplotlib.pyplot import hist
ages = user_fields.map(lambda
 
 x: int(x[1])).collect()
hist(ages, bins=20, color='lightblue',normed=True)
fig = plt.gcf()
fig.set_size_inches(12,6)
plt.show()

畫出使用者的職業的分佈圖：

#畫出使用者的職業的分佈圖：

import numpy as np
count_by_occupation = user_fields.map(lambda fields: (fields[3],1)).reduceByKey(lambda x,y:x+y).collect()
print count_by_occupation
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)]
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, width, color='lightblue')
plt.xticks(rotation=30)
fig = plt.gcf()
fig.set_size_inches(12,6)
plt.show()

輸出結果：

[(u'administrator', 79), (u'retired', 14), (u'lawyer', 12), (u'none', 9), (u'student', 196), (u'technician', 27), (u'programmer', 66), (u'salesman', 12), (u'homemaker', 7), (u'executive', 32), (u'doctor', 7), (u'entertainment', 18), (u'marketing', 26), (u'writer', 45), (u'scientist', 31), (u'educator', 95), (u'healthcare', 16), (u'librarian', 51), (u'artist', 28), (u'other', 105), (u'engineer', 67)]

Spark對RDD提供了一個名為countByValue的便捷函式。它會計算RDD裡各不同值所分別出現的次數,並將其以Pythondict函式的形式(或是Scala、Java下的Map函式)返回給驅動程式:

count_by_occupation2 = user_fields.map(lambda fields: fields[3]).countByValue()
print "Map-reduce approach:"
print dict(count_by_occupation2)
print "========================" 
print "countByValue approach:"
print dict(count_by_occupation)

Map-reduce approach:
{u'administrator': 79, u'retired': 14, u'lawyer': 12, u'healthcare': 16, u'marketing': 26, u'executive': 32, u'scientist': 31, u'student': 196, u'technician': 27, u'librarian': 51, u'programmer': 66, u'salesman': 12, u'homemaker': 7, u'engineer': 67, u'none': 9, u'doctor': 7, u'writer': 45, u'entertainment': 18, u'other': 105, u'educator': 95, u'artist': 28}
========================
countByValue approach:
{u'administrator': 79, u'writer': 45, u'retired': 14, u'lawyer': 12, u'doctor': 7, u'marketing': 26, u'executive': 32, u'none': 9, u'entertainment': 18, u'healthcare': 16, u'scientist': 31, u'student': 196, u'educator': 95, u'technician': 27, u'librarian': 51, u'programmer': 66, u'artist': 28, u'salesman': 12, u'other': 105, u'homemaker': 7, u'engineer': 67}

2.2 探索電影資料

接下來了解下電影分類資料的特徵。如之前那樣,我們可以先簡單看一下某行記錄,然後再統計電影總數。

movie_data = sc.textFile("/Users/youwei.tan/ml-100k//u.item")
print movie_data.first()
num_movies = movie_data.count()
print 'Movies: %d' % num_movies

1|Toy Story (1995)|01-Jan-1995||http://us.imdb.com/M/title-exact?Toy%20Story%20(1995)|0|0|0|1|1|1|0|0|0|0|0|0|0|0|0|0|0|0|0
Movies: 1682

繪製電影的age分佈圖：

繪製電影年齡的分佈圖的方法和之前對使用者年齡和職業分佈的處理類似。電影年齡即其發行年份相對於現在過了多少年(在本資料中現在是1998年)。從下面的程式碼可以看到,電影資料中有些資料不規整,故需要一個函式來處理解析releasedate時可能的解析錯誤。這裡命名該函式為convert_year。

#畫出電影的age分佈圖：

def convert_year(x):
    try:
        return int(x[-4:])
    except:
        return 1900

movie_fields = movie_data.map(lambda lines:lines.split('|'))
years = movie_fields.map(lambda fields: fields[2]).map(lambda x: convert_year(x))
years_filtered = years.filter(lambda x: x!=1900)
print years_filtered.count()
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)
fig = plt.gcf()
fig.set_size_inches(12,6)
plt.show()

輸出結果：

1681

2.3 探索評分資料

#檢視資料記錄數量：

rating_data = sc.textFile('/Users/youwei.tan/ml-100k/u.data')
print rating_data.first()
num_ratings = rating_data.count()
print 'Ratings: %d'% num_ratings

輸出結果：

196	242	3	881250949
Ratings: 100000

接下來，我們做一些資料的基本統計,以及繪製評級值分佈的直方圖。動手吧 ：

#對資料進行一些基本的統計：

rating_data = rating_data.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)/num_ratings
median_rating = np.median(ratings.collect())
ratings_per_user = num_ratings/num_users;
ratings_per_movie = num_ratings/ num_movies
print 'Min rating: %d' %min_rating
print 'max rating: %d' % max_rating
print 'Average rating: %2.2f' %mean_rating
print 'Median rating: %d '%median_rating
print 'Average # of ratings per user: %2.2f'%ratings_per_user
print 'Average # of ratings per movie: %2.2f' % ratings_per_movie

輸出結果：

Min rating: 1
max rating: 5
Average rating: 3.00
Median rating: 4 
Average # of ratings per user: 106.00
Average # of ratings per movie: 59.00

從上述結果可以看到,最低的評級為1,而最大的評級為5。這並不意外,因為評級的範圍便是從1到5。 

Spark對RDD也提供一個名為states的函式。該函式包含一個數值變數用於做類似的統計:

ratings.stats() 

輸出結果：

(count: 100000, mean: 3.52986, stdev: 1.12566797076, max: 5.0, min: 1.0)

可以看出,使用者對電影的平均評級(mean)是3.5左右,而評級中位數(median)為4。這就能期待說評級的分佈稍傾向高點的得分。要驗證這點,可以建立一個評級值分佈的條形圖。具體做法和之前的類似:

count_by_rating = ratings.countByValue()
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()
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)
fig = plt.gcf()
fig.set_size_inches(12,6)
plt.show()

輸出結果：

其特徵和我們之前所期待的相同,即評級的分佈的確偏向中等以上。 

接下來，計算每個使用者和其對應的評價次數：

#計算每個使用者和其對應的評價次數：

user_ratings_grouped = rating_data.map(lambda fields:(int(fields[0]),int(fields[2]))).groupByKey()
user_rating_byuser = user_ratings_grouped.map(lambda (k,v):(k,len(v)))
user_rating_byuser.take(5)

[(2, 62), (4, 24), (6, 211), (8, 59), (10, 184)]

繪製每個使用者的總共評價次數的分佈圖：