大数据毕设高口碑项目！基于 Hadoop+Spark 的城市空气质量分析可视化系统

💖💖作者：计算机编程小央姐
💙💙个人简介：曾长期从事计算机专业培训教学，本人也热爱上课教学，语言擅长Java、微信小程序、Python、Golang、安卓Android等，开发项目包括大数据、深度学习、网站、小程序、安卓、算法。平常会做一些项目定制化开发、代码讲解、答辩教学、文档编写、也懂一些降重方面的技巧。平常喜欢分享一些自己开发中遇到的问题的解决办法，也喜欢交流技术，大家有技术代码这一块的问题可以问我！
💛💛想说的话：感谢大家的关注与支持！ 💜💜

💕💕文末获取源码

大数据毕设高口碑项目！基于 Hadoop+Spark 的城市空气质量分析可视化系统-系统功能介绍

本系统是一套面向城市空气质量监测场景的大数据分析与可视化平台，数据来源于 OpenAQ 全球城市空气质量日度监测数据集，涵盖墨西哥城、北京、首尔、波哥大、曼谷、利马等 8 个城市、72 个监测站、约 3.3 万条记录，覆盖 PM2.5、PM10、NO2、SO2、O3、CO 六种主要污染物。系统后端采用 Hadoop 分布式存储与 Spark 分布式计算框架，使用 Spark SQL 完成多维度的数据清洗、分组聚合与统计分析，同时基于 K-Means、孤立森林、PCA 主成分分析和 GMM 高斯混合模型等无监督算法完成站点污染画像聚类、异常站点识别、污染物主导因子提取和站点软归属概率分析。后端以 Python 配合 Django 框架提供接口服务，前端基于 Vue、ElementUI 和 Echarts 实现交互式图表展示，支持按城市、站点类型、时间季节等维度联动筛选，最终将分析结果以柱状图、折线图、热力图、雷达图等形式直观呈现，整套流程从 HDFS 数据上传、Spark 离线计算到 MySQL 落库、前端渲染形成闭环。

大数据毕设高口碑项目！基于 Hadoop+Spark 的城市空气质量分析可视化系统-系统技术介绍

大数据框架：Hadoop+Spark（本次没用Hive，支持定制）
开发语言：Python+Java（两个版本都支持）
后端框架：Django+Spring Boot(Spring+SpringMVC+Mybatis)（两个版本都支持）
前端：Vue+ElementUI+Echarts+HTML+CSS+JavaScript+jQuery
详细技术点：Hadoop、HDFS、Spark、Spark SQL、Pandas、NumPy
数据库：MySQL

大数据毕设高口碑项目！基于 Hadoop+Spark 的城市空气质量分析可视化系统-系统背景意义

随着各地空气质量监测站越建越多，每天积累下来的污染物浓度数据量也在不断变大，这些数据如果只靠 Excel 去拉表格、做统计，速度慢不说，稍微复杂一点的交叉分析就很吃力。传统单机处理方式在面对几万条、十几列字段的监测记录时，内存和计算能力都有点跟不上，所以想借这次毕设的机会，试着用 Hadoop 加 Spark 这套大数据框架来跑一跑真实的空气质量数据，看看分布式计算在这种场景下到底好不好用。题目本身也不算新颖，就是把课堂上学过的大数据技术落到一个具体的行业数据集上，过程中会涉及数据上传 HDFS、Spark SQL 聚合、无监督算法建模，最后再用 Echarts 把结果画出来，整体走一遍从数据到图表的完整流程。这套系统做出来以后，最直接的用处就是能把那几万条监测记录比较快地跑出城市排名、季节波动、站点聚类这些分析结果，比手动处理省事不少，对自己熟练掌握 Spark 和 Pandas 这套工具链也有帮助。从实际角度说，把空气质量数据用可视化方式摆出来，哪怕只是课程作业层面，也能让看的人一眼看出哪个城市 PM2.5 偏高、哪个季节污染更严重，算是把数据变成了能看懂的东西。另外 K-Means 和 PCA 这两块代码写下来，也算是把机器学习里无监督那部分常用算法过了一遍，对后面继续看同类文献或者做类似课题都算是打了点基础。当然这只是一个本科毕设水平的小系统，离真正能上线给环保部门用还差得远，但作为一次完整的大数据项目实践，该踩的坑基本都踩到了。

大数据毕设高口碑项目！基于 Hadoop+Spark 的城市空气质量分析可视化系统-系统演示视频

演示视频

大数据毕设高口碑项目！基于 Hadoop+Spark 的城市空气质量分析可视化系统-系统演示图片

大数据毕设高口碑项目！基于 Hadoop+Spark 的城市空气质量分析可视化系统-系统部分代码

frompyspark.sqlimportSparkSession,functionsasFfrompyspark.ml.featureimportVectorAssembler,StandardScalerfrompyspark.ml.clusteringimportKMeansfrompyspark.ml.featureimportPCA spark=SparkSession.builder \.appName("AirPollutionAnalysis")\.config("spark.sql.adaptive.enabled","true")\.getOrCreate()# ===== 核心功能一：各城市六种污染物均值对比 + PM2.5 超标率分析 =====df=spark.read.csv("/user/hadoop/air_pollution/City_Types_OpenAQ_daily_complete_60k_balanced_labeled.csv",header=True,inferSchema=True)df_clean=df.na.fill(0)df_with_month=df_clean.withColumn("year",F.year(F.to_date("Date"))).withColumn("month",F.month(F.to_date("Date")))city_avg=df_with_month.groupBy("City").agg(F.round(F.avg("PM2.5"),2).alias("pm25_avg"),F.round(F.avg("PM10"),2).alias("pm10_avg"),F.round(F.avg("NO2"),5).alias("no2_avg"),F.round(F.avg("SO2"),5).alias("so2_avg"),F.round(F.avg("O3"),5).alias("o3_avg"),F.round(F.avg("CO"),5).alias("co_avg"))over_days=df_with_month.filter(F.col("PM2.5")>75).groupBy("City").agg(F.count("*").alias("over_days"))total_days=df_with_month.groupBy("City").agg(F.count("*").alias("total_days"))over_rate=over_days.join(total_days,"City").withColumn("over_rate",F.round(F.col("over_days")/F.col("total_days")*100,2))result=city_avg.join(over_rate,"City").orderBy(F.col("pm25_avg").desc())result=result.withColumn("city_cn",F.when(F.col("City")=="Beijing","北京").when(F.col("City")=="Seoul","首尔").when(F.col("City")=="Mexico City","墨西哥城").otherwise(F.col("City")))result.write.mode("overwrite").csv("/user/hadoop/air_pollution/output/city_pollution_avg_analysis.csv",header=True)# ===== 核心功能二：站点污染聚类画像分析（K-Means） =====station_profile=df_clean.groupBy("station_id").agg(F.avg("PM2.5").alias("pm25"),F.avg("PM10").alias("pm10"),F.avg("NO2").alias("no2"),F.avg("SO2").alias("so2"),F.avg("O3").alias("o3"),F.avg("CO").alias("co"))assembler=VectorAssembler(inputCols=["pm25","pm10","no2","so2","o3","co"],outputCol="features_raw")feature_vec=assembler.transform(station_profile)scaler=StandardScaler(inputCol="features_raw",outputCol="features",withStd=True,withMean=True)scaled_data=scaler.fit(feature_vec).transform(feature_vec)kmeans=KMeans(featuresCol="features",k=4,seed=42)kmeans_model=kmeans.fit(scaled_data)clustered=kmeans_model.transform(scaled_data)cluster_with_desc=clustered.withColumn("cluster_desc",F.when(F.col("prediction")==0,"高PM2.5高PM10型").when(F.col("prediction")==1,"臭氧突出型").when(F.col("prediction")==2,"综合清洁型").otherwise("工业SO2型"))cluster_with_desc.select("station_id","prediction","cluster_desc").write.mode("overwrite").csv("/user/hadoop/air_pollution/output/station_cluster_analysis.csv",header=True)# ===== 核心功能三：污染物综合得分降维分析（PCA） =====pca_src=df_clean.select("PM2.5","PM10","NO2","SO2","O3","CO")pca_assembler=VectorAssembler(inputCols=["PM2.5","PM10","NO2","SO2","O3","CO"],outputCol="pca_raw")pca_vec=pca_assembler.transform(pca_src)pca_scaler=StandardScaler(inputCol="pca_raw",outputCol="pca_features",withStd=True,withMean=True)pca_scaled=pca_scaler.fit(pca_vec).transform(pca_vec)pca=PCA(k=2,inputCol="pca_features",outputCol="pca_pc")pca_model=pca.fit(pca_scaled)pca_transformed=pca_model.transform(pca_scaled)explained=pca_model.explainedVariance loadings=pca_model.pc vars_col=["PM2.5","PM10","NO2","SO2","O3","CO"]loadings_rows=[]fori,nameinenumerate(vars_col):loadings_rows.append((name,float(loadings.row(0)[i]),float(loadings.row(1)[i]),float(explained[i])ifi<len(explained)else0.0))loadings_df=spark.createDataFrame(loadings_rows,schema=["variable","pc1","pc2","var_explained"])loadings_df.write.mode("overwrite").csv("/user/hadoop/air_pollution/output/pca_loadings_analysis.csv",header=True)

大数据毕设高口碑项目！基于 Hadoop+Spark 的城市空气质量分析可视化系统-结语

💟💟如果大家有任何疑虑，欢迎在下方位置详细交流。