Spark機器學(xué)習(xí)實戰(zhàn)（三）電影評分數(shù)據(jù)處理與特征提取
這部分主要講了進行數(shù)據(jù)可視化之后如何進行必要的數(shù)據(jù)處理，原因是原始數(shù)據(jù)并非完整晌坤。隨后，我們要從數(shù)據(jù)中提取出我們需要的特征估盘。使用的數(shù)據(jù)集依然是MovieLens 100k數(shù)據(jù)集辅搬，平臺為Python Spark唯沮。
文章中列出了關(guān)鍵代碼，完整代碼見我的github repository堪遂，這篇文章的代碼在
chapter03/movielens_feature.py
第1步：數(shù)據(jù)處理與轉(zhuǎn)換
數(shù)據(jù)出現(xiàn)缺失或者異常時，常見的處理方法有：
- 過濾或刪除非規(guī)整或有缺失的數(shù)據(jù)
- 填充非規(guī)整或有缺失的數(shù)據(jù)
- 對異常值作魯棒處理
- 對可能的異常值進行轉(zhuǎn)換
由于我們采用的數(shù)據(jù)集數(shù)據(jù)缺失問題幾乎沒有萌庆，因此這部分不用特別處理溶褪。
第2步：特征提取
特征主要包含以下三種：
- 數(shù)值特征：比如年齡，可以直接作為數(shù)據(jù)的一個維度
- 類別特征：多個類別中的一種践险，但是類別特征一般有多少個類就會有多少個維度
- 文本特征：如電影評論
數(shù)值特征
數(shù)值特征也需要進行轉(zhuǎn)換猿妈，因為不是所有的數(shù)值特征都有意義。
比如年齡就是一個很好的數(shù)值特征巍虫，可以不加處理直接用彭则，因為年齡的增加與減少與目標有直接關(guān)系。然而占遥，如經(jīng)緯度的位置特征俯抖，有時就不太好直接用，需要做一些處理瓦胎，甚至可以轉(zhuǎn)換為類別特征芬萍。
類別特征
k類的類別特征需要轉(zhuǎn)換成一個k bits的向量
我們來對MovieLens數(shù)據(jù)集中的用戶職業(yè)進行處理，轉(zhuǎn)換為類別特征搔啊。
all_occupations = occupation_data.distinct().collect()
all_occupations.sort()
occupation_dict = {}
for i, occu in enumerate(all_occupations):
occupation_dict[occu] = i
user_tom_occupation = 'programmer'
tom_occupation_feature = np.zeros(len(all_occupations))
tom_occupation_feature[occupation_dict[user_tom_occupation]] = 1
print("Binary feature of tom's occupation (programmer) is:")
print(tom_occupation_feature)
結(jié)果為：
Binary feature of tom's occupation (programmer) is:
[ 0. 0. 0. 0. 0. 0. 0. 0. 0. 0. 0. 0. 0. 0. 1. 0. 0. 0.
0. 0. 0.]
派生特征
派生特征是指從原始數(shù)據(jù)經(jīng)過一些處理后得到的特征柬祠，如之前計算過的用戶打分電影總數(shù)，電影年齡等等负芋。
下面的例子漫蛔，是把u.data中的時間戳特征轉(zhuǎn)換為類別特征，表征這條評分是在一天中的什么時段給出的旧蛾。
rating_data = sc.textFile("%s/ml-100k/u.data" % PATH)
rating_fields = rating_data.map(lambda line: line.split('\t'))
timestamps = rating_fields.map(lambda fields: int(fields[3]))
hour_of_day = timestamps.map(lambda ts: datetime.fromtimestamp(ts).hour)
times_of_day_dict = {}
for hour in range(24):
if hour in range(7, 12):
times_of_day_dict[hour] = "morning"
elif hour in range(12, 14):
times_of_day_dict[hour] = "lunch"
elif hour in range(14, 18):
times_of_day_dict[hour] = "afternoon"
elif hour in range(18, 23):
times_of_day_dict[hour] = "evening"
else:
times_of_day_dict[hour] = "night"
time_of_day = hour_of_day.map(lambda hour: times_of_day_dict[hour])
print(hour_of_day.take(5))
print(time_of_day.take(5))
這段代碼的運行結(jié)果為
[23, 3, 15, 13, 13]
['night', 'night', 'afternoon', 'lunch', 'lunch']
可以看到莽龟，時間戳先被轉(zhuǎn)化為當天的小時點，隨后轉(zhuǎn)化為了時段蚜点，之后可以轉(zhuǎn)化為類別特征
文本特征
理論上來說轧房，文本特征也可以看作一個類別特征，然而文本很少出現(xiàn)重復(fù)绍绘，因此效果會很不理想奶镶。
下面用的是自然語言處理（NLP）常見的詞袋法（bag-of-word）迟赃，簡而言之，詞袋法就是把數(shù)據(jù)集中出現(xiàn)過的所有單詞構(gòu)成一個詞典厂镇，比如說有K個單詞纤壁。隨后以一個K維向量表示一段文字，文字中出現(xiàn)過的單詞記錄為1捺信，其余為0酌媒。由于大部分詞不會出現(xiàn)，因此很適合用稀疏矩陣表示迄靠。
首先我們用正則表達式去除電影標題中括號內(nèi)的年份信息秒咨，再把每個電影標題分解為單詞的列表。
def extract_title(raw):
grps = re.search("\((\w+)\)", raw)
if grps:
return raw[:grps.start()].strip()
else:
return raw
movie_data = sc.textFile("%s/ml-100k/u.item" % PATH)
movie_fields = movie_data.map(lambda line: line.split('|'))
raw_titles = movie_fields.map(lambda fields: fields[1])
print
print("Remove year information in '()'")
for raw_title in raw_titles.take(5):
print(extract_title(raw_title))
movie_titles = raw_titles.map(extract_title)
title_terms = movie_titles.map(lambda line: line.split(' '))
print
print("Split words.")
print(title_terms.take(5))
輸出為：
Remove year information in '()'
Toy Story
GoldenEye
Four Rooms
Get Shorty
Copycat
Split words.
[[u'Toy', u'Story'], [u'GoldenEye'], [u'Four', u'Rooms'], [u'Get', u'Shorty'], [u'Copycat']]
再利用flatMap RDD操作把所有出現(xiàn)過的單詞統(tǒng)計出來掌挚，構(gòu)建成單詞辭典雨席，形式為(單詞，編號)吠式。
all_terms = title_terms.flatMap(lambda x: x).distinct().collect()
all_terms_dict = {}
for i, term in enumerate(all_terms):
all_terms_dict[term] = i
print
print("Total number of terms: %d" % len(all_terms_dict))
最后把標題映射成一個高維的稀疏矩陣陡厘，出現(xiàn)過的單詞處為1。注意我們把詞典all_terms_dict作為一個廣播變量是因為這個變量會非常大特占，事先分發(fā)給每個計算節(jié)點會比較好糙置。
from scipy import sparse as sp
def create_vector(terms, term_dict):
num_terms = len(term_dict)
x = sp.csc_matrix((1, num_terms))
for t in terms:
if t in term_dict:
idx = term_dict[t]
x[0, idx] = 1
return x
all_terms_bcast = sc.broadcast(all_terms_dict)
term_vectors = title_terms.map(lambda
terms: create_vector(terms, all_terms_bcast.value))
print
print("The first five terms of converted sparse matrix of title")
print(term_vectors.take(5))
輸出為：
[<1x2645 sparse matrix of type '<type 'numpy.float64'>'
with 2 stored elements in Compressed Sparse Column format>,
..., <1x2645 sparse matrix of type '<type 'numpy.float64'>'
with 1 stored elements in Compressed Sparse Column format>]
正則化特征
通常我們獲得的特征需要進行一下正則化處理。正則化特征分為兩種：
第一種為正則化某一個特征是目，比如對數(shù)據(jù)集中的年齡進行正則化谤饭，使它們的平均值為0，方差為1
第二種為正則化特征向量胖笛，就是對某一個sample的特征進行正則化网持，使得它的范數(shù)為 1（常見為二階范數(shù)為1，二階范數(shù)是指平方和開根號）
例子是第二種长踊，正則化特征向量功舀。第一種方式是用numpy的函數(shù)。
np.random.seed(42)
x = np.random.randn(4)
norm_x = np.linalg.norm(x)
normalized_x = x / norm_x
print
print("x: %s" % x)
print("2-norm of x: %.4f" % norm_x)
print("normalized x: %s" % normalized_x)
輸出為：
x: [ 0.49671415 -0.1382643 0.64768854 1.52302986]
2-norm of x: 1.7335
normalized x: [ 0.28654116 -0.07976099 0.37363426 0.87859535]
第二種方式是用MLlib正則化特征向量
from pyspark.mllib.feature import Normalizer
normalizer = Normalizer()
vector = sc.parallelize([x])
normalized_x_mllib = normalizer.transform(vector).first().toArray()
print("MLlib normalized x: %s" % normalized_x)
結(jié)果自然是一樣的身弊，當然是選擇使用MLlib自帶函數(shù)更好了辟汰。
至此，這篇文章內(nèi)容就結(jié)束了阱佛。
