博主介绍：✌全网粉丝10W+,前互联网大厂软件研发、集结硕博英豪成立工作室。专注于计算机相关专业项目实战6年之久，选择我们就是选择放心、选择安心毕业✌
> 🍅想要获取完整文章或者源码，或者代做，拉到文章底部即可与我联系了。🍅

点击查看作者主页，了解更多项目！

🍅感兴趣的可以先收藏起来，点赞、关注不迷路，大家在毕设选题，项目以及论文编写等相关问题都可以给我留言咨询，希望帮助同学们顺利毕业 。🍅

1、毕业设计：2026年计算机专业毕业设计选题汇总（建议收藏）✅

2、大数据毕业设计：2026年选题大全 深度学习 python语言 JAVA语言 hadoop和spark（建议收藏）✅

1、项目介绍

技术栈
融合Hadoop、Spark、Hive等大数据技术与虚拟机环境，以Python为核心开发语言，基于Django框架搭建系统架构，采用MySQL数据库存储数据，运用协同过滤推荐算法、随机森林回归模型等机器学习技术，结合猫眼电影数据实现票房预测与电影推荐。

功能模块

• 电影数据大屏
• 电影类型分析
• 电影评分分析
• 电影时间分析
• 票房分析
• 数据中心
• 电影推荐
• 词云图分析
• 票房预测
• 我的收藏
• 后台管理
• spark数据分析
• 数据采集

项目介绍
本项目聚焦电影数据分析系统的开发与落地，依托Spark等大数据技术处理海量猫眼电影数据，基于Python、Django搭建Web系统架构，利用MySQL存储结构化数据。系统整合数据采集、多维度可视化分析、基于随机森林回归模型的票房预测、协同过滤算法驱动的电影推荐等核心功能，通过Echarts将分析结果以可视化形式呈现。该系统能够高效处理大规模电影数据，输出精准的票房预测结果与个性化电影推荐，为电影行业从业者提供数据支撑与决策参考，具备功能完善、运行稳定的优势。

2、项目界面

电影数据大屏
集成了多类数据展示功能：通过饼图呈现类型占比，列表展示影片信息，柱状图统计票房评分，折线图分析类型票房趋势，还有国家发行、电影票房占比等图表，可直观查看电影多维度数据的统计、分布与趋势，实现电影数据的综合可视化分析。

电影类型分析
左侧设含多种分析功能的导航栏，页面主体通过折线图展示指定类型电影的票房 TOP10 数据，搭配柱状图呈现该类型评分分布，同时以饼图展示各类型电影平均时长，支持查看特定类型电影的多维度数据统计与分布情况，实现分类型电影数据的可视化分析。

电影评分分析
左侧设含多种数据功能的导航栏，页面主体通过柱状图展示年度评分值分布，以折线图呈现地区评分值变化，还包含类型评分值分析板块，可查看电影在年度、地区、类型维度的评分数据情况，实现多维度电影评分的可视化统计与分析。

电影时间分析
左侧设含各类数据功能的导航栏，页面主体通过柱状图展示电影年度、月度的产量分布，搭配饼图呈现电影时长的分类占比，可查看电影在产量时间维度及时长分类维度的数据情况，实现电影产量与时长的多维度可视化统计与分析。

票房分析
左侧设含各类数据功能的导航栏，页面主体通过折线图展示各类型电影票房走势，用柱状图呈现年度票房值分布，还包含各地区票房分析板块，可查看电影在类型、年度、地区维度的票房数据情况，实现多维度电影票房的可视化统计与分析。

数据中心
左侧设含各类数据功能的导航栏，页面主体以列表形式展示电影的名称、类型、时长、评分等详细信息，配备搜索框可检索内容，同时支持勾选操作，实现电影数据的集中展示、查阅与检索管理。

电影推荐
左侧设含各类数据功能的导航栏，页面主体以卡片形式展示推荐电影的海报、名称、简介、时长等信息，配备 “收藏” 按钮支持用户收藏影片，实现电影推荐内容的展示与收藏操作管理。

词云图分析
左侧设含各类数据功能的导航栏，页面主体展示电影相关关键词组成的词云图，通过文字大小差异体现关键词的热度分布，实现电影相关元素的可视化统计与直观呈现，帮助用户快速了解电影领域的高频内容。

票房预测
左侧设含各类数据功能的导航栏，页面主体提供类型、地区、首周票房的条件选择栏与查询按钮，提交后展示对应条件下的票房预测结果，实现根据指定条件进行电影票房的预测计算与结果展示。

我的收藏
左侧设含各类数据功能的导航栏，页面主体以卡片形式展示电影海报、名称、类型、地区、时长等信息，配备 “获取数据” 按钮，实现电影相关数据的集中展示与数据获取操作，便于后续分析或预测使用。

后台管理
左侧设包含历史表、用户表等功能的导航栏，页面主体展示快速操作区域（含登录、用户表等功能入口）与最近动作板块，实现后台管理功能的快速访问与操作记录查看，辅助管理员高效管理系统数据与权限。

spark数据分析
左侧展示项目文件结构，中间编辑区包含 Spark 会话初始化、数据读取、字段拆分与分组统计的代码，底部显示数据结果，实现通过 Spark 完成数据读取、处理及统计分析的功能，支持代码编写与数据分析结果查看。

数据采集
左侧展示项目文件结构，中间编辑区是数据读取、字段处理的代码，右侧显示电影详情页面，底部呈现采集的电影数据信息，实现从网页采集电影数据、通过代码处理数据的功能，支持数据采集、代码编写与数据信息查看。

3、项目说明

一、技术栈
本项目融合Hadoop、Spark、Hive等大数据技术与虚拟机环境，以Python为核心开发语言，基于Django框架搭建系统架构，采用MySQL数据库存储数据，运用协同过滤推荐算法、随机森林回归模型等机器学习技术，结合猫眼电影数据实现票房预测与电影推荐。

二、功能模块详细介绍

• 电影数据大屏：集成多类数据展示功能，通过饼图、列表、柱状图、折线图等呈现电影类型占比、影片信息、票房评分、类型票房趋势等数据，实现电影数据综合可视化分析。
• 电影类型分析：左侧设多功能导航栏，主体通过折线图、柱状图、饼图展示指定类型电影票房TOP10、评分分布、各类型平均时长，实现分类型电影数据可视化分析。
• 电影评分分析：左侧设数据功能导航栏，主体以柱状图、折线图展示年度评分分布、地区评分变化及类型评分分析，实现多维度评分可视化统计分析。
• 电影时间分析：左侧设数据功能导航栏，主体通过柱状图、饼图展示电影年度/月度产量分布、时长分类占比，实现产量与时长多维度可视化统计分析。
• 票房分析：左侧设数据功能导航栏，主体以折线图、柱状图展示各类型票房走势、年度票房分布及地区票房分析，实现多维度票房可视化统计分析。
• 数据中心：左侧设数据功能导航栏，主体以列表展示电影详细信息，配备搜索框与勾选功能，实现电影数据集中展示、查阅与检索管理。
• 电影推荐：左侧设数据功能导航栏，主体以卡片展示推荐电影信息，配备收藏按钮，实现推荐内容展示与收藏操作管理。
• 词云图分析：左侧设数据功能导航栏，主体展示电影相关关键词词云图，通过文字大小体现热度分布，实现电影元素可视化统计。
• 票房预测：左侧设数据功能导航栏，主体提供条件选择栏与查询按钮，提交后展示票房预测结果，实现指定条件下的票房预测。
• 我的收藏：左侧设数据功能导航栏，主体以卡片展示电影信息，配备获取数据按钮，实现收藏数据集中展示与获取操作。
• 后台管理：左侧设功能导航栏，主体展示快速操作区与最近动作板块，实现后台功能快速访问与操作记录查看。
• spark数据分析：左侧展示文件结构，中间编辑区编写Spark数据分析代码，底部显示结果，实现数据读取、处理与统计分析。
• 数据采集：左侧展示文件结构，中间编辑区编写数据采集代码，右侧显示数据源页面，底部呈现采集数据，实现电影数据采集与处理。

三、项目总结
本电影数据分析系统依托Spark等大数据技术处理海量猫眼电影数据，基于Python+Django搭建Web架构，MySQL存储结构化数据。系统整合数据采集、多维度可视化分析、随机森林模型票房预测、协同过滤算法电影推荐等功能，通过Echarts可视化呈现分析结果。该系统可高效处理大规模电影数据，输出精准的票房预测与个性化推荐结果，为电影行业提供全面的数据支撑与决策依据，具备功能完整、运行稳定的显著优势。

4、核心代码

#导包
from pyspark.sql import SparkSession
from pyspark.sql.functions import monotonically_increasing_id
from pyspark.sql.types import StructType,StructField,IntegerType,StringType,FloatType
from pyspark.sql.functions import count,mean,col,sum,when,max,min,avg,explode,split,row_number,year,month
from pyspark.sql.window import Window


if __name__ == '__main__':
    #构建
    spark = SparkSession.builder.appName("sparkSQL").master("local[*]"). \
        config("spark.sql.shuffle.partitions", 2). \
        config("spark.sql.warehouse.dir", "hdfs://node1:8020/user/hive/warehouse"). \
        config("hive.metastore.uris", "thrift://node1:9083"). \
        enableHiveSupport(). \
        getOrCreate()


    #读取
    catMovieData = spark.read.table("catMovieData")

    #需求一 类型分析 统计
    explode_df = catMovieData.withColumn("type",explode(split(col("type"),"-")))
    result1 = explode_df.groupBy("type").agg(count("type").alias("type_count"))

    # sql
    result1.write.mode("overwrite"). \
        format("jdbc"). \
        option("url", "jdbc:mysql://node1:3306/bigdata?useSSL=false&useUnicode=true&charset=utf8"). \
        option("dbtable", "movieTypeCount"). \
        option("user", "root"). \
        option("password", "root"). \
        option("encoding", "utf-8"). \
        save()

    result1.write.mode("overwrite").saveAsTable("movieTypeCount", "parquet")
    spark.sql("select * from movieTypeCount").show()

    #需求二 票房TOP10
    sorted_df =catMovieData.orderBy(col("allBoxOffice").desc())
    result2 = sorted_df.limit(10)

    # sql
    result2.write.mode("overwrite"). \
        format("jdbc"). \
        option("url", "jdbc:mysql://node1:3306/bigdata?useSSL=false&useUnicode=true&charset=utf8"). \
        option("dbtable", "boxTopMovie"). \
        option("user", "root"). \
        option("password", "root"). \
        option("encoding", "utf-8"). \
        save()

    result2.write.mode("overwrite").saveAsTable("boxTopMovie", "parquet")
    spark.sql("select * from boxTopMovie").show()

    #需求三 类型最大票房
    result3 = explode_df.groupBy("type").agg(
        max("firstBoxOffice").alias("max_firstBoxOffice"),
        max("allBoxOffice").alias("max_allBoxOffice")
    )

    # sql
    result3.write.mode("overwrite"). \
        format("jdbc"). \
        option("url", "jdbc:mysql://node1:3306/bigdata?useSSL=false&useUnicode=true&charset=utf8"). \
        option("dbtable", "typeMaxBox"). \
        option("user", "root"). \
        option("password", "root"). \
        option("encoding", "utf-8"). \
        save()

    result3.write.mode("overwrite").saveAsTable("typeMaxBox", "parquet")
    spark.sql("select * from typeMaxBox").show()

    #需求四 国家统计
    explode_df2 = catMovieData.withColumn("country",explode(split(col("country"),",")))

    result4 = explode_df2.groupBy("country").agg(count("country").alias("country_count"))

    # sql
    result4.write.mode("overwrite"). \
        format("jdbc"). \
        option("url", "jdbc:mysql://node1:3306/bigdata?useSSL=false&useUnicode=true&charset=utf8"). \
        option("dbtable", "mcountryCount"). \
        option("user", "root"). \
        option("password", "root"). \
        option("encoding", "utf-8"). \
        save()

    result4.write.mode("overwrite").saveAsTable("mcountryCount", "parquet")
    spark.sql("select * from mcountryCount").show()

    #需求五 评分分类

    catMovieData = catMovieData.withColumn(
        "rateCategory",
        when(col("rate") == 0, None).otherwise(
            when((col("rate") >= 10) & (col("rate") < 20), "半星").
            when((col("rate") >= 20) & (col("rate") < 30), "1星").
            when((col("rate") >= 30) & (col("rate") < 40), "1.5星").
            when((col("rate") >= 40) & (col("rate") < 50), "2星").
            when((col("rate") >= 50) & (col("rate") < 60), "2.5星").
            when((col("rate") >= 60) & (col("rate") < 70), "3星").
            when((col("rate") >= 70) & (col("rate") < 80), "3.5星").
            when((col("rate") >= 80) & (col("rate") < 90), "4星").
            when((col("rate") >= 90) & (col("rate") < 100), "4.5星")
        )
    )

    filter_df = catMovieData.filter(col("rate") != 0)
    result5 = filter_df.groupBy("rateCategory").count()

    # sql
    result5.write.mode("overwrite"). \
        format("jdbc"). \
        option("url", "jdbc:mysql://node1:3306/bigdata?useSSL=false&useUnicode=true&charset=utf8"). \
        option("dbtable", "starCategory"). \
        option("user", "root"). \
        option("password", "root"). \
        option("encoding", "utf-8"). \
        save()

    result5.write.mode("overwrite").saveAsTable("starCategory", "parquet")
    spark.sql("select * from starCategory").show()

    #需求6 按类型票房分析
    grouped_df = explode_df.groupBy("type","title").agg({"allBoxOffice":"sum"})
    grouped_df = grouped_df.withColumnRenamed("sum(allBoxOffice)","allBoxOffice")

    window = Window.partitionBy("type").orderBy(col("allBoxOffice").desc())
    ranked_df = grouped_df.withColumn("row_num",row_number().over(window))
    #过滤出每组的前十
    result6 = ranked_df.filter(col("row_num") <= 10).drop("row_num")

    # sql
    result6.write.mode("overwrite"). \
        format("jdbc"). \
        option("url", "jdbc:mysql://node1:3306/bigdata?useSSL=false&useUnicode=true&charset=utf8"). \
        option("dbtable", "perTypeBox"). \
        option("user", "root"). \
        option("password", "root"). \
        option("encoding", "utf-8"). \
        save()

    result6.write.mode("overwrite").saveAsTable("perTypeBox", "parquet")
    spark.sql("select * from perTypeBox").show()


    #需求7 按类型评分
    filter_df2 = explode_df.filter(col("rate") != 0)
    result7 = filter_df2.groupBy("type","rate").agg(count("rate").alias("rate_count"))

    # sql
    result7.write.mode("overwrite"). \
        format("jdbc"). \
        option("url", "jdbc:mysql://node1:3306/bigdata?useSSL=false&useUnicode=true&charset=utf8"). \
        option("dbtable", "perTypeRate"). \
        option("user", "root"). \
        option("password", "root"). \
        option("encoding", "utf-8"). \
        save()

    result7.write.mode("overwrite").saveAsTable("perTypeRate", "parquet")
    spark.sql("select * from perTypeRate").show()

    #需求8类型平均时长
    result8 = explode_df.groupBy("type").agg(avg("duration").alias("avg_duration"))

    # sql
    result8.write.mode("overwrite"). \
        format("jdbc"). \
        option("url", "jdbc:mysql://node1:3306/bigdata?useSSL=false&useUnicode=true&charset=utf8"). \
        option("dbtable", "typeAvgTime"). \
        option("user", "root"). \
        option("password", "root"). \
        option("encoding", "utf-8"). \
        save()

    result8.write.mode("overwrite").saveAsTable("typeAvgTime", "parquet")
    spark.sql("select * from typeAvgTime").show()

    #需求9 年度平均评分分析
    time_df1 = catMovieData.withColumn("releaseTime",col("releaseTime").cast("date"))

    time_df2 = catMovieData.withColumn("year",year(col("releaseTime")))

    result9 = time_df2.groupBy("year").agg(avg("rate").alias("year_rate"))

    # sql
    result9.write.mode("overwrite"). \
        format("jdbc"). \
        option("url", "jdbc:mysql://node1:3306/bigdata?useSSL=false&useUnicode=true&charset=utf8"). \
        option("dbtable", "mYearRate"). \
        option("user", "root"). \
        option("password", "root"). \
        option("encoding", "utf-8"). \
        save()

    result9.write.mode("overwrite").saveAsTable("mYearRate", "parquet")
    spark.sql("select * from mYearRate").show()

    #需求十 国家平均
    result10 = explode_df2.groupBy("country").agg(avg("rate").alias("avg_rate"))

    # sql
    result10.write.mode("overwrite"). \
        format("jdbc"). \
        option("url", "jdbc:mysql://node1:3306/bigdata?useSSL=false&useUnicode=true&charset=utf8"). \
        option("dbtable", "mCountryRate"). \
        option("user", "root"). \
        option("password", "root"). \
        option("encoding", "utf-8"). \
        save()

    result10.write.mode("overwrite").saveAsTable("mCountryRate", "parquet")
    spark.sql("select * from mCountryRate").show()

    #需求十一 类型
    result11 = explode_df.groupBy("type").agg(avg("rate").alias("avg_rate"))

    # sql
    result11.write.mode("overwrite"). \
        format("jdbc"). \
        option("url", "jdbc:mysql://node1:3306/bigdata?useSSL=false&useUnicode=true&charset=utf8"). \
        option("dbtable", "mTypeRate"). \
        option("user", "root"). \
        option("password", "root"). \
        option("encoding", "utf-8"). \
        save()

    result11.write.mode("overwrite").saveAsTable("mTypeRate", "parquet")
    spark.sql("select * from mTypeRate").show()


    #需求十二 时间分析
    result12 = time_df2.groupBy("year").agg(count("year").alias("year_count"))

    # sql
    result12.write.mode("overwrite"). \
        format("jdbc"). \
        option("url", "jdbc:mysql://node1:3306/bigdata?useSSL=false&useUnicode=true&charset=utf8"). \
        option("dbtable", "mYearCount"). \
        option("user", "root"). \
        option("password", "root"). \
        option("encoding", "utf-8"). \
        save()

    result12.write.mode("overwrite").saveAsTable("mYearCount", "parquet")
    spark.sql("select * from mYearCount").show()

    #月度
    time_df3 = time_df1.withColumn("month",month(col("releaseTime")))
    result13 = time_df3.groupBy("month").agg(count("month").alias("year_count"))

    # sql
    result13.write.mode("overwrite"). \
        format("jdbc"). \
        option("url", "jdbc:mysql://node1:3306/bigdata?useSSL=false&useUnicode=true&charset=utf8"). \
        option("dbtable", "mMonthCount"). \
        option("user", "root"). \
        option("password", "root"). \
        option("encoding", "utf-8"). \
        save()

    result13.write.mode("overwrite").saveAsTable("mMonthCount", "parquet")
    spark.sql("select * from mMonthCount").show()

    #时长分析
    catMovieData = catMovieData.withColumn(
        "durationCategory",
        when(col("duration") == 0, None).otherwise(
            when((col("duration") >= 0) & (col("rate") < 50), "很短").
            when((col("duration") >= 50) & (col("rate") < 80), "较短").
            when((col("duration") >= 80) & (col("rate") < 120), "中").
            when((col("duration") >= 120) & (col("rate") < 150), "较长").
            otherwise('很长')
        )
    )

    filter_df4 = catMovieData.filter(col("duration") != 0)
    result14 = filter_df4.groupBy("durationCategory").count()

    # sql
    result14.write.mode("overwrite"). \
        format("jdbc"). \
        option("url", "jdbc:mysql://node1:3306/bigdata?useSSL=false&useUnicode=true&charset=utf8"). \
        option("dbtable", "mTimeCategory"). \
        option("user", "root"). \
        option("password", "root"). \
        option("encoding", "utf-8"). \
        save()

    result14.write.mode("overwrite").saveAsTable("mTimeCategory", "parquet")
    spark.sql("select * from mTimeCategory").show()

    #需求15 各类型平均票房
    result15 = explode_df.groupBy("type").agg(
        avg("allBoxOffice").alias("avg_allBoxOffice")
    )

    # sql
    result15.write.mode("overwrite"). \
        format("jdbc"). \
        option("url", "jdbc:mysql://node1:3306/bigdata?useSSL=false&useUnicode=true&charset=utf8"). \
        option("dbtable", "mTypeAvgBox"). \
        option("user", "root"). \
        option("password", "root"). \
        option("encoding", "utf-8"). \
        save()

    result15.write.mode("overwrite").saveAsTable("mTypeAvgBox", "parquet")
    spark.sql("select * from mTypeAvgBox").show()

    #各国家票房
    result16 = explode_df2.groupBy("country").agg(
        avg("allBoxOffice").alias("avg_allBoxOffice")
    )

    # sql
    result16.write.mode("overwrite"). \
        format("jdbc"). \
        option("url", "jdbc:mysql://node1:3306/bigdata?useSSL=false&useUnicode=true&charset=utf8"). \
        option("dbtable", "mCountryAvgBox"). \
        option("user", "root"). \
        option("password", "root"). \
        option("encoding", "utf-8"). \
        save()

    result16.write.mode("overwrite").saveAsTable("mCountryAvgBox", "parquet")
    spark.sql("select * from mCountryAvgBox").show()


    #年度平均票房

    result17 = time_df2.groupBy("year").agg(
        avg("allBoxOffice").alias("avg_allBoxOffice")
    )

    # sql
    result17.write.mode("overwrite"). \
        format("jdbc"). \
        option("url", "jdbc:mysql://node1:3306/bigdata?useSSL=false&useUnicode=true&charset=utf8"). \
        option("dbtable", "mYearAvgBox"). \
        option("user", "root"). \
        option("password", "root"). \
        option("encoding", "utf-8"). \
        save()

    result17.write.mode("overwrite").saveAsTable("mYearAvgBox", "parquet")
    spark.sql("select * from mYearAvgBox").show()

5、源码获取方式

🍅由于篇幅限制，获取完整文章或源码、代做项目的，查看【用户名】、【专栏名称】就可以找到我啦🍅

感兴趣的可以先收藏起来，点赞、关注不迷路，下方查看👇🏻获取联系方式👇🏻