48小时紧急上线：疫情数据清洗与轻量API实战

发布时间：2026/10/12 13:32:07
48小时紧急上线：疫情数据清洗与轻量API实战
1. 项目缘起与整体设计思路1.1 这个项目到底要解决什么问题先把这个项目的背景说清楚。当时的情况是有一批公开的疫情历史数据需要被快速整理、清洗、结构化并且要在一个极短的时间窗口内上线供内部团队和合作方查询使用。标题里写的“紧急上线”四个字不是修辞是真的紧急——从接到需求到数据可查留给我们的时间只有不到48小时。这类项目的核心难点从来都不是“写代码”而是在信息不完整、时间极度压缩的情况下做出足够好的工程决策。你要在数据准确性、上线速度、后续可维护性之间做取舍而且每一个取舍都要能说服自己和团队。我先把当时面对的核心约束列一下这样你就能理解后面所有技术选型背后的逻辑时间约束48小时内必须有一个可访问的查询接口数据要能按时间、地区维度检索。数据约束原始数据来源分散格式不统一存在缺失值、重复记录、字段命名混乱等问题。人力约束能投入的只有两三个人不可能做完整的ETL平台。合规约束所有数据必须是公开渠道可获取的不涉及任何非公开信息且展示时要做脱敏和聚合处理。这里要特别强调一点任何涉及公共卫生数据的项目第一原则永远是合规。数据来源必须可追溯、可说明展示粒度要合理不能暴露任何个体层面的信息。这条底线比上线速度重要得多。1.2 为什么选择“先跑通再优化”的方案很多人做数据项目容易犯一个错误一上来就想搭一个完美的数据管道结果时间全花在架构设计上真正能用的东西一个都没有。这个项目最大的教训就是——紧急场景下能用比优雅重要一百倍。我们当时的决策逻辑是这样的方案优点缺点是否采用完整ETL平台可维护性强至少两周工期否脚本静态文件半天可跑通更新麻烦过渡采用脚本轻量数据库API平衡速度与可用性需要额外半天是最终选的是第三条路用Python脚本做数据清洗写入一个轻量级关系型数据库再套一层最简单的HTTP接口。整个链路没有任何“高级”组件但每一个环节都是我能在半小时内调试通的。这个选择背后的核心判断是紧急项目的技术选型应该以“我闭着眼睛都能部署”为标准而不是以“业界最佳实践”为标准。你选一个自己不熟的框架光是查文档就能把48小时耗光。1.3 整体架构长什么样最终上线的架构简单到有点寒酸但确实管用原始数据文件CSV/JSON混杂 ↓ 清洗脚本Python pandas ↓ 标准化中间表统一字段、统一时间格式 ↓ SQLite数据库单文件方便迁移 ↓ Flask轻量接口三个查询端点 ↓ 前端简单页面 / 内部调用为什么用SQLite而不是MySQL或者PostgreSQL因为SQLite是单文件数据库不需要单独部署服务拷贝一个文件就能迁移对于这种临时性、查询量不大的场景它是最优解。实测下来几十万条记录的查询响应在毫秒级完全够用。为什么用Flask而不是FastAPI因为当时我对Flask的部署流程更熟而且Flask的依赖更少在一台干净的机器上pip install flask就能跑。FastAPI虽然性能更好、自带文档但在那个时间点少一个不确定性就多一分按时上线的把握。2. 数据清洗的核心细节与实操要点2.1 原始数据的典型问题长什么样我拿到的原始数据大概有这么几类问题我一个个说你对号入座第一类是字段命名混乱。同样是“累计确诊人数”有的文件叫confirmed有的叫total_confirmed有的叫累计确诊还有的叫确诊总数。这种问题在多人协作、多来源汇总的场景下极其常见。第二类是时间格式不统一。有的用2020-01-01有的用2020/1/1有的用20200101甚至还有用Excel序列号的。时间字段是这类项目的核心维度格式不统一后面所有查询都会出问题。第三类是地区名称不规范。“武汉市”和“武汉”和“湖北武汉”指的是同一个地方但在数据里就是三条不同的记录。如果不做标准化聚合查询的结果会完全错误。第四类是缺失值和异常值。有些日期的数据缺失有些数值明显不合理比如累计确诊人数比前一天少这在累计口径下是不可能的。2.2 清洗脚本的关键步骤拆解我把清洗流程拆成了五个步骤每一步都有明确的输入输出这样出问题的时候能快速定位是哪一步的锅。第一步统一字段映射。我建了一个映射字典把所有可能的字段名都映射到标准字段名上。FIELD_MAPPING { confirmed: cum_confirmed, total_confirmed: cum_confirmed, 累计确诊: cum_confirmed, 确诊总数: cum_confirmed, dead: cum_dead, 累计死亡: cum_dead, # ... 其他字段 } def standardize_columns(df): df df.rename(columnsFIELD_MAPPING) required [date, region, cum_confirmed, cum_dead] for col in required: if col not in df.columns: df[col] None return df[required]这个映射字典是逐步积累的每遇到一个新的字段名就加进去。别想着一次写全不可能的边跑边补才是真实情况。第二步时间格式归一化。这一步我用了一个比较暴力的方法——把所有时间都先转成字符串然后用正则匹配出年月日再统一格式化成YYYY-MM-DD。import re from datetime import datetime def normalize_date(raw): if raw is None: return None s str(raw).strip() # 处理 2020-01-01 / 2020/1/1 / 20200101 m re.match(r(\d{4})[-/]?(\d{1,2})[-/]?(\d{1,2}), s) if m: y, mo, d m.groups() return f{y}-{int(mo):02d}-{int(d):02d} return None注意这里一定要处理int(mo)的情况因为2020-1-1这种格式如果不补零后面做字符串排序会出错。我一开始就踩了这个坑导致按日期排序的结果是乱的。第三步地区名称标准化。这一步最麻烦因为地区名称的变体太多了。我的做法是建一个标准地区列表然后用模糊匹配把原始名称映射过去。from difflib import get_close_matches STANDARD_REGIONS [武汉市, 黄冈市, 孝感市, ...] # 完整列表 def normalize_region(raw): if raw is None: return None s str(raw).strip().replace(省, ).replace(市, ) matches get_close_matches(s, [r.replace(市,) for r in STANDARD_REGIONS], n1, cutoff0.8) if matches: idx [r.replace(市,) for r in STANDARD_REGIONS].index(matches[0]) return STANDARD_REGIONS[idx] return None模糊匹配的cutoff参数我设的是0.8这个值是试出来的。设太高会漏掉一些变体设太低会误匹配。0.8在我的数据集上表现最好但你的数据可能不一样需要自己调。第四步缺失值处理。累计口径的数据有个特点——如果某天缺失可以用前一天的值填充因为累计值不会减少。但如果连续多天缺失就不能简单填充了需要标记出来。def fill_missing_cumulative(df): df df.sort_values([region, date]) df[cum_confirmed] df.groupby(region)[cum_confirmed].ffill() df[cum_dead] df.groupby(region)[cum_dead].ffill() return df第五步异常值检测。累计值不能减少这是硬约束。我写了一个检查把违反这个约束的记录标记出来人工复核。def check_monotonic(df): df df.sort_values([region, date]) df[prev_confirmed] df.groupby(region)[cum_confirmed].shift(1) anomalies df[df[cum_confirmed] df[prev_confirmed]] return anomalies实测下来异常记录大概占总量的千分之几大部分是数据录入错误少数是口径调整。这些记录我没有直接删除而是标记后单独存放因为删除可能会掩盖问题。2.3 清洗过程中踩过的坑坑一编码问题。原始CSV文件有的是UTF-8有的是GBK直接读会报错。我的处理方式是先尝试UTF-8失败就试GBK再失败就试GB18030。def read_csv_safe(path): for enc in [utf-8, gbk, gb18030]: try: return pd.read_csv(path, encodingenc) except UnicodeDecodeError: continue raise ValueError(f无法解码文件: {path})坑二日期排序陷阱。前面提到过如果日期没有补零字符串排序会出错。2020-1-10排在2020-1-2前面因为字符1小于2。这个bug很隐蔽因为数据量小的时候不容易发现。坑三地区匹配的边界情况。有些记录的地区名称是空的或者写的是“待确认”“未知”之类的。这些记录我统一归到“其他”类别不参与地区维度的聚合但在总量统计里保留。实操心得清洗脚本一定要写日志。每处理完一个文件记录处理了多少条、跳过了多少条、异常多少条。这样出问题的时候你能快速定位是哪个文件、哪个环节出的错。我当时就是靠日志发现某个文件的时间字段全是空的及时补了数据源。3. 接口设计与上线部署的完整流程3.1 接口设计少即是多时间紧迫的情况下接口设计的原则是只做必须的不做想要的。我最终只设计了三个端点端点功能参数返回/api/daily按日期查询全国汇总date当日累计确诊、累计死亡/api/region按地区查询时间序列region,start,end该地区时间序列/api/latest查询最新一天数据无最新日期的全地区数据为什么只有三个因为这三个覆盖了90%以上的查询需求。其他需求可以通过组合这三个接口实现或者直接查数据库。在紧急场景下每多一个接口就多一份测试和维护成本。接口返回格式统一用JSON字段名和数据库字段保持一致不做额外转换。这样前端拿到数据就能直接用少一层映射就少一个出错的地方。from flask import Flask, request, jsonify import sqlite3 app Flask(__name__) DB_PATH data.db def query_db(sql, args()): conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row cur conn.execute(sql, args) rows cur.fetchall() conn.close() return [dict(r) for r in rows] app.route(/api/daily) def daily(): date request.args.get(date) if not date: return jsonify({error: date is required}), 400 rows query_db( SELECT date, SUM(cum_confirmed) as cum_confirmed, SUM(cum_dead) as cum_dead FROM records WHERE date ?, (date,) ) return jsonify(rows) app.route(/api/region) def region(): region request.args.get(region) start request.args.get(start) end request.args.get(end) if not region: return jsonify({error: region is required}), 400 sql SELECT date, cum_confirmed, cum_dead FROM records WHERE region ? args [region] if start: sql AND date ? args.append(start) if end: sql AND date ? args.append(end) sql ORDER BY date return jsonify(query_db(sql, args)) app.route(/api/latest) def latest(): row query_db(SELECT MAX(date) as d FROM records)[0] if not row[d]: return jsonify([]) return jsonify(query_db( SELECT region, cum_confirmed, cum_dead FROM records WHERE date ?, (row[d],) ))3.2 数据库设计与索引优化表结构非常简单就一张表CREATE TABLE records ( id INTEGER PRIMARY KEY AUTOINCREMENT, date TEXT NOT NULL, region TEXT NOT NULL, cum_confirmed INTEGER, cum_dead INTEGER, UNIQUE(date, region) ); CREATE INDEX idx_date ON records(date); CREATE INDEX idx_region ON records(region); CREATE INDEX idx_date_region ON records(date, region);UNIQUE(date, region)这个约束很重要它能防止重复插入。我在清洗阶段就做了去重但数据库层面再加一道保险防止后续更新时出错。索引方面idx_date和idx_region分别支持按日期和按地区的查询idx_date_region是联合索引支持同时按两个维度过滤。实测下来加了索引之后查询响应从几百毫秒降到了几毫秒。注意SQLite的索引对写入性能有影响但我们的场景是读多写少所以索引带来的读性能提升远大于写入损耗。如果你的场景是频繁写入索引策略需要重新评估。3.3 部署怎么在半小时内让服务跑起来部署这块我选的是最朴素的方式——一台普通服务器直接跑Python脚本用nohup挂后台。# 安装依赖 pip install flask pandas # 启动服务 nohup python app.py server.log 21 # 检查是否启动成功 curl http://localhost:5000/api/latest没有用Docker没有用Nginx没有用Gunicorn。为什么因为每多一层就多一个可能出问题的地方。Flask自带的开发服务器虽然性能一般但对于内部使用、查询量不大的场景完全够用。等稳定运行之后再考虑加Gunicorn和Nginx也不迟。当然这里有个前提——服务只在内部网络访问不直接暴露到公网。如果需要对外那必须加反向代理和基本的访问控制这是安全底线。3.4 数据更新机制数据不是一次性的后续还会有新数据进来。我设计了一个简单的更新流程新数据文件放到指定目录运行更新脚本清洗后增量写入数据库脚本自动检查数据完整性输出报告def incremental_update(new_file): df read_csv_safe(new_file) df standardize_columns(df) df[date] df[date].apply(normalize_date) df[region] df[region].apply(normalize_region) df df.dropna(subset[date, region]) conn sqlite3.connect(DB_PATH) for _, row in df.iterrows(): conn.execute( INSERT OR REPLACE INTO records (date, region, cum_confirmed, cum_dead) VALUES (?, ?, ?, ?), (row[date], row[region], row[cum_confirmed], row[cum_dead]) ) conn.commit() conn.close()INSERT OR REPLACE是SQLite的特性如果(date, region)已经存在就替换不存在就插入。这样更新脚本可以重复运行不会产生重复数据。4. 常见问题排查与实战避坑指南4.1 数据层面的高频问题问题一查询结果为空。最常见的原因是日期格式不匹配。比如你传的是2020-1-1但数据库里存的是2020-01-01查询就会返回空。解决办法是在接口层做一次日期格式化确保传入的日期和数据库格式一致。问题二聚合结果偏大。通常是因为地区名称没有完全标准化同一个地区有多条记录。排查方法是先查一下SELECT DISTINCT region FROM records看看有没有明显的重复变体。问题三累计值出现下降。前面提到过这通常是数据录入错误。我的处理方式是在接口层不做修正但在返回结果里加一个标记字段让调用方知道这条数据有问题。4.2 服务层面的高频问题问题一服务突然无响应。最常见的原因是数据库文件被锁。SQLite在写入时会锁库如果同时有大量读请求可能会超时。解决办法是设置timeout参数让读请求等待锁释放。conn sqlite3.connect(DB_PATH, timeout10)问题二内存占用持续增长。如果每次查询都新建数据库连接而不关闭连接数会持续增长。我的做法是用上下文管理器确保连接一定被关闭。from contextlib import contextmanager contextmanager def get_db(): conn sqlite3.connect(DB_PATH, timeout10) try: yield conn finally: conn.close()问题三日志文件过大。nohup输出的日志如果不做轮转几天就能涨到几个G。我的做法是在启动脚本里加一个简单的日志切割每天生成一个新文件。4.3 独家避坑技巧汇总问题类型典型表现排查思路解决方案日期格式查询为空检查数据库中的日期格式接口层统一格式化地区重复聚合偏大DISTINCT region检查完善标准化映射数据异常累计值下降按地区排序检查标记异常人工复核服务无响应请求超时检查数据库锁设置timeout参数内存泄漏内存持续增长检查连接是否关闭使用上下文管理器日志过大磁盘占满检查日志文件大小日志轮转实操心得紧急项目上线后一定要留一个人盯着。前24小时是最容易出问题的时候有人盯着就能快速响应。我当时是设了告警一旦接口响应超过1秒就发通知这样能在用户感知之前发现问题。4.4 后续可扩展的方向这个项目上线后稳定运行了一段时间后续如果要继续优化我会考虑这几个方向第一加缓存。对于不常变的数据可以在内存里缓存查询结果减少数据库压力。最简单的做法是用Python的functools.lru_cache装饰器。第二加数据校验。在写入数据库之前增加更严格的数据校验规则比如累计值必须单调递增、日期必须在合理范围内等。第三加监控。记录每个接口的调用次数、响应时间、错误率这样能及时发现异常。第四考虑迁移到更正式的数据库。如果查询量增长到SQLite扛不住的程度可以迁移到PostgreSQL。但因为表结构简单迁移成本很低。我个人在实际操作中的体会是紧急项目最大的敌人不是技术难度而是决策瘫痪。你总想找一个完美的方案但时间不等人。先做一个能用的版本上线之后再迭代这个思路在90%的紧急场景下都是对的。踩过几次坑之后我现在遇到类似需求第一反应就是“最小可用版本是什么”然后围绕这个目标做减法而不是做加法。