基于豆瓣电影数据进行RDD编程分析（含豆瓣电影top250爬虫）

爬虫代码：豆瓣_电影.txt

爬取数据：

开始：

• 导入库，配置java环境，初始化数据：

import os
from pyspark import SparkConf,SparkContext
import pandas as pd
import random
import matplotlib.pyplot as plt

#先配置java环境
os.environ["JAVA_HOME"]="/opt/Cluster/jdk1.8.0_162"#这里需要换成自己jdk的路径
os.environ["PYSPARK_PYTHON"]='/usr/bin/python3.6'

#初始化SparkConf时，或者提交Spark任务时，都会有master参数需要设置
#Local模式就是运行在一台计算机上的模式，通常就是用于在本机上练手和测试
conf = SparkConf().setMaster("local").setAppName("MovieDouban")
sc=SparkContext(conf=conf)
print(sc.version)#测试是否连接成功

#spark部分
filepath= r'file:///home/SparkProgram/QZ/data.csv' #将爬取的数据传到虚拟机上自己选择的地址
rdd = sc.textFile(filepath)
# rdd.foreach(print)

rdd1 = rdd.map(lambda x:x.split(","))#单纯分
# rdd1.foreach(print)

• 统计时间列出来，（k,v）,然后进行统计各年份数量统，绘制柱状图年份与数量V1.png

#统计时间列出来，（k,v）,然后进行统计各年份数量
rdd_year = rdd.map(lambda x:(x.split(",")[3],1)).reduceByKey(lambda a,b:a+b).filter(lambda x:'Jason Flemyng'not in x[0])
# rdd_year.foreach(print)

# #对时间进行绘制柱状图
# x_data=[]#横坐标数据
# y_data=[]#竖坐标数据
# for ii in range(len(rdd_year.collect())):
#     x_data.append(rdd_year.collect()[ii][0])
#     y_data.append(rdd_year.collect()[ii][1])
# # print(x_data)#查看是否单列出来
# # print(y_data)
# plt.bar(x=x_data,height=y_data,color='steelblue',alpha=0.8)#作柱状图
# plt.xticks(rotation='vertical')#因为横坐标文字重复，将其设置为竖排
# plt.xlabel('Year')
# plt.ylabel('Data')
# plt.title('Year_Data')
# # plt.savefig('年份与数量V1.png',dpi=600)
# plt.show()

• 找出出现最频繁的电影类型，并以年份和数量为横轴纵轴的折线图，保存为年份与数量V2.png

#先统计哪个类型最频繁
rdd_type = rdd.map(lambda x: x.split(",")[5]).flatMap(lambda x:x.split(' ')).map(lambda x:(x,1)).reduceByKey(lambda a,b:a+b).sortBy(lambda x:x[1],False)
# rdd_type.foreach(print)

#已知爱情最多,对年份进行统计
rdd_love = rdd.map(lambda x:(x.split(",")[3],x.split(',')[5])).filter(lambda x:'爱情'in x[1]).map(lambda x:(x[0],1))
# rdd_love.foreach(print)

# #对年份进行绘制折线图
# x_data=[]#横坐标数据
# y_data=[]#竖坐标数据
# for ii in range(len(rdd_love.collect())):
#     x_data.append(rdd_love.collect()[ii][0])
#     y_data.append(rdd_love.collect()[ii][1])
# # print(x_data)#查看是否单列出来
# # print(y_data)
# plt.plot(x_data,y_data,color='steelblue',linewidth=4)#作折线图
# plt.xticks(rotation='vertical')#因为横坐标文字重复，将其设置为竖排
# plt.xlabel('Year')
# plt.ylabel('Data')
# plt.title('Lovetype_Year_Data')
# plt.savefig('年份与数量V2.png',dpi=600)
# plt.show()

• 观察美国与我国电影的发展史，已年份和数量为横纵轴，绘制两国柱状图，保存为年份与数量V3.png

rdd_country_china = rdd.map(lambda x:(x.split(",")[3],x.split(',')[4])).filter(lambda x:'中国' in x[1]).map(lambda x:(x[0],1)).reduceByKey(lambda a,b:a+b).sortByKey(True)
rdd_country_USA = rdd.map(lambda x:(x.split(",")[3],x.split(',')[4])).filter(lambda x:'美国' in x[1]).map(lambda x:(x[0],1)).reduceByKey(lambda a,b:a+b).sortByKey(True)
rdd_usa=rdd_country_USA.map(lambda x:x[0]) #看美国有哪些年份，因为中国的年份要少于美国，制图时得对应
# rdd_country_china.foreach(print)
# rdd_country_USA.foreach(print)
# rdd_usa.foreach(print)

# # #对美国、中国合并柱状图
# #中国
# x_data=[]#横坐标数据
# y_data=[]#竖坐标数据
# for ii in range(len(rdd_country_china.collect())):
#     if rdd_country_china.collect()[ii][0] in rdd_usa.collect():
#         x_data.append(rdd_country_china.collect()[ii][0])
#         y_data.append(rdd_country_china.collect()[ii][1])
#     else :
#         x_data.append(rdd_usa.collect()[ii])
#         y_data.append(0)
# print(len(x_data),len(y_data))
# print('x',x_data)
# print('y',y_data)
#
# #美国
# y_data1=[]#竖坐标数据
# for jj in range(len(rdd_country_china.collect())):
#     y_data1.append(rdd_country_USA.collect()[jj][1])
# print(y_data1)
# x=list(range(len(x_data)))
# width = 0.4
# plt.xticks(rotation=270)
# plt.bar(x,y_data,width=width,label='China',tick_label=x_data,fc='y')#作柱状图
# for i in range(len(x)):
#     x[i]=x[i]+width
# print(len(y_data1),len(y_data))
# plt.bar(x,y_data1,width=width,fc='r')#作柱状图
# # plt.xticks(rotation='vertical')#因为横坐标文字重复，将其设置为竖排
# plt.xlabel('Year')
# plt.ylabel('Data')
# plt.title('China_USA_Year_Data')
# plt.savefig('年份与数量V3.png',dpi=600)
# plt.legend()
# plt.show()

• 找出当前评价人数最多的电影，并以电影类型为指标绘制饼状图，保存为电影类型.png

rdd_comment= rdd.map(lambda x:(x.split(',')[7],x.split(',')[5])).filter(lambda x:'98956' not in x[0]).sortByKey(False).first()
# rdd_comment.foreach(print)
# print(rdd_comment[1])#剧情、爱情
rdd_love1 = rdd.map(lambda x:(x.split(",")[3],x.split(',')[5])).filter(lambda x:'爱情'in x[1]).map(lambda x:(x[0],1)).count()
rdd_pass = rdd.map(lambda x:(x.split(",")[3],x.split(',')[5])).filter(lambda x:'剧情'in x[1]).map(lambda x:(x[0],1)).count()
# rdd_pass.foreach(print)
# print(rdd_love1)
# print(rdd_pass)

# #饼状图
# year_x=['剧情','爱情']#横坐标数据
# year_y=[rdd_pass,rdd_love1]#竖坐标数据
# num=rdd_pass+rdd_love1
# print(num)
# pa=rdd_pass/num*100
# lo=rdd_love1/num*100
# size=[pa,lo]
# colors = ['red','blue']
# print(len(year_x),len(year_y),len(size))
# plt.pie(size,labels=year_x,shadow=True,colors=colors)
# plt.axis('equal')
# plt.savefig('电影类型.png',dpi=600)
# plt.show()

• 找出动画片、科幻片、剧情片和喜剧片下最有价值的导演和演员，保存为T1.txt

#导演
rdd_katong= rdd.map(lambda x:(x.split(',')[1],x.split(',')[5])).filter(lambda x:'动画'in x[1]).map(lambda x:(x[0],1)).reduceByKey(lambda a,b:a+b).sortBy(lambda x:x[1],False).first()
rdd_kehuan= rdd.map(lambda x:(x.split(',')[1],x.split(',')[5])).filter(lambda x:'科幻'in x[1]).map(lambda x:(x[0],1)).reduceByKey(lambda a,b:a+b).sortBy(lambda x:x[1],False).first()
rdd_juqing= rdd.map(lambda x:(x.split(',')[1],x.split(',')[5])).filter(lambda x:'剧情'in x[1]).map(lambda x:(x[0],1)).reduceByKey(lambda a,b:a+b).sortBy(lambda x:x[1],False).first()
rdd_xiju= rdd.map(lambda x:(x.split(',')[1],x.split(',')[5])).filter(lambda x:'喜剧'in x[1]).map(lambda x:(x[0],1)).reduceByKey(lambda a,b:a+b).sortBy(lambda x:x[1],False).first()
# rdd_katong.foreach(print)
daoyan=[rdd_katong[0],rdd_kehuan[0],rdd_juqing[0],rdd_xiju[0]]
# print(daoyan)
#演员
rdd_katong1= rdd.map(lambda x:(x.split(',')[2],x.split(',')[5])).filter(lambda x:'动画'in x[1]).map(lambda x:(x[0],1)).reduceByKey(lambda a,b:a+b).sortBy(lambda x:x[1],False).first()
rdd_kehuan1= rdd.map(lambda x:(x.split(',')[2],x.split(',')[5])).filter(lambda x:'科幻'in x[1]).map(lambda x:(x[0],1)).reduceByKey(lambda a,b:a+b).sortBy(lambda x:x[1],False).first()
rdd_juqing1= rdd.map(lambda x:(x.split(',')[2],x.split(',')[5])).filter(lambda x:'剧情'in x[1]).map(lambda x:(x[0],1)).reduceByKey(lambda a,b:a+b).sortBy(lambda x:x[1],False).first()
rdd_xiju1= rdd.map(lambda x:(x.split(',')[2],x.split(',')[5])).filter(lambda x:'喜剧'in x[1]).map(lambda x:(x[0],1)).reduceByKey(lambda a,b:a+b).sortBy(lambda x:x[1],False).first()
yanyuan=[rdd_katong1[0],rdd_kehuan1[0],rdd_juqing1[0],rdd_xiju1[0]]
# print(yanyuan)
sum = yanyuan+daoyan
# print(sum)

# #存txt文件
# f=open("T1.txt","w")
# f.write(str(sum))
# f.close()

• 找出评分与评价人数最多的那部电影,并输出它的评价，保存为T2.txt

rdd_score= rdd.map(lambda x:(x.split(',')[8],x.split(',')[7],x.split(',')[6])).filter(lambda x:'98956' not in x[1]).sortBy(lambda x:(x[1],x[2]),False).first()
# rdd_score.foreach(print)
# print(rdd_score[0])
w=open("T2.txt","w")
w.write(str(rdd_score[0]))
w.close()

代码中可能不全面，还请多包涵