简介：这是一份基于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_length=32, unique=True) class Meta: db_table = "dim_city" ordering = ["name"] def __str__(self): return self.name class Station(models.Model): code = models.CharField(max_length=16, unique=True) # 站点编码，例如 CD_1001A name = models.CharField(max_length=64) city = models.ForeignKey(City, on_delete=models.CASCADE, related_name="stations") class Meta: db_table = "dim_station" def __str__(self): return self.name class AirQuality(models.Model): station = models.ForeignKey(Station, on_delete=models.CASCADE, related_name="records") monitor_time = models.DateTimeField(db_index=True) # 监测时间，查询热字段 pm25 = models.DecimalField(max_digits=6, decimal_places=2, null=True) pm10 = models.DecimalField(max_digits=6, decimal_places=2, null=True) aqi = models.IntegerField(null=True) class Meta: db_table = "fact_air_quality" unique_together = ("station", "monitor_time") # 同一站点同一时刻只留一条这里最关键的两个设计是db_index=True和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_mode='STRICT_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开头的表头就说明有 BOM，Python 里用encoding="utf-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", required=True, help="CSV 文件路径") parser.add_argument("--city", required=True, help="城市名，例如 北京") parser.add_argument("--station", required=True, help="站点编码") def handle(self, *args, **options): city, _ = City.objects.get_or_create(name=options["city"]) station, _ = Station.objects.get_or_create( code=options["station"], defaults={"name": options["station"], "city": city} ) rows = [] skipped = 0 with open(options["csv"], encoding="utf-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( station=station, monitor_time=t, pm25=parse_float(line.get("pm25")), pm10=parse_float(line.get("pm10")), aqi=int(parse_float(line.get("aqi")) or 0) or None, )) # ignore_conflicts=True 依赖数据库层的唯一约束，命中就跳过 AirQuality.objects.bulk_create(rows, batch_size=2000, ignore_conflicts=True) self.stdout.write(self.style.SUCCESS(f"导入完成：{len(rows)} 条，跳过 {skipped} 条，新增去重后数据"))这段脚本里两个参数必须解释清楚。batch_size=2000控制每次 INSERT 的记录数，太大容易超过 MySQL 的 max_allowed_packet，太小又体现不出批量插入的优势，我长期用 2000 这个值没出过问题。ignore_conflicts=True是配合模型里的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 只读接口不需要 DRF：Django 原生 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"}, status=400) granularity = request.GET.get("granularity", "day") trunc_expr = TruncHour("monitor_time") if granularity == "hour" else TruncDay("monitor_time") rows = ( AirQuality.objects .filter(station__city_id=city_id, monitor_time__range=(start, end)) .annotate(bucket=trunc_expr) .values("bucket") .annotate(avg_pm25=Avg("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, safe=False)这个视图有几个细节值得展开。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(), name="city-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)}&granularity=day`) .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_id=1 %}".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 utf8mb4；settings 的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"], name="idx_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(code="TEST_01", name="测试站", city=self.city) AirQuality.objects.create( station=self.station, monitor_time=timezone.now(), pm25=12.5, pm10=35.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 可视化项目跑通，后面换城市、换污染因子、换数据库都只是参数级改动，希望帮到你。
本文还有配套的精品资源，点击获取
