目 录
摘 要 I
Abstract II
1 绪论 1
1.1 选题背景及意义 1
1.2 研究现状 2
1.3 研究内容及论文组织结构 2
2 关键技术和工具环境 4
2.1 IDEA简介 4
2.2 HTML/CSS简介 4
2.3 Spark简介 5
2.4 Spark SQL简介 5
2.5 Hadoop简介 6
2.6 ECharts简介 6
2.7 MySQL简介 6
3 系统分析 8
3.1 功能需求分析 8
3.2 业务流程分析 8
3.3 数据流图 12
3.4 数据库概念模型设计 15
4 系统设计 16
4.1 系统网络架构设计 16
4.2 系统总体设计 16
4.3 系统功能模块设计 17
4.4 数据库逻辑结构设计 19
5 详细设计 20
5.1 程序系统的结构 20
5.2 大数据集群框架模块设计说明 20
5.2.1 程序描述 20
5.2.2 功能 20
5.2.3 算法 21
5.3 大数据处理模块设计说明 21
5.3.1 程序描述 21
5.3.2 功能 22
5.3.3 算法 22
5.4 数据可视化模块设计 22
5.4.1 程序描述 22
5.4.2 功能 23
5.4.3 算法 23
5.4 数据库详细设计 23
5.4.1 数据库表设计 23
5.4.2 数据库连接设计 24
6 系统实现 27
6.1 数据清洗的实现 27
6.2 数据库工具类编写实现 29
6.3导入IPUtils工具类对IP进行解析 30
6.4编写Dao层将数据解析并存储到数据库中 31
6.5 对各维度数据的统计并调用Dao入库 35
6.6 构建数据可视化项目 39
6.7 使用echarts进行数据可视化 42
7 系统测试 45
7.1 运行环境说明 45
7.2 服务器集群测试 45
7.3 单元测试 47
7.4 测试结果 49
总 结 50
参考文献 51
致 谢 52
1.3 研究内容及论文组织结构
(1)内容概述
在对大数据的深入研究后，根据其前景，包括数据方面的发展与价值，本套系统从用户痛点需求进行分析入手，对系统架构进行了设计，随后完成了系统方面的具体设计，最后为数据入库对DB进行配置和设计，最后到系统搭建和编码实现，分别为后台数据处理，在数据转换方面包括数据的clean、临时存储落地，数据经过完全处理后入库，和前台的ECharts可视化系统，对处理后落地的数据使用饼图进行可视化展现。对系统进行功能叙述、进行详细的系统分析、进行整体的结构性框架设计和对系统详细的设计、最终完成系统的搭建部分和对系统进行的单元测试这几个方面描述了整个系统的流程。本系统从当前的技术成熟度和需求出发，经过反复的学习和查阅资料，最后选择用Spark SQL这个开源项目，使用scala这种原生的spark框架编写语言来书写，对数据使用MySQL进行数据落地，最后使用echarts进行可视化处理，综合完成数据的处理工作。最后进行系统的测试工作，来检测是否能够进行数据的正常处理并可视化，以便达到并符合标准。结合大数据的发展历史和技术手段，通过了解企业的数据开发和处理流程，本系统采用了大多数企业的数据采集和处理技术。通过对豆瓣电影分析功能的完成，最后得到电影详情的排名，各部省份的访问量排名，流量排名，最终为了优化系统的目的。快速完成豆瓣电影的离线批处理，在豆瓣电影数据中提取了有效的潜在价值，并为企业和组织减轻了处理时服务器的压力，和帮助更好的决策。
(2)系统开发环境介绍
IDEA简介、HTML/CSS简介、Spark简介、Spark SQL简介、hadoop简介、Echarts简介、MySQL数据库概念和特点、本章小结。
(3)系统分析
概述、用户需求研究、系统业务流程分析、系统数据组成、本章总结。
(4)系统设计
大数据系统的结构设计、系统总体设计、系统功能模块划分、系统功能模块需求、数据库设计概述、数据库概览结构设计、数据逻辑结构设计。
(5)系统的实现
数据采集模块的实现、豆瓣电影收集模块设计、豆瓣电影主控模块的实现、消息队列模块的实现、大数据集群的实现、业务逻辑的实现、数据可视化的实现。
(6)系统测试
系统编码实现后，需要进行测试，首先有运行环境的测试，包括相应的硬件要求及软件要求，然后是功能测试，本文转载自http://www.biyezuopin.vip/onews.asp?id=14108其中的测试包括黑盒测试和白盒测试，本系统采用黑盒测试，通过输入不同组的测试数据进行测试的功能模块测试，从而使本系统达到一种安全使用的状态。

# -*-coding:utf-8-*-

from urllib import request
from download import down_html
from parase import parase_html
from output import output_all
from url_manager import manage_url
root_url = "https://movie.douban.com/tag/?view=cloud"

class DouBan_Spider(object):
    def __init__(self):
        self.category_dic = {} #用来存储所有热门分类的名字和对应的电影数目,catename和catename_num两个属性
        self.down_class = down_html.DownHtml()  #下载网页
        self.parase_class = parase_html.ParaseHtml() #解析网页
        self.output_class = output_all.OutPut() #存储信息
        self.manage_class = manage_url.UrlManager() #链接管理
        self.tag_right = 1
        self.tag_error = 0

    #获取分类下所有热门分类
    def get_hotcategory(self,url):
        print("get all category!")
        page_content = self.down_class.download(url)
        self.category_dic = self.parase_class.parase_category(page_content)
        self.output_class.output_category(self.category_dic) #将类别信息写入本地文件

    #得到某个类别下所有电影的链接
    def get_one_cate_all_movie_href(self,tag_url):
        page_content = self.down_class.download(tag_url)
        page_num = self.parase_class.parase_pagenum(page_content) #得到该分类总共多少页
        movies_href = []
        try:
            for page in range(int(page_num)):
                page_url = "https://movie.douban.com/tag/%E8%8B%B1%E5%9B%BD?start="+str(page*20)+"&type=T"
                tag_page_content = self.down_class.download(page_url)
                movies_href = self.parase_class.parase_page_all_movies(tag_page_content,movies_href)
                print("all:",page_num,"  right:",self.tag_right,"  error：",self.tag_error,"  page:",page+1,"  URL 获取完毕")
                self.tag_right +=1
        except Exception as e:
            print(e)
            self.tag_error+=1
            pass
        print("该类别下对应的电影数目为：\t",len(movies_href))
        self.output_class.output_all_movies_href(movies_href)
        print("开始获取该类别下对应的电影信息：\n ")
        self.get_one_movie_message(movies_href)      #该类别对应的链接抓取完毕，进行这些链接对应电影信息的抓取

    #下载每个电影的详细信息
    def get_one_movie_message(self,movie_link):
        all_count = 1
        error = 0
        self.manage_class.add_new_urls(movie_link) #将一个类别对应的链接全部加载到manage_url管理的新的链接中
        while(self.manage_class.has_new_url()):
            try:
                one_url = self.manage_class.get_new_url()  #获取一个url
                # one_url=" https://movie.douban.com/subject/1297970/"
                print("Right：",all_count,"  URL:",one_url.strip(),"  ","Error:",error)
                page_content_one = self.down_class.download(one_url)  #下载该网页对应的源代码
                one_movic_dic = self.parase_class.parase_one_movie_message(page_content_one) #解析得到一部电影的具体数据
                id = one_url.split("/")[-2]
                self.output_class.output_one_movie_message(one_movic_dic,id) #将该部电影的数据输出
                all_count+=1
            except Exception as e:
                error +=1
                print(e)
                pass

    #获取每部电影的短评
    def get_one_movie_short_dis(self,movie_url):
        error = 0
        try:
            m_id = movie_url.split("/")[-2]
            movie_url = "https://movie.douban.com/subject/"+str(m_id)+"/comments?sort=new_score"
            short_dis_num = self.parase_class.parase_dis_num(self.down_class.download(movie_url))
            for i in range(int(short_dis_num/20)):  #每页20条数据
                url = "https://movie.douban.com/subject/"+str(m_id)+"/comments?start="+str(i *20)+"&limit=20&sort=new_score"
                duanping_list = self.parase_class.parase_one_movie_duanping(self.down_class.download(url))
                self.output_class.output_duanping(m_id,duanping_list)
                print("ID:",m_id,"Page:",(i+1),"短评写入OK")
        except Exception as e:
            error += 1
            print("电影短评异常:",e)
            pass
        print("ID:", m_id,"短评全部写入OK","异常次数:",error)

    #获取每部电影的影评
    def get_one_movie_long_dis(self,movie_url):
        error = 0
        m_id = movie_url.split("/")[-2]
        movie_url = "https://movie.douban.com/subject/" + str(m_id) + "/reviews"
        long_dis_num = self.parase_class.parase_dis_num(self.down_class.download(movie_url))
        print(long_dis_num)
        for i in range(int(long_dis_num / 20)):  # 每页20条数据
            try:
                url = "https://movie.douban.com/subject/" + str(m_id) + "/reviews?start=" + str(i * 20) + "&filter=&limit=20"
                yingping_list = self.parase_class.parase_one_movie_yingping(self.down_class.download(url))
                self.output_class.output_yingping(m_id, yingping_list)
                print("ID:", m_id, "Page:", (i + 1), "影评写入OK")
            except Exception as e:
                error += 1
                print("电影影评异常:", e)
                pass
        print("ID:", m_id, "影评全部写入OK", "异常次数:", error)

if __name__=="__main__":
    spider = DouBan_Spider()
    '''
    spider.get_hotcategory(root_url)    #获取所有热门分类，，并存储在self.category_dic中
    count = 1
    for cate_name,cate_num in spider.category_dic.items():    #输出所有标签和对应的数目
        tag_url = "https://movie.douban.com/tag/" + request.quote(cate_name)       #将链接中的中文进行转化形成二级tag url
        print(count,"\t标签：",cate_name,"\t数目：",cate_num)
        count += 1  #计数
        spider.get_one_cate_all_movie_href(tag_url)  #得到某个类别下所有电影的链接
    '''
    # while(spider.manage_class.has_new_url()):
    #     spider.get_one_movie_short_dis(spider.manage_class.get_new_url())
    #     spider.get_one_movie_long_dis(spider.manage_class.get_new_url())
    #spider.get_one_movie_short_dis("https://movie.douban.com/subject/5045678/")
    spider.get_one_movie_long_dis("https://movie.douban.com/subject/5045678/")