第一篇：使用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功能有點相似，但是有區別。請讀者自行百度。
       結果為：