简介这是一份基于Django与MySQL实现的城市PM2.5空气质量数据可视化分析源码面向需要完成Python课程设计、毕业设计或希望快速上手Web可视化开发的读者。项目包含完整的Django工程结构内置北京、上海、广州、成都、沈阳等城市六年PM2.5数据覆盖温度、湿度、露点、风向、大气压等影响因子并提供登录注册、数据筛选、图表展示等功能。压缩包共66个文件以17个py源码文件、24个csv数据文件、4个xml配置、3个html页面和2个md文档为主同时包含SQL数据库脚本、依赖清单与详细部署文档整体大小仅12.38MB目录清晰便于按模块学习。目前已有145人学习下载。借助这份资源读者可以直接替换CSV数据完成自己的可视化分析也可从数据预处理脚本、Django视图配置和前端模板中理解完整的项目组织方式是兼具实用性与教学价值的Python高分参考项目。1. 拿到这份 PM2.5 可视化源码先认清它到底解决什么问题当一张城市空气质量数据表落到你手里你能看到的只是几百行数字哪个站点、哪个时刻、PM2.5 浓度多少。可换到业务或答辩现场对方要的是在浏览器里按城市、按时间段拖出趋势线和柱状图甚至能一眼看出哪几天污染爆表。基于 Django MySQL 实现的城市 PM2.5 空气质量数据可视化分析源码做的就是这件事用 Django 把数据库里的时序数据整理成网页接口再用 ECharts 这类前端图表把 PM2.5、PM10、AQI 变成可交互图形。它适合两类人一类是需要课程设计或求职作品的 Python 学习者想找一个“后端 数据库 可视化”都占全的完整项目另一类是环保、气象相关岗位上想把监测数据快速做成内部看板的后台开发。下面我把这套项目的表结构、导数流程、接口设计和部署排错完整拆开照着搭就能跑通。2. Django MySQL 的数据模型与项目骨架先把表设计对再谈可视化2.1 为什么是这个组合Django 管业务MySQL 扛数据PM2.5 可视化这类项目选型时最常见的两个替代方案是 Flask SQLite 和 Django SQLite。从小项目角度看 Flask 更轻但如果最终要让数据可视化长期可维护Django MySQL 的优势在几个地方Django 自带 Admin 后台、ORM、数据迁移脚本、表单校验做一个“后台录入 前端展示”的全栈项目时不用东拼西补MySQL 对多用户并发写入的支持比 SQLite 好尤其当你有定时任务每小时拉一批监测数据进来另一侧浏览器同时在查图表SQLite 很容易报出 database is locked。MySQL 有行级锁和连接池写入和查询可以并行不打架。当然选 MySQL 不是没有代价你得先装好数据库、建库建用户、把驱动包装对。这也是为什么很多新手卡在第 0 步就放弃了——Django 代码还没写MySQL 先给了一记下马威。我的建议是环境问题集中花半天解决后面收益是值得的课程设计、论文实验和企业内部看板这套组合都能直接用。2.2 数据表怎么拆站点维表、城市维表、空气质量事实表空气质量数据可视化最常见的脏做法是把所有字段塞进一张表城市名、站点名、时间、PM2.5、再带上 SO2、NO2、O3……表也能跑但等你要做“对比北京和上海 2024 年上半年日均 PM2.5”的查询时SQL 会写得非常别扭而且城市名重复存储会带来大量冗余。更合理的做法是拆成维度表和事实表城市是一张表监测站点是一张表逐小时监测记录单独放一张大表用外键关联。一张能支撑可视化的最小模型长这样# monitor/models.py from django.db import models class City(models.Model): name models.CharField(max_length32, uniqueTrue) class Meta: db_table dim_city ordering [name] def __str__(self): return self.name class Station(models.Model): code models.CharField(max_length16, uniqueTrue) # 站点编码例如 CD_1001A name models.CharField(max_length64) city models.ForeignKey(City, on_deletemodels.CASCADE, related_namestations) class Meta: db_table dim_station def __str__(self): return self.name class AirQuality(models.Model): station models.ForeignKey(Station, on_deletemodels.CASCADE, related_namerecords) monitor_time models.DateTimeField(db_indexTrue) # 监测时间查询热字段 pm25 models.DecimalField(max_digits6, decimal_places2, nullTrue) pm10 models.DecimalField(max_digits6, decimal_places2, nullTrue) aqi models.IntegerField(nullTrue) class Meta: db_table fact_air_quality unique_together (station, monitor_time) # 同一站点同一时刻只留一条这里最关键的两个设计是db_indexTrue和unique_together。monitor_time建索引是因为可视化接口基本都会按时间范围过滤不建索引的话百万行记录会全表扫unique_together则从数据库层挡住了重复数据比在导入脚本里手动判重可靠得多。2.3 创建 Django app 与连接 MySQL 的配置顺序模型设计好之后先把这个数据入库的流程走通再做接口。常见做法是手动创建名为 monitor 的应用然后配置数据库连接并执行迁移。按这个顺序走python manage.py startapp monitor python manage.py makemigrations monitor python manage.py migratestartapp会自动生成 migration、views、models 这些文件makemigrations monitor把模型翻译成迁移脚本第一次执行时如果提示 No changes detected多半是 settings 的 INSTALLED_APPS 里没有加monitor这是新手第一个高频翻车点。连接 MySQL 时settings.py 里这段配置可以直接抄# config/settings.py 片段 import os DATABASES { default: { ENGINE: django.db.backends.mysql, NAME: os.getenv(DB_NAME, air_quality), USER: os.getenv(DB_USER, root), PASSWORD: os.getenv(DB_PASSWORD, change_me), HOST: os.getenv(DB_HOST, 127.0.0.1), PORT: os.getenv(DB_PORT, 3306), OPTIONS: { charset: utf8mb4, init_command: SET sql_modeSTRICT_TRANS_TABLES, }, } }charset: utf8mb4建议保留它决定中文备注和城市名能不能正常存取init_command里的 sql_mode 会避免 MySQL 对非法日期做静默转换宁可导入报错也不能让脏日期悄悄进库。密码不要直接写死在代码里用环境变量或者本地.env都好这个项目将来要示人或者部署配置文件会被其他人看到。数据库本身需要在 MySQL 里先建好Django 的 migrate 只会建表不会帮你建库。我一般用这么一句CREATE DATABASE air_quality DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;建库这一步有讲究如果直接用默认 latin1 建库后面所有中文都是问号还得推倒重来。迁移完成后可以用python manage.py shell手工插入一条测试数据再用 SELECT 验证一遍中文存取这一步值得做避免后面接口都写完了才发现数据库字符集不对。3. 把 PM2.5 采样数据清洗入库管理命令脚本与去重策略3.1 原始数据长什么样CSV 的编码和无效值最坑从公开监测站点拿到的 PM2.5 原始数据最常见的是 CSV 或 Excel 导出的表格。列一般包括站点编号、时间、PM2.5、PM10、SO2、NO2、O3、CO 这些。这里最容易翻车的不是数据量而是三件事文件编码可能是 UTF-8 带 BOM也可能是 GBK缺测值有空白、有-999、还有--这样乱写的时间列格式不统一有的是2025-01-01 08:00有的是2025/1/1 8:00。在写任何导入脚本之前我建议先花两分钟看一眼文件头部file data/pm25_2025.csv head -n 5 data/pm25_2025.csvfile命令输出里有UTF-8 (with BOM)或ISO-8859字样直接决定你用utf-8-sig还是gbk去读。打开看到\ufeff开头的表头就说明有 BOMPython 里用encodingutf-8-sig可以自动吃掉它。3.2 写一个 Django 管理命令做批量导入bulk_create 是唯一正解手动在 shell 里一条条创建对象只适合调试。正式导数据要写成 Django management command好处是能复用项目的 settings 和 ORM将来部署到服务器上可以直接用python manage.py import_pm25 --csv ...触发。一个最小可用的导入命令如下# monitor/management/commands/import_pm25.py import csv from datetime import datetime from django.core.management.base import BaseCommand, CommandError from django.utils import timezone from monitor.models import City, Station, AirQuality def parse_float(value): 把 CSV 里的无效值统一转成 None而不是强行填充 0。 if value is None: return None value str(value).strip() if value in (, -, --, null, 999, -999): return None try: return float(value) except ValueError: return None class Command(BaseCommand): help 导入 PM2.5 站点监测 CSV 数据 def add_arguments(self, parser): parser.add_argument(--csv, requiredTrue, helpCSV 文件路径) parser.add_argument(--city, requiredTrue, help城市名例如 北京) parser.add_argument(--station, requiredTrue, help站点编码) def handle(self, *args, **options): city, _ City.objects.get_or_create(nameoptions[city]) station, _ Station.objects.get_or_create( codeoptions[station], defaults{name: options[station], city: city} ) rows [] skipped 0 with open(options[csv], encodingutf-8-sig) as fp: reader csv.DictReader(fp) for line in reader: try: t datetime.strptime(line[time], %Y-%m-%d %H:%M) except (ValueError, KeyError): skipped 1 continue rows.append(AirQuality( stationstation, monitor_timet, pm25parse_float(line.get(pm25)), pm10parse_float(line.get(pm10)), aqiint(parse_float(line.get(aqi)) or 0) or None, )) # ignore_conflictsTrue 依赖数据库层的唯一约束命中就跳过 AirQuality.objects.bulk_create(rows, batch_size2000, ignore_conflictsTrue) self.stdout.write(self.style.SUCCESS(f导入完成{len(rows)} 条跳过 {skipped} 条新增去重后数据))这段脚本里两个参数必须解释清楚。batch_size2000控制每次 INSERT 的记录数太大容易超过 MySQL 的 max_allowed_packet太小又体现不出批量插入的优势我长期用 2000 这个值没出过问题。ignore_conflictsTrue是配合模型里的unique_together使用的意思是遇到同一个站点同一时刻的重复记录直接跳过不报错、不覆盖这比先 SELECT 再 INSERT 的方式快一个数量级也不用自己维护去重逻辑。parse_float的取舍也要注意我把空值和-999这类典型缺测标记统一转成None而不是填 0。数据可视化时 0 会被画成一条贴地的线很容易被误读为“空气质量很好”而 None 可以让前端显示为空缺图表更诚实。AQI 那行写得稍微绕了一点是为了兼容 CSV 里没有 aqi 列的情况。3.3 数据质量边界脏数据处理到什么程度该收手很多人拿到数据第一反应是用 pandas 把缺测值 fillna(0)把异常值删掉再入库。这个思路对 pandas 分析没问题但对可视化项目未必合适。监测站的-999和空值只是“没测到”不代表浓度为零擅自把异常值删掉又会造成时间序列断档。我一般只在三个地方做处理格式非法的时间行直接跳过PM2.5 为负但不在-999这种约定范围内的值转 None重复记录交给数据库唯一约束去挡。其余数值保持原样入库重度清洗留给后续分析脚本不要把清洗逻辑在导入环节做死。另外有个经验一个城市可能对应多个站点导入时--station参数必须真实存在否则air_quality表里所有记录都会挂在同一个站点下后面按站点维度的图表全都会失真。如果你的 CSV 里本身带了站点编号列那就应该从 CSV 中读取站点而不是用命令行参数硬指定我最早就是偷懒用命令行参数结果导了十几个站点的数据全部串成了同一个站点。4. 可视化层怎么搭REST 接口 ECharts 图表联动4.1 只读接口不需要 DRFDjango 原生 View 就够用代码里如果数据表和导入都搞定了接下来就是把这个项目最好看的可视化部分搭起来。很多人在这一步会直接引入 Django REST Framework写 serializer、写 router。但如果你只是要从数据库读出聚合结果、返回 JSON 给前端原生 Django View 完全够用少一层依赖部署时也少一点版本兼容问题。等以后真要加登录鉴权、分页、过滤、API 文档再迁移到 DRF 也不迟。以“城市 PM2.5 趋势”接口为例最核心的视图可以这样写# monitor/views.py from django.http import JsonResponse from django.utils.dateparse import parse_datetime from django.views import View from django.db.models import Avg from django.db.models.functions import TruncHour, TruncDay from monitor.models import AirQuality class CityTrendView(View): 按城市和时间范围返回 PM2.5 聚合值支持按小时或按天聚合。 def get(self, request, city_id): start parse_datetime(request.GET.get(start, )) end parse_datetime(request.GET.get(end, )) if start is None or end is None or start end: return JsonResponse({error: start 和 end 参数必填且 start 需早于 end}, status400) granularity request.GET.get(granularity, day) trunc_expr TruncHour(monitor_time) if granularity hour else TruncDay(monitor_time) rows ( AirQuality.objects .filter(station__city_idcity_id, monitor_time__range(start, end)) .annotate(buckettrunc_expr) .values(bucket) .annotate(avg_pm25Avg(pm25)) .order_by(bucket) ) payload [ { t: item[bucket].strftime(%Y-%m-%d %H:%M), pm25: round(item[avg_pm25], 2) if item[avg_pm25] is not None else None, } for item in rows ] return JsonResponse(payload, safeFalse)这个视图有几个细节值得展开。TruncHour和TruncDay是 Django 提供的数据库时间截断函数它会在 SQL 层完成DATE_FORMAT这类操作比把所有明细拉回 Python 再按时间分组快得多也避免了时区被 Python 侧二次解释。values(bucket)配合annotate(Avg(pm25))是标准的 group by 写法注意聚合结果字段名是avg_pm25前端拿到的 JSON 里就是这个键不要写成pm25__avg这种自动名那样前端代码会很难看。avg_pm25可能是 None响应里保留显式 null让前端图表能跳过空缺点。URL 配置同样很直接不需要注册到 DRF router# config/urls.py from django.urls import path from monitor.views import CityTrendView urlpatterns [ path(api/city/int:city_id/trend/, CityTrendView.as_view(), namecity-trend), ]int:city_id会做参数类型转换传入视图的是一个 Python int不是字符串。接口写好后先用 Django 测试客户端试一下python manage.py shell里用Client().get(/api/city/1/trend/, {start: ..., end: ...})这一步确认能返回 JSON再写前端避免前后端同时出错时不知道锅该甩给谁。4.2 ECharts 接数据折线图先跑通再扩展热力图前端可视化我默认用 ECharts因为国内能检索到的数据可视化示例几乎一半以上都是用 ECharts 做的折线图、柱状图、地图热力图都有成熟配置。页面里通常先准备一个容器 div再在脚本里初始化图表// templates/index.html 中内联或单独 static/js/trend.js const chart echarts.init(document.getElementById(trendChart)); function loadTrend(cityId, start, end) { fetch(/api/city/${cityId}/trend/?start${encodeURIComponent(start)}end${encodeURIComponent(end)}granularityday) .then(res res.json()) .then(data { chart.setOption({ tooltip: { trigger: axis }, xAxis: { type: category, data: data.map(d d.t) }, yAxis: { type: value, name: μg/m³ }, series: [{ name: PM2.5, type: line, connectNulls: false, data: data.map(d d.pm25) }] }); }); }connectNulls: false在这里很关键。接口返回的pm25字段如果是 null折线要断开而不是用直线跨越缺口否则会让人误以为那段时间浓度是连续变化的。如果数据跨度是一整年、粒度选了 hour折线会上万个点浏览器渲染会卡顿此时要么把粒度切到 day要么在接口里再做一次 LIMIT。接口里没有写死 LIMIT是因为这类内部看板的需求变化很快我个人更倾向于在 URL 参数里暴露 granularity把前端的展示压力交给用户选择。4.3 页面模板与静态资源别把接口地址写死在 localhost模板页面放在 Django 的 templates 目录图表 JS 和 ECharts 库放在 static 目录。有一个特别常见的坑是本地开发时接口地址写http://127.0.0.1:8000/api/...一切正常部署到服务器后就空白因为浏览器地址变了但 JS 里的 localhost 没变。我一般会在模板里把接口前缀注入到全局变量!-- templates/index.html -- script window.API_BASE {% url city-trend city_id1 %}.replace(/api/city/1/trend/, /api); /script这样前端只要基于API_BASE拼路径本地和服务器都能用相对地址省去部署时改代码的麻烦。ECharts 本身是纯前端库建议把它下载到static/vendor/echarts.min.js不要用 CDN。城市级可视化项目经常部署在不能访问外网的内网环境CDN 一断图表就全白这是很多新手连 ECharts 的边都摸不到就开始怀疑后端接口的原因。5. 部署与开发排错PM2.5 可视化项目最常见的 5 个坑5.1 MySQL 8.0 连不上Authentication plugin 报错现象python manage.py migrate时报django.db.utils.OperationalError: Authentication plugin caching_sha2_password cannot be loaded。原因MySQL 8.0 默认认证插件是 caching_sha2_password而 PyMySQL 版本过旧或系统里的 mysqlclient 不认这个插件。解决升级依赖库在 requirements.txt 里把 PyMySQL 版本写新一点并在项目的__init__.py里显式注册# config/__init__.py import pymysql pymysql.install_as_MySQLdb()如果你用的是 mysqlclient则优先检查系统是否装了 libmysqlclient-dev 这类底层库。也可以在 MySQL 里为项目单独创建用户指定使用 mysql_native_password 插件但这属于临时方案新库还是建议升级驱动毕竟 MySQL 8.4 之后 mysql_native_password 也开始被边缘化。5.2 中文乱码写入是问号读出来也是问号现象城市名“北京”在 Django Admin 里显示为北京或??。原因建库时用了默认字符集或者表不是 utf8mb4或者 Django 连接串没指定 charset。解决从三个层面拉齐——建库时显式指定DEFAULT CHARACTER SET utf8mb4settings 的OPTIONS里写charset: utf8mb4打开 MySQL 客户端连接时执行SET NAMES utf8mb4。数据已经乱掉的场景没有后悔药只能清空重导所以建库时这道命令一定不要省。5.3 时间错位早上 8 点的数据跑到了 0 点现象接口返回的 JSON 里时间比原始 CSV 少了 8 小时或者图表横轴每天从 16 点开始。原因Django 默认TIME_ZONE UTC而 CSV 里的时间是北京时间Django 的 DateTimeField 一旦启用USE_TZ True存库时会按 UTC 转换。解决settings 里改时区配好后重新导数据LANGUAGE_CODE zh-hans TIME_ZONE Asia/Shanghai USE_TZ True注意这里USE_TZ True保留让 ORM 自动处理时区转换配合前端只显示monitor_time的本地时间字符串即可。另一个相关问题是聚合查询里的TruncHour在 MySQL 中会基于数据库会话时区计算最好在连接数据库前确认连接时区。血泪经验是先写一个脚本随机抽几条记录比对 CSV 原时间和接口返回时间再做可视化别等图表出来了才发现整体平移了 8 小时。5.4 接口慢一次查一年数据卡了 5 秒现象折线图加载时接口耗时 4000ms请求体 200KB前端渲染也卡。原因粒度使用了小时但查询跨度是一年返回上万个聚合点或者模型没写db_index时间范围过滤是全表扫描。解决先把接口默认粒度改成 day再用EXPLAIN或 Django 的connection.queries看有没有走索引最后在AirQuality表上加复合索引让过滤和排序同时受益class Meta: db_table fact_air_quality unique_together (station, monitor_time) indexes [ models.Index(fields[monitor_time, station], nameidx_time_station), ]做了这一步同一份数据接口耗时基本能降到原来的十分之一。可视化项目一旦数据进入十万行索引和聚合粒度就是两个最关键的调优点。5.5 图表白屏HTML 加载了但 ECharts 不渲染现象打开页面看到按钮和标题图表区域空白浏览器 F12 里报echarts is not defined或 404。原因引用的是 CDN 上的 echart.min.js内网环境加载失败或者 Django 的DEBUG False时静态文件路由失效。解决先把 ECharts 库下载到本地 static然后确认 settings 配置STATIC_URL /static/ STATICFILES_DIRS [BASE_DIR / static]开发阶段保持DEBUG True部署到 Nginx 时把/static/一起代理给 Nginx 处理不要让 Django 承担静态文件服务。如果 Nginx 配好后仍然白屏先 curl 一下http://你的域名/static/vendor/echarts.min.js确认这个文件在服务器上真实存在而不是部署时 static 目录没同步过去。6. 验证接口与进阶方向从“能跑”到“能上线”6.1 用 Django TestCase 给接口上保险这个项目最脆弱的部分不是前端而是接口的查询参数。只要有人把start传错格式或者数据库里混进了空值接口就会花式报错。我会用 Django 自带的 TestCase 写最小回归验证确保改表结构或加索引时不破坏接口# monitor/tests.py from django.test import TestCase from django.test import Client from django.utils import timezone from monitor.models import City, Station, AirQuality class CityTrendApiTest(TestCase): def setUp(self): self.city City.objects.create(name测试城市) self.station Station.objects.create(codeTEST_01, name测试站, cityself.city) AirQuality.objects.create( stationself.station, monitor_timetimezone.now(), pm2512.5, pm1035.0, ) def test_trend_returns_json_with_daily_avg(self): resp Client().get(/api/city/{}/trend/.format(self.city.id), { start: 2025-01-01T00:00, end: 2025-12-31T23:59, granularity: day, }) self.assertEqual(resp.status_code, 200) self.assertIn(pm25, resp.json()[0])这个测试断言了接口返回 200 且包含pm25字段后面你改接口参数名、改聚合字段名时测试能直接提醒你前端同事也会跟着遭殃。跑一遍python manage.py test monitor几秒钟就能证明这个可视化项目的后端是可靠的。6.2 从看板到监测产品AQI 分级、定时采集、预警推送基础可视化跑通之后值得投入的三个方向是加 AQI 计算逻辑和污染等级配色而不是只展示一个 PM2.5 数值用系统 crontab 或 Celery beat 每小时拉一次监测站数据保持看板实时在 PM2.5 连续三小时超过阈值时给业务群推送告警。这三个方向都能在你的课程设计说明或项目文档里作为亮点出现也和我做项目的习惯一致——可视化不是终点是数据被人使用和响应的起点。做第一个版本时我曾在时区配置上翻了车整张图表和真实监测数据差了 8 小时后来每次接到时序数据项目第一件事就是先把“源数据时间 → 数据库存储时间 → 接口返回时间”全链路比对一遍。这个习惯帮我避掉了后续大量返工。如果你照着这套步骤把 PM2.5 可视化项目跑通后面换城市、换污染因子、换数据库都只是参数级改动希望帮到你。本文还有配套的精品资源点击获取