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

@TOC

中国水污染监测数据可视化分析系统系统介绍

本系统《基于大数据的中国水污染监测数据可视化分析系统》面向海量水质监测数据，依托Hadoop分布式文件系统完成原始CSV与API接口数据的可靠存储，利用Spark内存计算框架及Spark SQL进行亿级记录的并行清洗、缺失值补全与异常检测，随后借助Pandas与NumPy在Python端进行二次特征工程，提炼出溶解氧、氨氮、高锰酸盐指数等核心污染物指标，并将处理结果回流MySQL以支撑在线查询；后端以Django ORM与Spring Boot双版本实现RESTful接口，分别封装用户管理、水污染监测数据管理、污染成因探索分析、水质综合评价分析、核心污染物深度分析、水质时空分布分析六大功能模块，确保高并发场景下的稳定调用；前端采用Vue与ElementUI构建响应式单页应用，通过Echarts与jQuery联动渲染动态折线、热力、迁徙及3D柱状图，直观呈现各流域、各时段水质等级演变规律及污染扩散轨迹，帮助用户一键定位突变点、生成可交互的时空分布报告，从而为流域治理决策提供精准的大数据支持。

中国水污染监测数据可视化分析系统系统演示视频

[演示视频]https://www.bilibili.com/video/BV1jCt2zmE6x/?spm_id_from=333.1387.homepage.video_card.click

中国水污染监测数据可视化分析系统系统演示图片

登陆界面.png

用户管理.png

核心污染物深度分析.png

水污染监测数据管理.png

水质时空分布分析.png

水质综合评价分析.png

污染成因探索分析.png

大屏上.png

大屏下.png

中国水污染监测数据可视化分析系统系统代码展示

# 1. 污染成因探索分析：spark_sql_correlation.py

def explore_pollution_cause():

    df = spark.read.parquet("hdfs:///clean/water/*.parquet")

    df = df.filter(df.indicator.isin(["NH3N", "TP", "CODMn"]))

    corr_df = df.groupBy("station_id").agg(

        corr("NH3N", "TP").alias("corr_nh3n_tp"),

        corr("NH3N", "CODMn").alias("corr_nh3n_cod"),

        corr("TP", "CODMn").alias("corr_tp_cod"),

        avg("temperature").alias("avg_temp"),

        avg("flow").alias("avg_flow")

    ).filter("corr_nh3n_tp is not null")

    corr_df.write.mode("overwrite").jdbc(

        "jdbc:mysql://localhost:3306/water", "pollution_correlation",

        properties={"user": "root", "password": "123456", "driver": "com.mysql.jdbc.Driver"}

    )

# 2. 水质综合评价分析：django_wqi.py

def calculate_wqi(request):

    import pandas as pd

    from django.http import JsonResponse

    df = pd.read_sql(

        "SELECT station_id, ph, do, nh3n, tp, codmn, date "

        "FROM water_quality WHERE date BETWEEN %s AND %s",

        connection, params=[request.GET['start'], request.GET['end']]

    )

    df['wqi'] = (

        0.12 * (df['ph'] - 7).abs() +

        0.18 * (14.6 - df['do']).clip(lower=0) +

        0.15 * df['nh3n'] +

        0.15 * df['tp'] +

        0.4  * df['codmn']

    )

    df['grade'] = pd.cut(df['wqi'], bins=[-1, 20, 40, 60, 80, 100],

                         labels=['Ⅰ', 'Ⅱ', 'Ⅲ', 'Ⅳ', 'Ⅴ'])

    df.to_sql("wqi_result", connection, if_exists='replace', index=False)

    return JsonResponse({"msg": "WQI computed"})

# 3. 水质时空分布分析：springboot_spatial.py

@PostMapping("/api/spatial")

public ResponseEntity<Void> spatialDistribution(@RequestBody Map<String,String> p) {

    Dataset<Row> df = spark.read()

        .option("header", true)

        .csv("hdfs:///clean/water/*.csv")

        .filter(col("date").between(p.get("start"), p.get("end")))

        .groupBy(window(col("timestamp"), "1 hour"), col("station_id"))

        .agg(

            avg("nh3n").alias("nh3n"),

            avg("tp").alias("tp"),

            avg("codmn").alias("codmn"),

            first("longitude").alias("lng"),

            first("latitude").alias("lat")

        )

        .withColumn("hour", hour(col("window.start")))

        .orderBy("hour");

    df.write()

        .mode(SaveMode.Overwrite)

        .format("json")

        .save("hdfs:///result/spatial/");

    return ResponseEntity.ok().build();

中国水污染监测数据可视化分析系统系统文档展示

文档.png