spark机器学习练习代码

悠悠 2022-07-14 14:42 191阅读 0赞

\# coding: utf-8  
  
  
\# In\[1\]:  
  
  
user\_data=sc.textFile("spark机器学习/数据集/ml-100k/u.user")  
user\_data.first()  
  
  
  
  
\# In\[72\]:  
  
  
user\_fields=user\_data.map(lambda line:line.split("|"))  
print user\_fields.first()  
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, gender: %d, occcupation:%d, zip code: %d" %(num\_users, num\_genders, num\_occupations,num\_zipcodes )  
  
  
  
  
\# In\[10\]:  
  
  
get\_ipython().magic(u'pylab inline')  
ages=user\_fields.map( lambda x: int(x\[1\])).collect()  
hist( ages, bins=20, color='lightblue',normed=True)  
fig=matplotlib.pyplot.gcf()  
fig.set\_size\_inches(16,10)  
  
  
  
  
\# In\[22\]:  
  
  
count\_by\_occupation=user\_fields.map(lambda fields:(fields\[3\],1)).reduceByKey( lambda x,y:x+y).collect()  
x\_axis1=np.array(   \[c\[0\]\] for c in count\_by\_occupation  )  
y\_axis1=np.array(   \[c\[1\]\] for c in count\_by\_occupation  )  
  
  
  
  
  
  
  
  
  
  
\# In\[24\]:  
  
  
count\_by\_occupation = user\_fields.map(lambda fields: (fields\[3\], 1)).reduceByKey(lambda x, y: x + y).collect()  
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 = matplotlib.pyplot.gcf()  
fig.set\_size\_inches(16, 10)  
  
  
  
  
\# In\[25\]:  
  
  
movie\_data=sc.textFile(  "/spark机器学习/数据集/ml-100k/u.item")  
print movie\_data.first()  
num\_movie=movie\_data.count()  
print "movies: %d" % num\_movie  
  
  
  
  
\# In\[29\]:  
  
  
count\_by\_occupation  
  
  
  
  
\# In\[28\]:  
  
  
x\_axis1  
  
  
  
  
\# In\[30\]:  
  
  
y\_axis1  
  
  
  
  
\# In\[31\]:  
  
  
def convert\_year(x):  
    try:  
        return int(x\[-4:\])  
    except:  
        return 1900  
  
  
  
  
\# In\[37\]:  
  
  
movie\_data=sc.textFile(  "/spark机器学习/数据集/ml-100k/u.item")  
print movie\_data.first()  
num\_movie=movie\_data.count()  
print "movies: %d" % num\_movie  
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)  
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=matplotlib.pyplot.gcf()  
fig.set\_size\_inches(16,10)  
  
  
  
  
\# In\[54\]:  
  
  
rating\_data=sc.textFile("/spark机器学习/数据集/ml-100k/u.data")  
print rating\_data.first()  \#xx  xxx   评分  时间  
num\_ratings=rating\_data.count()  
print "Ratings: %d" % num\_ratings  
  
  
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\_movie   \#取每部电影的平均分  
  
  
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  
  
  
ratings.stats()    \#spark对rdd提供的  stats()函数，也可以进行统计.   
  
  
  
  
  
  
  
  
  
  
\# In\[55\]:  
  
  
\#以下生成评级值分布的条形图  
count\_by\_rating = ratings.countByValue()  \#求出每个评分的评分次数  
x\_axis = np.array(count\_by\_rating.keys())  \#以1,2,3,4,5这五个分数做为x坐标  
y\_axis = np.array(\[float(c) for c in count\_by\_rating.values()\]) \#求出五个分数每个分类的评分次数形成y轴值  
\#让y轴正则化，使它表示为百分比  
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 = matplotlib.pyplot.gcf()  
fig.set\_size\_inches(16, 10)  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
\# In\[56\]:  
  
  
\# 对评级以用户id为主键进行分组  
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)      
  
  
  
  
\# In\[58\]:  
  
  
\#绘制用户评级分布的直方图  
user\_ratings\_byuser\_local=user\_ratings\_byuser.map( lambda(k,v):v).collect()  
hist(user\_ratings\_byuser\_local,bins=200,color='lightblue',normed=True)  
fig=matplotlib.pyplot.gcf()  
fig.set\_size\_inches(16,10)  
  
  
  
  
\# In\[71\]:  
  
  
\#对发行日期有问题的数据采取填充策略,即用发行日期的中位数来填充问题数据  
\# 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  
years\_pre\_processed=movie\_fields.map(  lambda fields:fields\[2\]).map(lambda x: convert\_year(x)).collect()  
years\_pre\_processed\_array=np.array( years\_pre\_processed)  
\#首先计算发行年份的平均数和中位数。    将中位数作为非规整数据的发行年份  
mean\_year=np.mean( years\_pre\_processed\_array\[years\_pre\_processed\_array!=1900\])  
median\_year=np.median(years\_pre\_processed\_array\[years\_pre\_processed\_array!=1900\])  
print '中位数的年份为: %d' %median\_year    
print '平均数的年份为: %d' %mean\_year  
  
  
index\_bad\_data=np.where(  years\_pre\_processed\_array==1900)\[0\]\[0\]  
years\_pre\_processed\_array\[index\_bad\_data\]=median\_year      \#用中位数的年份来填充  
  
  
print 'index of 1900 after assigning median: %s' % np.where( years\_pre\_processed\_array==1900 )\[0\]  
  
  
  
  
  
  
\# In\[77\]:  
  
  
\#  用户数据结构userfields:     u'1', u'24', u'M', u'technician', u'85711'  
\#取回occupation的所有可能取值  
all\_occupations=user\_fields.map( lambda fields: fields\[3\]).distinct().collect()  
all\_occupations.sort()  
\#依次对各可能的职业分配序号  
idx=0  
all\_occupations\_dict=\{\}  
for o in all\_occupations:  
    all\_occupations\_dict\[o\]=idx  
    idx+=1  
print "doctor的出现次数:%d" %all\_occupations\_dict\['doctor'\]  
print "programmer的出现次数:%d" %all\_occupations\_dict\['programmer'\]  
print "输出所有的职业: %s" % all\_occupations  
  
  
\#创建一个长度和可能的职业数目相同的numpy数组，其各元素值为0.这可通过numpy的zeros函数实现  
K=len( all\_occupations\_dict)  
binary\_x=np.zeros( K  )  
k\_programmer=all\_occupations\_dict\['programmer'\]  
binary\_x\[k\_programmer\]=1  
print "二进制特征数组为: %s" % binary\_x  
print "特征数组长度为: %d" % K  
  
  
  
  
\# In\[92\]:  
  
  
def extract\_datetime( ts ):  
    import datetime  
    return datetime.datetime.fromtimestamp(ts)  
  
  
\#提取点钟数  
timestamps=rating\_data.map( lambda fields:int(fields\[3\]))  
hour\_of\_day=timestamps.map( lambda ts: extract\_datetime(ts).hour)  
hour\_of\_day.take(10)  
  
  
  
  
  
  
\# In\[94\]:  
  
  
\#将点钟数转划成一天中的不同时段  
def assign\_tod(hr):  
    times\_of\_day=\{  
        'morning' : range(7,12),  
        'lunch' : range(12,14),  
        'afternoon' : range(14,18),  
        'evening' : range( 18, 23),  
        'night' : range(23, 24) + range(0, 7)  
         
    \}  
    for  k, v in times\_of\_day.iteritems():  
        print '%d' %hr  
        if hr in v:  
            return k  
          
time\_of\_day=hour\_of\_day.map( lambda hr: assign\_tod(hr))  
time\_of\_day.take(10)  
  
  
  
  
\# In\[100\]:  
  
  
\#提取标题  
def extract\_title(raw):  
    import re  
    \#该表达式找括号间的非单词(数字)  
    grps=re.search( "\\((\\w+)\\)",raw)  
    if grps:  
        \#只选取标题部分，并删除最后的空白字符  
        return raw\[:grps.start()\].strip()  
    else:  
        return raw  
      
raw\_titles=movie\_fields.map( lambda fields:fields\[1\])  
\#测试前五个标题的提取  
for raw\_title in raw\_titles.take(5):  
    print extract\_title( raw\_title)  
      
\#处理原始标题，将标题转化为单词  
movie\_titles=raw\_titles.map( lambda m: extract\_title(m))  
\#下面用简单空白分词法将标题分词为词  
title\_terms=movie\_titles.map( lambda t: t.split(" "))  
\#输出前五组标题分词后的单词  
print title\_terms.take(5)  
  
  
\#以上的文本处理很简单，其它的处理如:  转小写，删除标点符号，特殊字符等，删除连接词和词干提取。  
  
  
  
  
  
  
\# In\[102\]:  
  
  
\#下面取回所有的可能的词，以构建军一个词到序号的映射字典  
all\_terms=title\_terms.flatMap( lambda x:x).distinct().collect()  
\#创建一个新的字典来保存词，并分配 k之1序号  
idx=0  
all\_terms\_dict=\{\}  
for term in all\_terms:  
    all\_terms\_dict\[term\]=idx  
    idx+=1  
  
  
print "单词总数:%d" % len( all\_terms\_dict )  
print "Rooms 的位置:%d" % all\_terms\_dict\['Rooms'\]  
  
  
  
  
\# In\[103\]:  
  
  
\#以上也可以通过spark的zipWithIndex函数来更高效得到相同的结果.   
all\_terms\_dict2=title\_terms.flatMap( lambda x: x).distinct().zipWithIndex().collectAsMap()  
print "单词总数:%d" % len( all\_terms\_dict2 )  
print "Rooms 的位置:%d" % all\_terms\_dict2\['Rooms'\]  
  
  
  
  
\# In\[110\]:  
  
  
\#创建一个函数，将一个词集合转换成一个稀疏向量的表示。  
def create\_vector( terms, term\_dict):  
    from scipy import sparse as sp  
    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  
  
  
\#对提取出的各个词的rdd的各记录都应用该函数  
all\_terms\_bcast=sc.broadcast( all\_terms\_dict)  
term\_vectors=title\_terms.map( lambda terms: create\_vector(terms, all\_terms\_bcast.value))  
term\_vectors.take(5)  
  
  
\# 提取出两个单词的标题对应的向量里也是2个非零元素  
  
  
  
  
\# In\[112\]:  
  
  
np.random.seed( 42 )  
x=np.random.randn(2)  
norm\_x\_2=np.linalg.norm(x)  
normalized\_x=x/norm\_x\_2  
print "x:\\n%s" % x  
print "2-Norm of x: %2.4f" % norm\_x\_2  
print "Normalized x:\\n%s" % normalized\_x  
print "2-Norm of normalized\_x: %2.4f" % np.linalg.norm(normalized\_x)  
  
  
  
  
\# In\[113\]:  
  
  
\#使用 mllib机器学习库完成与上面相同的功能  
from pyspark.mllib.feature import Normalizer  
normalizer=Normalizer()  
vector=sc.parallelize( \[x\])  
normalized\_x\_mllib=normalizer.transform( vector ).first().toArray()  
print "x:\\n%s" % x  
print "2-Norm of x: %2.4f" % norm\_x\_2  
print "Normalized x MLlib:\\n%s" % normalized\_x\_mllib  
print "2-Norm of normalized\_x: %2.4f" % np.linalg.norm(normalized\_x\_mllib)  
  
  
  
  
\# In\[ \]: