全国天气数据采集源码：限速重试、幂等写入与分层建模

简介：面向数据分析与爬虫学习者的全国天气数据采集与处理资料包，围绕2011—2024年全国多城市历史天气的获取、整理与分析展开，适合想练习网页数据采集、CSV清洗和可视化建模的初中级读者。包内共300个文件，以261个城市天气CSV为核心数据，配套29个Python脚本、6个XML配置及2个Jupyter Notebook，压缩包约1.23MB；CSV便于按城市检索与统计，脚本展示采集与处理思路，Notebook适合复现实验。已有164人学习。数据覆盖气温、风向风力等天气状况，可用于农业、交通、旅游、健康等场景的趋势分析与案例练习；读者可据此了解多城市数据组织方式、字段清洗流程和批量处理脚本结构，也能参考API优先、网页爬虫为辅的采集策略，完成从数据获取到分析建模的完整练习。

1. 从一条“平了的曲线”说起：全国天气数据采集这套源代码解决什么问题

去年冬天帮朋友排查一个温湿度看板，曲线在下午三点以后变成一条水平直线，页面不报错，日志里也干干净净。追到采集端才发现，上游把字段名改了，脚本照旧去取旧键，拿到一串 None，写库时被默认值兜住，于是整条曲线“平稳”了三天。这件事之后我对全国天气数据采集这事的判断变了：难点从来不是把一次请求发出去，而是在几百个城市、每天几千次请求、多数据源格式各不相同的前提下，保证数据不丢、不重、可追溯。这套全国天气数据采集源代码把爬虫、数据集入库、数据处理切成三段，配上调度、重试、幂等写入和分层建模，适合两类人：一类是想要一份能直接跑的天气数据集做分析或建模练手，另一类是手里有采集任务、被字段漂移和数据重复折磨过的后端或数据开发。下面按“项目长什么样 → 怎么跑起来 → 怎么改”的顺序拆一遍。

2. 工程骨架与技术选型：采集、解析、存储、调度怎么拆成四层

拿到一份采集类源码，我第一件事不是看代码，是看目录。目录结构能直接暴露作者有没有把“取数据”和“处理数据”分开，分开的项目后续好维护，混在一起的项目改一个字段要动五个文件。这份全国天气数据采集项目的分层思路很直接：网络请求只负责拿回原始响应，解析层只负责把响应变成结构化字典，存储层只管写，清洗和汇总单独放在处理层，调度器在最外层按城市和时间片驱动。

2.1 目录结构：从 collectors 到 schedulers 的分工

目录大致长这样，我按实际项目的习惯补全了缺失的部分，你下载后对照着看即可：

weather-collector/ ├── config/ │ ├── settings.yaml # 接口地址、超时、并发数、限速区间 │ └── cities.csv # city_code, city_name, province, lat, lon ├── core/ │ ├── fetcher.py # 请求封装：Session、限速、重试、状态码处理 │ ├── parser.py # 解析分发：json / html / csv 三条分支 │ ├── storage.py # 幂等写入、原始层落库 │ └── logger.py # 统一日志格式，带 trace_id ├── pipelines/ │ ├── clean.py # 字段标准化、单位换算、缺失值处理 │ └── aggregate.py # pandas 日粒度汇总，产出宽表 ├── schedulers/ │ └── job.py # APScheduler 入口，按数据源周期注册任务 ├── scripts/ │ ├── init_db.py # 建表 + 初始化城市维表 │ └── backfill.py # 历史回补，按日期区间拉取 ├── data/ │ ├── raw/ # 原始响应落地（可选，排障用） │ └── weather.db # SQLite，生产可换 MySQL └── requirements.txt

这套结构里有两个地方值得留意。config/cities.csv是城市维表，全国采集的规模全靠它撑着，城市增删只改这个文件，不碰代码，这是把“范围”和“逻辑”解耦的基本操作。data/raw/则是后悔药：解析失败时至少原始报文还在，不用重新跑一遍全国任务。我见过太多项目把响应直接丢进内存解析，解析一报错，现场就没了，只能干瞪眼。

2.2 技术栈选型：为什么不用 Scrapy 而用 requests + APScheduler

选型上作者走的是轻量路线，核心依赖就四块：requests负责 HTTP，lxml和内置json负责解析，pandas负责清洗汇总，APScheduler负责定时，数据库默认 SQLite，量大了换 MySQL。另外tenacity做重试，PyYAML读配置。

这里值得说一下为什么不是 Scrapy。全国地级市三百多个，主流数据源大多是按小时更新的观测接口，一小时撑死几百次请求，一天也就几千到一万次。Scrapy 的价值在高并发、大规模去重和分布式调度，而在这个量级上，它的调度器和去重组件反而是负担，调试成本也高：想单步跟一个请求，得先搞懂 downloader middleware 的执行顺序。用requests.Session复用连接加一个线程池，配合 APScheduler 的 cron 触发器，代码量小一半，出问题一眼能看出在哪。这是我一般会做的取舍——采集规模没到十万级 qps 的量，别急着上重型框架。

依赖清单大致是：

python -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate pip install requests lxml pandas APScheduler tenacity PyYAML # 解析 HTML 若遇到不规范文档，补一个 html5lib 兜底 pip install html5lib python scripts/init_db.py # 建表 + 导入城市维表 python schedulers/job.py # 启动调度

版本号建议装完统一pip freeze > requirements.txt锁死，别在源码里写死小版本，否则过半年重装必翻车。

2.3 配置与启动入口：第一次跑先关掉并发

配置集中在settings.yaml，几个关键参数直接决定第一次能不能跑通：

# config/settings.yaml request: timeout: 8 # 单次请求超时，秒 min_interval: 1.2 # 同一数据源两次请求的最小间隔 max_interval: 2.0 # 抖动上限，实际间隔在 [min, max] 随机 workers: 4 # 线程池大小，首次跑建议 1 retry_times: 3 storage: db_url: "sqlite:///data/weather.db" keep_raw: true # 是否把原始响应落盘 schedule: hourly_source: "0 10 * * * *" # 每小时第 10 分钟拉一次观测数据 daily_source: "0 30 6 * * *" # 每天 6:30 拉一次日预报

第一次跑，我会把workers改成 1，cities.csv只留三五行，确认单城市全链路通了再放量。这不是保守，是省时间：全国任务跑到第 200 个城市报错，你得先判断是接口限流、城市编码错了还是解析挂了，变量太多。先把变量压到一个，再放开，这是排查成本最低的顺序。

3. 采集核心链路：请求封装、三源解析与幂等写入

采集这块是整个项目的重心，也是二次开发时改动最频繁的地方。数据源不是一家的，形态至少三种：返回 JSON 的观测接口、需要从 HTML 表格里抠数据的页面、以及整包下载的 CSV 开放数据集。三种源的失败模式完全不同，所以代码里请求层是共用的，解析层必须分叉。

3.1 请求层：Session 复用、随机抖动与 429 退避

先看core/fetcher.py的写法：

# core/fetcher.py import random import time import logging import requests from tenacity import ( retry, stop_after_attempt, wait_exponential, retry_if_exception_type, ) log = logging.getLogger("fetcher") class Fetcher: def __init__(self, timeout=8, min_interval=1.2, max_interval=2.0, retry_times=3): self.session = requests.Session() # 带联系方式的 UA 是礼貌，也是被限流时对方能找你的唯一途径 self.session.headers.update({ "User-Agent": "weather-collector/1.0 (+ops@example.com)", "Accept": "application/json, text/html;q=0.9,*/*;q=0.8", }) self.timeout = timeout self.min_interval = min_interval self.max_interval = max_interval self.retry_times = retry_times self._last_ts = 0.0 def _pace(self): """两次请求之间插随机间隔，避免固定节奏""" gap = random.uniform(self.min_interval, self.max_interval) wait = self._last_ts + gap - time.time() if wait > 0: time.sleep(wait) self._last_ts = time.time() def get(self, url, **params): @retry( stop=stop_after_attempt(self.retry_times), wait=wait_exponential(multiplier=1, min=2, max=20), retry=retry_if_exception_type( (requests.Timeout, requests.ConnectionError)), reraise=True, ) def _do(): self._pace() resp = self.session.get(url, params=params, timeout=self.timeout) if resp.status_code == 429: # 被限流：拉长退避再抛，让外层重试而不是马上打回去 log.warning("429 on %s, sleep 30s", url) time.sleep(30) raise requests.ConnectionError("rate limited") resp.raise_for_status() return resp return _do()

逐段说下逻辑。Session复用 TCP 连接，几百个城市连续请求时比每次新建连接快一个量级，这在全国任务里很实在。_pace()里用随机区间而不是固定 sleep，原因很简单：固定 1.0 秒间隔的请求序列在服务端看起来就是一台机器，随机抖动能把这种特征打散，属于低成本操作。wait_exponential让重试间隔按 2、4、8 秒递增，避免网络抖动时三次重试挤在同一秒内打过去——这种情况下重试基本无效，还容易把自己送进黑名单。

参数上，timeout我给 8 秒，太短会在弱网下误杀，太长会让单个城市卡住整批任务；min_interval依据对方接口的公开频率限制来，拿不准就往上调，宁可慢。注意 429 单独处理的原因：服务端限流返回 429 时，立刻重试只会加重惩罚，所以先睡 30 秒再抛异常触发外层退避。

3.2 解析层：JSON、HTML 表格、CSV 三条分支怎么分发

core/parser.py的核心是一个按source_type分发的字典：

# core/parser.py import json import csv import io from lxml import etree def parse_json(resp, city): data = resp.json() # 接口偶尔会套一层 {code:0, data:{...}}，这里做一次兼容下钻 if isinstance(data, dict) and "data" in data and "obs_time" not in data: data = data["data"] return { "city_code": city["city_code"], "source": "api_json", "obs_time": data.get("obs_time") or data.get("time"), "temp": data.get("temp", data.get("temperature")), # 字段名兼容 "humidity": data.get("humidity"), "wind_dir": data.get("wind_dir"), "wind_speed": data.get("wind_speed"), } def parse_html(resp, city): # 用 lxml 的 html 解析器，容错比 xml 解析器强 tree = etree.HTML(resp.text) rows = tree.xpath('//table[contains(@class,"obs")]//tr[position()>1]') out = [] for row in rows: cells = [c.text_content().strip() for c in row.xpath("./td")] if len(cells) < 5: continue # 跳过合行/表头残留 out.append({ "city_code": city["city_code"], "source": "html_table", "obs_time": cells[0], "temp": cells[1], "humidity": cells[2], "wind_dir": cells[3], "wind_speed": cells[4], }) return out def parse_csv(resp, city): text = resp.content.decode("utf-8-sig") # 带 BOM 的 CSV 很常见 reader = csv.DictReader(io.StringIO(text)) return [{ "city_code": city["city_code"], "source": "open_csv", "obs_time": r.get("time"), "temp": r.get("temp"), "humidity": r.get("humidity"), "wind_dir": r.get("wind"), "wind_speed": r.get("wind_speed"), } for r in reader] PARSERS = {"api_json": parse_json, "html_table": parse_html, "open_csv": parse_csv} def parse(source_type, resp, city): fn = PARSERS.get(source_type) if fn is None: raise ValueError(f"unknown source_type: {source_type}") return fn(resp, city)

三处细节值得抄走。parse_json里那个"temp", data.get("temperature")的双键兼容，就是文章开头那条“平了的曲线”的直接解药——不指望上游字段名永远不变，而是在解析层保留一层映射。parse_html用etree.HTML而不是 XML 解析器，HTML 里少个闭合标签是常态，严格解析器会直接抛错。parse_csv的utf-8-sig是踩出来的，带 BOM 的文件用utf-8解出来第一列名字会多一个不可见字符，然后DictReader取不到time字段，全程不报错，值全是 None。

注意parse返回值统一成 list 或 dict 两种形态，调用方要用isinstance兜一下再写库，否则写库层要么吃 dict 报错，要么对单条记录迭代出字符串。

3.3 幂等写入：靠唯一键把重复数据挡在库外

采集任务一定会重跑——回补历史、修完 bug 重刷、调度重叠。如果写入不是幂等的，库里的重复行会在汇总时把日均温算歪。做法是在表上建唯一键，让数据库来兜底：

-- scripts/init_db.py 里的建表语句（SQLite 方言） CREATE TABLE IF NOT EXISTS ods_weather_raw ( id INTEGER PRIMARY KEY AUTOINCREMENT, city_code TEXT NOT NULL, source TEXT NOT NULL, obs_time TEXT NOT NULL, -- 统一 ISO8601，UTC payload TEXT NOT NULL, -- 原始 JSON，不做裁剪 fetched_at TEXT NOT NULL, UNIQUE (city_code, source, obs_time) ); -- 写入用 UPSERT，重复就更新，不会产生第二行 INSERT INTO ods_weather_raw (city_code, source, obs_time, payload, fetched_at) VALUES (?, ?, ?, ?, ?) ON CONFLICT (city_code, source, obs_time) DO UPDATE SET payload = excluded.payload, fetched_at = excluded.fetched_at;

MySQL 下这段要换成INSERT ... ON DUPLICATE KEY UPDATE，PostgreSQL 用ON CONFLICT ... DO UPDATE语法基本一致。这是移植时最容易翻车的地方：SQLite 的ON CONFLICT后面必须跟具体冲突列，写成OR REPLACE会把整行删了重建，自增 id 就断了，外键引用这张表的地方全乱。

payload存原始 JSON 而不存拆解后的字段，是为了可重放：清洗规则改了，不用重新采集全国数据，直接拿payload重跑解析就行。这个设计多占一点存储，但省掉的是整轮采集的时间成本，划算。

4. 数据处理链路：原始报文到可分析宽表的清洗与汇总

采集只完成了“拿到”，真正决定这份数据集能不能用起来的是处理段。很多采集项目的失败不在爬虫，在于落库的是一堆字符串：温度带单位、风向是中文、时间是各种格式的本地时间，下游想做个日均温都得先写十行正则。这个项目的处理链路分成原始层（ODS）、明细层（DWD）、汇总层（DWS）三层，每层职责清晰。

4.1 三层建模：ODS 存原文，DWD 存标准值，DWS 存指标

ODS 层的唯一要求是“忠实还原”，所以payload原样保留，不在这里做任何清洗。DWD 层才做类型转换和单位统一。DWS 层面向分析场景，宽表一行一个城市一天，接 BI 或直接喂模型都行。分三层的实际收益是重跑成本：清洗逻辑写错，只重跑 DWD 和 DWS，ODS 不动，几秒钟的事。

建 DWD 表时字段类型要收住，别全用 TEXT：

CREATE TABLE IF NOT EXISTS dwd_weather_obs ( city_code TEXT NOT NULL, obs_time TEXT NOT NULL, -- UTC，ISO8601 temp_c REAL, -- 摄氏度 humidity REAL, -- 0-100 wind_dir_deg INTEGER, -- 0-360，正北为 0 wind_speed_ms REAL, -- 米/秒 quality INTEGER DEFAULT 0, -- 0 正常，1 可疑，2 缺失 UNIQUE (city_code, obs_time) );

quality字段是我坚持要加的。清洗阶段永远有拿不准的值，比如温度 -999 这种明显是缺测标记，直接置 NULL 会丢掉“这里本来有数据但采到的是缺测”的信息，打个标记位后续统计有效样本数时能用上。

4.2 字段标准化：单位换算、风向编码与缺测标记

清洗的核心在pipelines/clean.py，难点集中在三处：

# pipelines/clean.py import re from datetime import datetime, timezone # 中文风向到角度的映射，不同数据源写法不一，统一收口 WIND_DIR_MAP = { "北": 0, "东北": 45, "东": 90, "东南": 135, "南": 180, "西南": 225, "西": 270, "西北": 315, "N": 0, "NE": 45, "E": 90, "SE": 135, "S": 180, "SW": 225, "W": 270, "NW": 315, } MISSING_FLAGS = {-999, -9999, 999999} def to_float(val): """把 '23.5℃'、'23.5 C'、' 23.5 ' 统一成 23.5；失败返回 None""" if val is None: return None m = re.search(r"-?\d+(\.\d+)?", str(val)) if not m: return None num = float(m.group()) if num in MISSING_FLAGS: return None return num def to_celsius(val, unit="C"): """华氏转摄氏；单位信息常藏在字段名或单独的列里""" num = to_float(val) if num is None: return None return round((num - 32) * 5 / 9, 1) if unit.upper() == "F" else num def wind_dir_to_deg(val): if not val: return None key = str(val).strip().replace("风", "") if key in WIND_DIR_MAP: return WIND_DIR_MAP[key] num = to_float(key) return int(num) if num is not None and 0 <= num <= 360 else None def to_utc(ts_str, src_tz_offset=8): """数据源普遍给本地时间且不带时区，按偏移量补成 UTC""" if not ts_str: return None for fmt in ("%Y-%m-%d %H:%M:%S", "%Y/%m/%d %H:%M", "%Y-%m-%dT%H:%M:%S"): try: dt = datetime.strptime(str(ts_str).strip(), fmt) return (dt.replace(tzinfo=timezone.utc) .timestamp() - src_tz_offset * 3600) except ValueError: continue return None def normalize(rec): """单条原始记录 -> DWD 行""" ts = to_utc(rec.get("obs_time")) if ts is None: return None # 没有时间戳的记录无法对齐，直接丢弃并计数 return { "city_code": rec["city_code"], "obs_time": datetime.fromtimestamp(ts, tz=timezone.utc) .isoformat(timespec="seconds"), "temp_c": to_celsius(rec.get("temp")), "humidity": to_float(rec.get("humidity")), "wind_dir_deg": wind_dir_to_deg(rec.get("wind_dir")), "wind_speed_ms": to_float(rec.get("wind_speed")), "quality": 0, }

逐项说明。to_float用正则抠数字而不是float()强转，因为数据源经常把单位拼在值里直接返回字符串，float("23.5℃")会抛异常，捕获异常再返回 None 的写法在几万行数据上会拖慢处理速度。MISSING_FLAGS集合里放的是气象领域常见的缺测标记值，这类值如果不拦住，算出来的日均温会是几百度的负数，而且完全不会报错。

to_celsius的坑在于单位来源不统一：有的接口在字段名里标（temp_f），有的在返回头里声明，有的干脆不标。我一般会在解析层就把单位信息带进记录，这里按需转换，而不是在中途猜。to_utc里那个按固定偏移减小时数的写法，是为了处理“本地时间不带时区”这个普遍情况，全国采集跨时区问题不大，但如果数据源混了多个时区，这个参数必须下沉到城市维表里，每行带自己的偏移量。

4.3 pandas 汇总：日粒度宽表与有效样本数

汇总层的目标是把小时级观测压成一天一行，同时保留样本数，避免“一个城市当天只采到 2 条数据，也和满 24 条的站一样算日均”：

# pipelines/aggregate.py import pandas as pd from sqlalchemy import create_engine def build_daily(engine, day: str): # day 形如 '2024-03-01'，按 UTC 日期过滤 sql = """ SELECT city_code, obs_time, temp_c, humidity, wind_speed_ms FROM dwd_weather_obs WHERE substr(obs_time, 1, 10) = :day """ df = pd.read_sql(sql, engine, params={"day": day}) if df.empty: return pd.DataFrame() df["obs_time"] = pd.to_datetime(df["obs_time"], utc=True) # 先按城市分桶，再做多指标聚合 g = df.groupby("city_code") out = pd.DataFrame({ "temp_avg": g["temp_c"].mean(), "temp_max": g["temp_c"].max(), "temp_min": g["temp_c"].min(), "humidity_avg": g["humidity"].mean(), "wind_max": g["wind_speed_ms"].max(), "sample_cnt": g["temp_c"].count(), # 有效温度样本数，不是行数 }).reset_index() out["stat_date"] = day # 样本数低于阈值的城市打标记，别让空值混进分析 out["is_complete"] = out["sample_cnt"] >= 12 return out.round(2) def save_daily(engine, df): if df.empty: return 0 df.to_sql("dws_weather_daily", engine, if_exists="append", index=False, method="multi", chunksize=500) return len(df)

关键在sample_cnt用的是g["temp_c"].count()而不是len，count()会跳过 NaN，反映的是真正有温度值的观测条数。is_complete阈值我设 12，即当天至少采到一半的小时数据才算完整日，低于这个数的城市在分析时应该剔除或者单独标注，否则一个只有 3 条观测的城市和满采集城市会混在一个均值里。to_sql用method="multi"批量插入，单条 insert 在几千行上会慢得离谱。

5. 避坑与排查：跑全国任务时踩过的五个坑

采集脚本在单城市测试时岁月静好，一放到全国范围就原形毕露。这一章记的是我在类似项目上真实踩过的问题，每条按现象、原因、解决三步走，你跑之前扫一眼能省下不少排查时间。

5.1 采集侧的三类翻车

现象一：日志显示全部成功，库里只有零星几条数据。打开日志，每个城市都是200 OK，但 DWD 表只有几十行。原因在解析层：接口在特定城市返回的是错误提示 JSON，HTTP 状态码依然是 200，解析函数取不到obs_time，normalize里if ts is None: return None把记录静默丢了，而调用方没有统计丢弃数。解决办法是让丢弃变成可见事件——在normalize外层加计数器，每批任务结束打一行parsed=320 dropped=45，同时把被丢弃的原始 payload 落到data/raw/dropped/。线上我一般还会给丢弃率设个阈值，超过 5% 直接告警。

现象二：跑到第 150 个城市左右开始大面积 429。前 100 个城市顺风顺水，之后请求全部被限流，退避 30 秒后仍然失败。原因不在重试逻辑，在并发。workers=4意味着 4 个线程各自维护节奏，但对服务端来说这是 4 倍速率，聚合起来就超了对方按 IP 统计的频率上限。解决办法是把限速从“每线程”提到“每数据源”的全局级别：用threading.Lock保护一个共享时间戳，让所有线程排队共享同一个最小间隔，或者干脆把workers降到 1，用队列串行跑。全国任务跑 4 小时和跑 6 小时的区别，远小于被拉黑封一天的区别。

现象三：HTML 源突然一条都解析不出来，xpath 返回空列表。前一天还正常，第二天rows长度为 0。原因通常是页面结构变了，比如class从obs改成obs-table，或者数据从服务端渲染改成前端 JS 渲染。前者是 xpath 太脆，后者是采集方式不匹配。解决办法分两步：短期把 xpath 从“精确 class”改成“包含匹配”或按表头文字定位（//table[.//th[contains(text(),"温度")]]），提高一点容错；长期要给解析结果加断言，rows为空时记录响应体前 500 字符到日志，一眼就能看出是返回了空壳页面还是结构变了。如果确认是 JS 渲染，那就别硬刚，转去找对应的 JSON 接口，比啃页面划算得多。

5.2 存储与调度侧的两个坑

现象四：回补历史数据后，日均温跟之前对不上，而且偏高。差异不大，0.5 度左右，但分析结果里同一个城市同一天出现两行。原因在唯一键设计：ods_weather_raw的唯一键是(city_code, source, obs_time)，包含source，同一时刻不同数据源的数据会并存，这本身没错；错在 DWS 层汇总时没有按source过滤，把两个源的观测混在一起算了。解决办法是在 DWD 层就选定“主源”，每个城市同一时刻只保留优先级最高的一条写进dwd_weather_obs，跨源的数据放在一张对照表里单独留存。汇总层只读 DWD，不再关心源的问题。

现象五：调度任务偶发重叠执行，同一条数据被写两次。APScheduler 的 cron 任务如果某次执行耗时超过周期，下一次会照常触发，两个任务同时操作同一批城市。原因是没有设max_instances和coalesce。解决办法是注册任务时显式限制：

# schedulers/job.py from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger sched = BlockingScheduler(timezone="UTC") sched.add_job( func=run_hourly_source, trigger=CronTrigger(minute=10), # 每小时第 10 分钟 max_instances=1, # 同一任务不并发 coalesce=True, # 积压的多次触发合并成一次 misfire_grace_time=300, # 迟到 5 分钟内仍执行 id="hourly_source", ) sched.start()

max_instances=1是最关键的那个参数，不加的话长任务和下一次触发会打架；coalesce=True解决的是机器休眠或重启后积压多次触发的问题，不加会连跑好几遍。再叠加上一层的唯一键，两道防线一起，重复数据基本进不来。

6. 二次开发：换源、加告警、提频率的三条改造路径

拿到源码之后真正动手改，方向无非三个，难度和收益差别挺大。换数据源是最常见的，因为公开接口的生命周期没人能保证。改造点在解析层：实现一个新的parse_xxx函数，注册进PARSERS字典，再把cities.csv和settings.yaml里对应的source_type指过去就行，请求层和存储层一行不用动。这就是分层带来的好处——加一个源的成本是写一个函数，而不是改一遍链路。要注意的是新源的字段名一定和旧源不同，normalize里那些.get("temp", data.get("temperature"))的兼容写法该加就加，不要假设字段名稳定。

加告警的改造比想象中重要。我一般会在三个位置埋点：单批次丢弃率超过阈值、连续 N 个城市请求失败、dws_weather_daily的is_complete完整率低于 80%。这三种情况分别对应解析失效、被封禁和数据断流，都是越早发现越省事。实现上不必引入重型组件，一个notify.py里封装 HTTP 推送，失败时降级写本地文件，够用了。

提频率是三条路里最需要克制的。从小时级提到分钟级，请求量直接翻十几倍，先算清楚对方的频率限制再动代码，把min_interval按“每分钟允许次数”倒推，并且给全局限速加锁，别指望靠调大workers蒙混过去。真要高频，优先看对方有没有批量接口——一次请求拿多个城市的数据，比提高请求频次划算得多。

那个换数据库的坑我吃过一次：SQLite 的ON CONFLICT (col) DO UPDATE迁到 MySQL 上直接语法错误，得重写成ON DUPLICATE KEY UPDATE，而且 MySQL 连接串不加charset=utf8mb4，城市名里的特殊字符会变成问号。从那以后我每次迁移存储层，都强制先拿三个城市跑一遍全链路，把 ODS、DWD、DWS 三张表的行数和抽样值逐个对一遍再放量——这个习惯救过我很多次。希望这份拆解能帮你少走两遍弯路。

本文还有配套的精品资源，点击获取