简介这份资源面向环境科学、数据挖掘与机器学习方向的学习者和研究者提供一套基于Python的空气质量数据可视化分析系统源码及配套数据可用于课程设计、毕业设计或科研复现。项目采用BS架构前端整合HTML、CSS、JavaScript与D3、ECharts、Mapbox等可视化库后端基于Python与Flask构建核心思路是先挖掘空气质量数据的时空特征完成降维聚类与城市群划分再通过粒子输运和相关分析构建大气污染传输网络并借助多维空间变换的渐进式探索方法分析污染传播过程。压缩包共约2000个文件以1295个json数据文件和697个py脚本为主另含少量txt、xml与md说明整体约76.43MB其中files目录存放原始数据与挖掘结果dataManager.py为后端数据模块main.py为服务入口。目前已有253人学习适合希望掌握数据挖掘、机器学习与可视化全流程的读者参考。1. 从一堆 JSON 到城市污染传输网络这套空气质量挖掘系统能跑出什么手里拿到一批按日期命名的空气质量 JSON 文件第一反应往往是「先画个折线图看看」。但真做过大气污染分析的人都知道单站点、单指标的时序曲线几乎说明不了问题——AQI 高的时候你分不清是本地排放还是外来输送低的时候也判断不出区域联防到底有没有生效。这套基于 Python 的空气质量数据挖掘与机器学习可视化分析系统解决的正是这个层面的问题它把原始监测数据先做降维聚类完成城市群划分再通过粒子输运和相关分析构建大气污染传输网络APTN最后用多维空间变换做渐进式探索把污染的传播过程摊开给人看。技术栈上它是标准的 BS 架构前端 HTML/CSS/JavaScript 打底可视化用了 D3、ECharts、Mapbox 三件套后端是 Python Flask。源码包里main.py是服务入口dataManager.py负责后端数据模块files文件夹同时存了原始数据和挖掘后的中间数据templates/index.html是前端入口。适合谁做环境数据方向的学生、需要快速搭一个空气质量分析原型的工程师以及想拿一套完整「数据挖掘 机器学习 可视化」链路练手的人。下面按「资源是什么 → 怎么跑起来 → 坑在哪 → 怎么改」的顺序拆。2. 数据层拆解JSON 结构、降维聚类与 APTN 是怎么串起来的2.1 原始 JSON 的字段结构与读取方式files里的文件按YYYYMMDD.json命名从 20130106 到 20170104 横跨几年说明这是一份带时间跨度的多站点监测数据。这类文件常见结构是「城市/站点 污染物浓度 时间戳」读取时最容易翻车的地方是编码和嵌套层级。我一般先用一段脚本把结构摸清楚再决定怎么进 DataFrame。import json import os import pandas as pd DATA_DIR files def peek_structure(filepath): 打印单个 JSON 的顶层结构避免盲目解析 with open(filepath, r, encodingutf-8) as f: obj json.load(f) if isinstance(obj, dict): print(顶层键:, list(obj.keys())[:10]) # 常见嵌套{data: [...]} 或 {city: {AQI: ...}} for k, v in obj.items(): print(f {k} - {type(v).__name__}) elif isinstance(obj, list): print(列表长度:, len(obj), 首元素键:, list(obj[0].keys())) return obj # 先看一个文件别急着批量读 sample peek_structure(os.path.join(DATA_DIR, 20130108.json))逻辑说明peek_structure只做结构探测不写死字段名因为不同来源的空气质量 JSON 差异很大——有的把 AQI 和 PM2.5 平铺有的按城市再嵌一层。参数上encodingutf-8是必须显式指定的国内监测数据偶尔带 BOM 头遇到json.decoder.JSONDecodeError时先换成utf-8-sig试一次。批量读取时建议用pd.concat而不是逐行 append后者在几千个文件规模下会慢到让人怀疑人生。2.2 降维聚类做城市群划分为什么不是直接按省份分摘要里提到的「基于降维聚类的城市群划分」是这套系统的第一层核心。直接按行政省份分组是最省事的做法但空气质量不认省界——河北南部的污染和河南北部往往比和河北北部更同步。所以正确做法是把每个城市的多污染物时序当成一个高维向量先降维再聚类。常见做法是 PCA 降到 35 维保留主要方差再用 KMeans 或层次聚类分组。降维前必须做标准化否则 PM2.5数值几十到几百会完全压过 CO数值个位数的贡献。from sklearn.preprocessing import StandardScaler from sklearn.decomposition import PCA from sklearn.cluster import KMeans # df 为 城市 x 污染物特征 的宽表index 是城市名 X df.values X_scaled StandardScaler().fit_transform(X) # 标准化量纲统一 pca PCA(n_components0.9) # 保留 90% 方差自动定维数 X_pca pca.fit_transform(X_scaled) print(保留主成分数:, pca.n_components_, 累计方差:, pca.explained_variance_ratio_.sum()) # 聚类数 k 用肘部法或轮廓系数定别拍脑袋 km KMeans(n_clusters5, random_state42, n_init10) labels km.fit_predict(X_pca) df[cluster] labels参数说明n_components0.9表示按累计方差比例自动选维比写死维数稳n_init10是 KMeans 多次初始化取最优sklearn 新版本默认值变过显式写死避免版本差异导致结果漂移random_state固定是为了复现。聚类数 k 是这套流程里最玄学的参数建议同时看肘部图和轮廓系数别只看一个。2.3 APTN 大气污染传输网络相关分析怎么变成有向图城市群分完之后第二层是构建大气污染传输网络。思路是如果城市 A 的污染物浓度变化总是领先城市 B 若干小时且两者相关性显著就认为存在 A→B 的潜在传输。这本质上是把「时滞相关」变成有向边。import numpy as np from scipy.stats import pearsonr def build_aptn(series_dict, max_lag12, threshold0.5): series_dict: {城市: 时序数组}返回有向边列表 cities list(series_dict.keys()) edges [] for i, a in enumerate(cities): for b in cities: if a b: continue best_r, best_lag 0, 0 for lag in range(1, max_lag 1): # a 领先 b lag 步 x, y series_dict[a][:-lag], series_dict[b][lag:] if len(x) 10: continue r, _ pearsonr(x, y) if abs(r) abs(best_r): best_r, best_lag r, lag if abs(best_r) threshold: edges.append((a, b, round(best_r, 3), best_lag)) return edges逻辑说明max_lag12假设最大传输时滞 12 个时间步若数据是小时级即 12 小时这个值要按数据粒度调threshold0.5是相关性门槛调低会引入大量弱边让网络变成一团毛线调高则可能漏掉真实传输。pearsonr要求两序列等长所以用切片对齐。注意这里算的是线性相关污染传输在冬季供暖期常有非线性突变必要时换成互信息或 Granger 因果检验会更稳。3. 把服务跑起来Flask 后端、前端可视化与数据接口对接3.1 main.py 启动与 Flask 路由结构main.py是服务入口dataManager.py是数据模块这种拆分意味着后端逻辑和数据处理是解耦的。启动前先确认依赖装齐Flask、pandas、scikit-learn、numpy 是跑不掉的。# 建议用虚拟环境别污染全局 python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install flask pandas numpy scikit-learn scipy python main.py启动后默认多半跑在127.0.0.1:5000。如果端口被占改main.py里app.run(portxxxx)即可。这里有个血泪经验dataManager.py里如果用了相对路径读files从别的目录启动服务会直接报文件找不到所以要么用os.path.dirname(__file__)拼绝对路径要么保证在项目根目录启动。3.2 前端三件套D3、ECharts、Mapbox 各管什么前端用了三个可视化库分工其实很清楚。ECharts 负责常规图表——时序折线、柱状对比、聚类结果的散点分布它的配置项丰富、上手快是主力。D3 负责需要高度自定义的图形比如 APTN 那种带箭头和权重的网络图ECharts 的 graph 也能画但定制自由度不如 D3。Mapbox 负责地理维度把城市群划分和传输路径落到地图上。templates/index.html是入口static放静态资源。对接后端时前端通过 fetch 调 Flask 接口拿 JSON再喂给对应图表。常见做法是后端统一返回{code:0, data:[...]}结构前端按code判断成败别让前端去猜返回格式。// 前端拉取聚类结果并渲染到 ECharts fetch(/api/cluster) .then(res res.json()) .then(res { if (res.code ! 0) { console.error(接口异常, res.msg); return; } const chart echarts.init(document.getElementById(clusterChart)); chart.setOption({ xAxis: { type: value }, yAxis: { type: value }, series: [{ type: scatter, data: res.data, // [[x, y, clusterId], ...] symbolSize: 8 }] }); });逻辑说明res.code ! 0是防御性判断后端出错时前端不至于白屏。data里每个点带 clusterId实际项目里通常按 clusterId 分组上色这里为简洁只画散点。ECharts 的init必须在 DOM 渲染完成后调用放在head里直接执行会拿到 null 容器这是新手最常翻的车。3.3 数据接口与 files 目录的对应关系files里既有原始 JSON 也有挖掘后的数据dataManager.py的职责就是决定接口读哪一份。合理的设计是原始数据只在预处理阶段读一次生成聚类和网络结果后落盘接口直接读结果文件避免每次请求都重跑一遍 PCA 和聚类。如果发现接口响应慢先看dataManager.py是不是每次都在实时计算——这是性能问题最常见的根因。4. 避坑与排查这套源码跑不起来时先看这几条4.1 现象启动报 ModuleNotFoundError缺的却是没听过的包原因dataManager.py或测试脚本里 import 了sklearn、scipy甚至networkx这类非标准库但项目没给requirements.txt只能靠报错一个个补。解决按报错顺序pip install遇到sklearn要装scikit-learn包名和导入名不一致是经典坑遇到网络图相关报错补networkx。装完再启动别一次装一堆。4.2 现象JSON 读取报编码错误或字段 KeyError原因文件带 BOM 头或不同年份的文件字段结构不一致——2013 年的文件和 2017 年的文件很可能不是同一套字段。解决读取时用utf-8-sig解析前先peek_structure探测结构对缺失字段用obj.get(PM2.5, None)兜底别用obj[PM2.5]硬取。4.3 现象聚类结果每次跑都不一样原因KMeans 随机初始化没固定或 PCA 前忘了标准化导致量纲主导结果。解决random_state写死StandardScaler必须在PCA之前顺序反了结果没有可比性。4.4 现象前端图表空白但接口有数据原因ECharts 容器在 DOM 未就绪时初始化或 Mapbox 缺 access token。解决把图表初始化放到DOMContentLoaded之后Mapbox 的 token 在初始化mapboxgl.accessToken时补上缺 token 地图会静默失败不报错。4.5 现象APTN 网络图边多到看不清原因相关性阈值设太低弱相关边全进来了。解决把threshold从 0.5 往上调到 0.60.7或只保留每个城市 top-N 强边网络图的可读性比完整性更重要。5. 进阶玩法把静态分析改成可交互的渐进式探索这套系统摘要里提到的「基于多维空间变换的渐进式探索」是它区别于普通可视化项目的地方。普通做法是一次性把所有城市所有污染物画出来信息过载渐进式探索是先给全局概览用户点选某个城市群后再下钻到该群的污染物细节再点某个城市看它的传输上下游。落地时可以在 Flask 加一个带参数的接口前端根据用户点击动态请求。from flask import Flask, request, jsonify app Flask(__name__) app.route(/api/drill) def drill(): cluster_id request.args.get(cluster, typeint) city request.args.get(city, typestr) # 按层级过滤先按群再按城市 result dataManager.query(cluster_idcluster_id, citycity) return jsonify({code: 0, data: result})参数说明request.args.get(..., typeint)会自动做类型转换传非整数时返回 None 而不是抛异常比手动int()安全。前端每次点击只请求当前层级需要的数据避免一次性传输全量数据。验证这套流程是否跑通我一般看三个信号聚类结果的轮廓系数是否大于 0.3、APTN 边数是否在合理范围几十到几百条不是几千条、前端下钻时接口响应是否在几百毫秒内。三个都过基本说明数据链路是通的。从那以后我每次拿到这种「源码 数据」的包都强制先跑一遍结构探测脚本再动手改代码而不是上来就python main.py——省下的排错时间远比写探测脚本多。希望帮到你。本文还有配套的精品资源点击获取