简介：面向数据分析学习者、爬虫初学者与需要长时序气象数据的科研人员，这套资料提供全国多城市2011—2024年天气数据及配套采集源代码，可用于气温、风向风力等气象指标的趋势分析、可视化与建模练习。包内共300个文件，以261个CSV分城市数据为主体，辅以29个Python脚本、2个Jupyter Notebook及若干XML配置，压缩包约1.23MB，目录按城市与功能拆分，便于按需检索和二次处理。已有164人学习，说明其在数据获取与清洗练习中具有一定参考价值。读者可借助脚本理解历史天气页面的请求、解析与落库思路，用Notebook复现数据整理和分析流程，并将结果用于农业、交通、旅游、健康等场景的探索性研究；使用时需注意遵守来源网站条款与相关法律法规。
1. 全国天气数据采集源码包：先看清它能采什么、不能采什么
这份全国天气数据采集源码包，不是把页面一抓就完事，它把爬虫请求、原始响应留存、数据集清洗和增量合并串成了一条可复跑的链路。做区域天气趋势、出行风险或农业小气候分析的人，最怕数据源分散、字段不统一、跑一周断三天；这份源代码把城市列表、请求节流、断点续采和字段标准化拆成独立模块，适合拿来改数据源、换城市范围，或者当作数据处理练手项目。
它不承诺绕过任何访问限制，也不替你保证数据源长期不变，能解决的是采集流程工程化的问题。新手可以按脚本顺序跑通一条城市链路，熟手更该关注状态表、失败队列和回放设计，后面几章按这个顺序拆。
2. 采集链路怎么搭：爬虫请求、解析与原始响应落盘
天气采集链路看着简单，真正跑全国范围时，问题通常不在“抓不到”，而在“抓回来没法用”。一条能复跑的链路至少要拆成六段：城市队列、请求层、解析层、原始响应落盘、清洗层、状态记录。城市队列决定采集范围，请求层决定稳定性，解析层决定字段能不能对上，原始响应落盘决定后面能不能回放，清洗层决定数据集能不能直接分析，状态记录决定中断后能不能续采。常见误用是写一个 for 循环从头抓到尾，中间任何一个城市失败就整批重来，最后既不知道哪些城市漏了，也不知道字段从哪天开始变。
2.1 先定数据粒度：城市、区县、站点还是网格
全国天气数据采集最先要定的不是代码，而是粒度。城市级数据适合做趋势和对比，区县数据适合做覆盖和下沉分析，站点数据更接近观测实况但对站点编码维护要求高，网格数据适合插值和空间分析但体积大、清洗成本高。选型时先看数据源本身支持到哪一级，再看业务问题需不需要那么细。比如只做城市级日趋势，硬上网格数据只会让存储和去重变复杂；反过来要做区县覆盖分析，只采城市级就会在后续关联时缺一层维度。
我一般会先把城市列表做成一份独立的 JSON 或 CSV，字段至少包含city_id、city_name、parent_id、level、source_code。city_id用内部稳定编码，source_code存数据源那边的编码，两者不要混用。下面这段代码把城市列表读进来，生成采集队列，并按优先级排序。示例城市用CITY_A这类代号，实际替换成自己的编码表即可。
import json from pathlib import Path # 城市列表：内部编码与数据源代码分开维护 city_file = Path("config/cities.json") cities = json.loads(city_file.read_text(encoding="utf-8")) # 只保留启用状态的城市，并按 level 排序，先跑城市级再跑区县 queue = [ { "city_id": item["city_id"], "city_name": item["city_name"], "source_code": item["source_code"], "level": item.get("level", "city"), "priority": item.get("priority", 100), } for item in cities if item.get("enabled", True) ] queue.sort(key=lambda x: (x["priority"], x["city_id"])) with open("runtime/city_queue.json", "w", encoding="utf-8") as f: json.dump(queue, f, ensure_ascii=False, indent=2) print(f"queue size: {len(queue)}")逻辑说明：这段代码不直接发请求，只负责把城市配置转成采集队列。把enabled为 false 的城市过滤掉，避免临时停采的城市反复进队列；priority越小越先跑，方便先跑重点城市验证链路。参数说明：city_id是内部主键，后面入库和去重都用它；source_code只在请求和解析时使用，不要拿它当内部主键，否则换数据源时历史数据会对不上；level用来区分城市级和区县级，后面导出数据集时可以按粒度拆表。
2.2 请求层：会话复用、超时、重试与退避
请求层最容易写成“能跑就行”，但天气采集通常要跑批量城市，连接不复用、超时不设、失败不重试，跑几十个城市就开始出现卡死。常见做法是用requests.Session复用连接，显式设置连接超时和读取超时，再配一个有限次数的重试。重试不要无限加，最多两到三次，并且退避时间要递增。这里不写任何绕过限制的逻辑，频率控制和访问边界放在第 3 章处理，请求层只负责把一次请求发稳。
import time import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry session = requests.Session() retry = Retry( total=3, # 总重试次数，别设太大 backoff_factor=0.8, # 退避基数，失败后等待 0.8s、1.6s、3.2s status_forcelist=(429, 500, 502, 503, 504), allowed_methods=("GET",), ) adapter = HTTPAdapter(max_retries=retry, pool_connections=20, pool_maxsize=20) session.mount("https://", adapter) session.headers.update({ "User-Agent": "WeatherDataCollector/1.0 (contact: example@example.com)", "Accept": "application/json,text/html;q=0.9", }) def fetch(url, params=None): # 连接超时 3 秒，读取超时 10 秒，避免一个城市卡住整批 resp = session.get(url, params=params, timeout=(3.05, 10)) resp.raise_for_status() return resp if __name__ == "__main__": url = "https://weather.example.com/api/v1/current" r = fetch(url, params={"city": "CITY_A"}) print(r.status_code, r.headers.get("content-type"))逻辑说明：HTTPAdapter把重试策略挂到会话上，遇到 429、5xx 这类临时失败会按退避重试；timeout=(3.05, 10)分别控制连接和读取，天气接口偶尔慢响应，读取超时设太短会制造大量假失败。参数说明：pool_connections和pool_maxsize控制连接池，批量采集时不要低于并发数；backoff_factor配合Retry使用，失败越多等待越长；User-Agent里留一个可联系标识，方便数据源侧识别流量来源。常见误用是把timeout只写一个数字，那只设读取超时，连接阶段仍可能挂住。
2.3 解析层：HTML 与 JSON 两条路径
天气数据源的响应格式不统一，有的给 JSON，有的给 HTML 表格，还有的两种都混。解析层不要假设永远是 JSON，先看content-type，再决定走哪条路径。JSON 路径按字段映射表取数，HTML 路径用选择器取表格行。字段缺失时不要直接让整批任务崩掉，记录缺失字段并继续，后面用失败队列回捞。下面这段代码演示两条路径的分流，示例选择器只做结构示意，实际按数据源页面替换。
from bs4 import BeautifulSoup def parse_response(resp): ctype = resp.headers.get("content-type", "").lower() if "application/json" in ctype: data = resp.json() # 字段映射集中在一个函数里，后面字段改名只改这里 return { "city_id": data.get("city", {}).get("id"), "observe_time": data.get("obsTime"), "temp_c": data.get("temp"), "precip_mm": data.get("precip", 0), "wind_ms": data.get("windSpeed"), "raw_status": "ok", } soup = BeautifulSoup(resp.text, "html.parser") rows = soup.select("table.weather-table tbody tr") records = [] for row in rows: cells = [c.get_text(strip=True) for c in row.select("td")] if len(cells) < 4: continue records.append({ "city_id": None, # HTML 页面里没有稳定编码时留空，后面用映射表补 "observe_time": cells[0], "temp_c": cells[1], "precip_mm": cells[2], "wind_ms": cells[3], "raw_status": "partial", }) return records逻辑说明：解析函数先判断响应类型，JSON 走字典取值，HTML 走表格行解析。字段映射集中写，后面数据源改字段名时只改这一段，不用满项目搜temp。参数说明：data.get("precip", 0)给缺失降水一个默认值，但默认值要在清洗层标记来源，不能把“缺测”当“0 毫米”；raw_status用来标记解析完整度，partial的记录后面要进人工核对或二次回捞。常见误用是解析时直接用data["temp"]，字段一改名就整批KeyError，日志里只剩堆栈，看不出是哪个城市哪次请求出的问题。
2.4 落盘层：原始层、清洗层、聚合层
落盘层决定后面能不能回放。我一般分三层：原始层存响应原文，清洗层存标准化后的记录，聚合层存按小时或按天汇总的数据集。原始层不要只存解析后的字段，一定把响应体原样留下来，压缩成 gzip 或按日期分文件。清洗层可以用 JSON Lines 或 Parquet，聚合层再按业务问题导出 CSV。下面这段代码把原始响应按“日期/数据源/城市”落盘，并写一份采集元数据。
import gzip import hashlib import json from datetime import datetime, timezone from pathlib import Path RAW_DIR = Path("data/raw") RAW_DIR.mkdir(parents=True, exist_ok=True) def save_raw(city_id, url, resp): now = datetime.now(timezone.utc) day = now.strftime("%Y-%m-%d") body = resp.content digest = hashlib.sha256(body).hexdigest()[:16] out_dir = RAW_DIR / day / "weather" out_dir.mkdir(parents=True, exist_ok=True) raw_path = out_dir / f"{city_id}_{now.strftime('%H%M%S')}_{digest}.json.gz" with gzip.open(raw_path, "wb") as f: f.write(body) meta = { "city_id": city_id, "url": url, "status_code": resp.status_code, "content_type": resp.headers.get("content-type"), "fetched_at_utc": now.isoformat(), "sha256_16": digest, "raw_path": str(raw_path), } with open(out_dir / "meta.jsonl", "a", encoding="utf-8") as f: f.write(json.dumps(meta, ensure_ascii=False) + "\n") return raw_path逻辑说明：原始响应用 gzip 压缩，文件名里带城市、时间和内容摘要前 16 位，后面排查“同一城市同一时刻为什么两条数据不一样”时可以直接对比。meta.jsonl追加写，记录请求 URL、状态码、内容类型和落盘路径，后面回放和审计都靠它。参数说明：sha256_16只取摘要前 16 位，够做文件级区分，不要拿它当安全校验；fetched_at_utc统一用 UTC，业务时区在清洗层再转；out_dir按天分目录，避免单目录文件过多。注意原始层不要做字段裁剪，裁剪过的响应回放不出原始字段，后面字段对不上时就没有后悔药。
3. 反爬与稳定性：爬虫节流、断点续采和失败队列
天气采集跑全国范围，稳定性比“抓得快”重要。数据源有访问边界，采集任务也有网络波动，真正要解决的是三件事：请求别超边界、中断能续、失败能回捞。这一章不讲绕过限制，只讲在合规访问前提下把爬虫任务做稳。血泪经验是，很多任务不是死在解析，而是死在“跑到一半断了，不知道从哪继续”，最后只能全量重跑。
3.1 为什么不能一把梭：并发、频率与封禁边界
并发不是越高越好。天气接口通常有频率限制，城市级采集哪怕只跑几百个城市，并发开到几十也可能触发限流。我的做法是先按数据源给的配额设一个全局速率，再用令牌桶或简单的间隔控制把请求摊开。不要用固定sleep(1)硬扛，城市多的时候总时长不可控；也不要用无上限线程池，失败重试叠加上去会把瞬时压力放大。下面是一个简单的令牌桶，按每秒允许的请求数放行。
import time import threading class TokenBucket: def __init__(self, rate_per_sec, capacity=None): self.rate = rate_per_sec self.capacity = capacity or rate_per_sec self.tokens = self.capacity self.updated = time.monotonic() self.lock = threading.Lock() def acquire(self, tokens=1): with self.lock: now = time.monotonic() elapsed = now - self.updated self.tokens = min(self.capacity, self.tokens + elapsed * self.rate) self.updated = now if self.tokens >= tokens: self.tokens -= tokens return wait = (tokens - self.tokens) / self.rate time.sleep(wait) return self.acquire(tokens) bucket = TokenBucket(rate_per_sec=2, capacity=2) def limited_get(fetch_func, url, params=None): bucket.acquire() return fetch_func(url, params=params)逻辑说明：令牌桶按固定速率补充令牌，请求前先取令牌，取不到就等。这样并发线程再多，实际发出去的请求速率也被压在设定值以内。参数说明：rate_per_sec=2表示每秒最多两个请求，实际按数据源允许范围调整；capacity是桶容量，设太大允许短时突发，设成和速率一致更平滑。常见误用是只在主线程sleep，子线程照样并发发请求，限速等于没做。
3.2 断点续采：用 SQLite 记录采集状态
断点续采的核心是状态表。每个城市每个采集批次一条记录，状态至少包含pending、running、success、failed。任务启动时先把当天批次写进表，跑到哪个城市就更新哪个城市。中断后重新启动，只取pending和failed的记录，不要从头再来。下面这段 SQL 建一张采集状态表，主键用“采集日期 + 城市 + 数据源”，避免同一天重复插。
CREATE TABLE IF NOT EXISTS collect_state ( collect_date TEXT NOT NULL, city_id TEXT NOT NULL, source_name TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'pending', attempt INTEGER NOT NULL DEFAULT 0, last_error TEXT, updated_at TEXT NOT NULL, PRIMARY KEY (collect_date, city_id, source_name) ); CREATE INDEX IF NOT EXISTS idx_collect_state_status ON collect_state (collect_date, status, attempt);逻辑说明：collect_date用业务采集日期，不用时间戳，方便按天重跑；status用四个固定值，后面失败队列只查failed，续采用pending；attempt记录尝试次数，超过上限不再自动重试。参数说明：source_name用来区分不同数据源，同一城市可能采两个源，主键必须带上；last_error存最后一次错误摘要，不要存完整堆栈，完整堆栈放日志文件。注意状态更新要在请求前后都写，不能只在成功后写，否则进程被杀时状态还是pending，下次会重复采。
3.3 失败队列与二次回捞
失败不可怕，可怕的是失败后不知道哪些失败。失败队列不用单独建表，直接复用状态表，把status='failed' AND attempt < 3的记录取出来二次回捞。回捞时退避时间要比第一次长，并且限制单次回捞数量，避免一次把失败量全打出去。下面这段 Python 代码从状态表取失败记录，更新为running后重新请求，成功写success，失败累加attempt。
import sqlite3 from datetime import datetime, timezone conn = sqlite3.connect("runtime/collect.db") conn.row_factory = sqlite3.Row def pick_failed(limit=50): rows = conn.execute( """ SELECT * FROM collect_state WHERE status = 'failed' AND attempt < 3 ORDER BY updated_at ASC LIMIT ? """, (limit,), ).fetchall() return rows def mark_running(row): conn.execute( "UPDATE collect_state SET status='running', updated_at=? WHERE collect_date=? AND city_id=? AND source_name=?", (datetime.now(timezone.utc).isoformat(), row["collect_date"], row["city_id"], row["source_name"]), ) conn.commit() def mark_result(row, ok, error=None): if ok: conn.execute( "UPDATE collect_state SET status='success', last_error=NULL, updated_at=? WHERE collect_date=? AND city_id=? AND source_name=?", (datetime.now(timezone.utc).isoformat(), row["collect_date"], row["city_id"], row["source_name"]), ) else: conn.execute( "UPDATE collect_state SET status='failed', attempt=attempt+1, last_error=?, updated_at=? WHERE collect_date=? AND city_id=? AND source_name=?", (str(error)[:300], datetime.now(timezone.utc).isoformat(), row["collect_date"], row["city_id"], row["source_name"]), ) conn.commit()逻辑说明：pick_failed只取尝试次数小于 3 的失败记录，按更新时间升序，先回捞最早失败的；mark_running在请求前把状态改成running，防止多个进程同时抢同一条；mark_result根据结果写成功或失败，失败时把错误截断到 300 字符。参数说明：limit=50控制单次回捞量，失败多的时候分批跑；attempt < 3是上限，超过后留给人工排查，不要无限重试。常见误用是失败后立刻原地重试，退避没有，反而把限流触发得更彻底。
3.4 日志与可观测性：至少留下四类字段
采集任务要能排查，日志至少留四类字段：请求标识、城市标识、阶段、错误摘要。请求标识可以用collect_date + city_id + source_name拼，阶段写fetch、parse、save、clean，错误摘要写异常类型和关键参数。下面是一张日志字段表，实际落 JSON Lines 或数据库都行。
| 字段 | 示例 | 用途 |
|---|---|---|
trace_id | 2025-06-01_CITY_A_weather | 串起一次采集的请求、解析、落盘 |
stage | fetch/parse/save | 定位失败发生在哪一层 |
city_id | CITY_A | 快速筛某个城市的所有日志 |
error_type | TimeoutError/KeyError | 聚合错误类型，判断是网络还是字段问题 |
elapsed_ms | 842 | 看请求耗时是否异常拉长 |
逻辑说明：日志不要只写“请求失败”，要能回答哪个城市、哪个阶段、什么错误、耗时多少。trace_id用采集日期加城市加数据源，同一批次的所有日志都能串起来。参数说明：elapsed_ms超过读取超时的 80% 就值得关注，可能是数据源变慢；error_type按异常类聚合，KeyError多说明字段变了，TimeoutError多说明网络或限流。注意日志里不要打印完整响应体，响应体进原始层，日志只留摘要和路径。
4. 数据处理：把原始天气响应洗成可分析的数据集
采集回来的原始响应只是原料，真正能拿去做分析的是标准化后的数据集。数据处理这一段最容易“看起来跑通、一分析就错”，常见问题是单位不统一、时间字段有时区、去重主键选错。下面按字段、时间、去重、导出四步拆，每一步都给可抄的代码片段。示例字段用temp_c、precip_mm、wind_ms，实际按数据源字段映射。
4.1 字段标准化：温度、降水、风速的单位与缺失值
字段标准化先做三件事：单位统一、类型转换、缺失值标记。温度统一到摄氏度，降水统一到毫米，风速统一到米每秒。数据源如果给华氏度或节，要在清洗层转掉，不要留到分析层再转。缺失值不要直接用 0 填，用一个单独的is_missing标记，否则后面算平均降水会把缺测当无降水。下面这段 pandas 代码做单位转换和缺失标记。
import pandas as pd import numpy as np def standardize_weather(df: pd.DataFrame) -> pd.DataFrame: df = df.copy() # 温度：如果源字段是华氏度，先转摄氏度；示例按摄氏度处理 df["temp_c"] = pd.to_numeric(df["temp_c"], errors="coerce") # 降水：统一毫米，缺测不要填 0，先标记 df["precip_mm"] = pd.to_numeric(df["precip_mm"], errors="coerce") df["precip_missing"] = df["precip_mm"].isna() # 风速：统一米每秒，源数据若为 km/h 则除以 3.6 df["wind_ms"] = pd.to_numeric(df["wind_ms"], errors="coerce") # 合理范围标记，超出范围的先不删，打标后人工看 df["temp_outlier"] = (df["temp_c"] < -60) | (df["temp_c"] > 60) df["wind_outlier"] = df["wind_ms"] < 0 return df逻辑说明：pd.to_numeric(..., errors="coerce")把无法转换的值变成NaN，避免字符串混在数值列里；precip_missing单独标记缺测，后面聚合时可以用sum(min_count=1)保留缺测语义；temp_outlier和wind_outlier只打标不删除，避免把真实极端值误删。参数说明：温度合理范围按业务调整，示例给的是宽范围；风速负值一定是异常，标记后回查原始响应。常见误用是fillna(0)一把梭，后面算降水日数会把缺测城市算成无降水日。
4.2 时间对齐：时区、采集时刻与预报时效
时间字段是天气数据集最容易翻车的地方。采集时刻、观测时刻、预报时效是三个不同概念，不能混在一个字段里。观测时刻用数据源给的业务时间，采集时刻用请求落盘时间，预报时效单独存小时数。统一入库用 UTC，展示时再转业务时区。下面这段代码把时间字段拆开，并标记时区。
def align_time(df: pd.DataFrame) -> pd.DataFrame: df = df.copy() # 观测时间：数据源给的业务时间，先按 UTC 解析，实际按源时区调整 df["observe_time_utc"] = pd.to_datetime( df["observe_time"], errors="coerce", utc=True ) # 采集时间：原始层元数据里的 fetched_at_utc df["fetched_at_utc"] = pd.to_datetime( df["fetched_at_utc"], errors="coerce", utc=True ) # 预报时效：数值型小时，实况数据填 0 df["forecast_hour"] = pd.to_numeric( df.get("forecast_hour", 0), errors="coerce" ).fillna(0).astype(int) # 业务日期：按业务时区取日期，示例用 UTC+8 df["business_date"] = ( df["observe_time_utc"].dt.tz_convert("Asia/Shanghai").dt.strftime("%Y-%m-%d") ) return df逻辑说明：observe_time_utc和fetched_at_utc分开存，后面排查“数据是几点采的、观测是几点”时不会混；forecast_hour实况填 0，预报数据按小时数存；business_date用业务时区转出来，方便按天聚合。参数说明：时区示例用Asia/Shanghai，实际按业务区域替换；errors="coerce"把无法解析的时间变NaN，后面进失败队列。注意不要用采集日期代替业务日期，跨天采集时会把昨天的观测算到今天。
4.3 去重与增量合并：主键怎么选
去重主键选错，后面数据要么重复要么覆盖。天气数据的主键至少包含城市、观测时间、数据源、预报时效。城市用内部city_id，观测时间用 UTC，数据源用source_name，预报时效用forecast_hour。增量合并时用主键做左连接，新数据更新旧数据，旧数据不重复插入。下面这段代码用 pandas 做增量合并。
def merge_incremental(old: pd.DataFrame, new: pd.DataFrame) -> pd.DataFrame: key_cols = ["city_id", "observe_time_utc", "source_name", "forecast_hour"] # 新数据内部先去重，保留最后一条 new = new.sort_values("fetched_at_utc").drop_duplicates( subset=key_cols, keep="last" ) # 旧数据里删掉与新数据主键相同的记录，再用新数据补上 old_keys = old[key_cols].apply(tuple, axis=1) new_keys = set(new[key_cols].apply(tuple, axis=1)) old = old[~old_keys.isin(new_keys)] merged = pd.concat([old, new], ignore_index=True) return merged.sort_values(key_cols).reset_index(drop=True)逻辑说明：新数据先按fetched_at_utc排序去重，同一主键保留最新采集的；旧数据里把即将被新数据覆盖的主键删掉，再拼接，避免重复。参数说明：key_cols必须包含source_name，不同数据源同一城市同一时刻的数据不能互相覆盖；forecast_hour区分实况和不同时效预报。常见误用是只用city_id + observe_time去重，结果两个数据源的数据互相覆盖，分析时说不清来源。
4.4 导出：CSV、Parquet 与 SQLite 的取舍
导出格式按使用场景选。CSV 适合小数据集和人工检查，Parquet 适合列式分析和压缩存储，SQLite 适合本地查询和状态管理。天气数据集如果按天分区，Parquet 更省空间；如果要把结果交给不写代码的人，CSV 更直接。下面这张表做对比，后面给一段导出代码。
| 格式 | 适合场景 | 注意点 |
|---|---|---|
| CSV | 人工检查、小批量交付 | 编码统一 UTF-8，日期用 ISO 格式 |
| Parquet | 列式分析、按天分区 | 字段类型先固定，避免混类型 |
| SQLite | 本地查询、状态表 | 不适合超大数据集，注意索引 |
def export_dataset(df: pd.DataFrame, out_dir: str): from pathlib import Path out = Path(out_dir) out.mkdir(parents=True, exist_ok=True) # CSV 给人工检查，日期统一 ISO csv_df = df.copy() csv_df["observe_time_utc"] = csv_df["observe_time_utc"].dt.strftime("%Y-%m-%dT%H:%M:%SZ") csv_df.to_csv(out / "weather_dataset.csv", index=False, encoding="utf-8-sig") # Parquet 按业务日期分区，方便后面增量读 df.to_parquet(out / "weather_dataset.parquet", index=False)逻辑说明：CSV 导出前把时间转成 ISO 字符串，避免 Excel 打开后格式乱；encoding="utf-8-sig"让带 BOM 的表格工具能正确识别中文。Parquet 保留原始类型，后面按业务日期分区时可以只读某一天。参数说明：out_dir按数据集版本分目录，不要覆盖历史导出；Parquet 写入前确认时间列是datetime64[ns, UTC]，混类型会写失败。注意导出不是终点，导出后要用行数、主键唯一性、时间范围做一次校验。
5. 避坑与排查：天气爬虫最容易翻车的五件事
天气采集的坑大多不是代码语法，而是数据源和运行环境的变化。下面五条是实际跑批量任务时最常遇到的翻车点，每条按现象、原因、解决写。排查顺序建议先看原始层，再看状态表，最后看清洗层，不要一上来改解析代码。
5.1 接口字段说变就变，解析全挂
现象：任务突然大面积失败，日志里全是KeyError或字段值为None，但请求状态码还是 200。原因：数据源调整了字段名或嵌套层级，解析代码还在按旧结构取值。解决：解析前做 schema 校验，字段映射集中管理，原始响应必须保留。下面这段代码在解析前检查关键字段，缺字段时记录并跳过，不让整批崩掉。
REQUIRED_FIELDS = {"obsTime", "temp", "precip", "windSpeed"} def validate_payload(data: dict) -> list[str]: missing = [] for field in REQUIRED_FIELDS: if field not in data: missing.append(field) return missing def safe_parse(data: dict): missing = validate_payload(data) if missing: return {"ok": False, "missing": missing, "data": data} return {"ok": True, "data": data}逻辑说明：validate_payload只检查关键字段是否存在，不检查类型，类型问题留给清洗层；safe_parse返回结构化结果，调用方根据ok决定进成功还是失败队列。参数说明：REQUIRED_FIELDS按数据源实际字段维护，字段改名时先改这里；缺失字段不要直接抛异常，记录后继续处理其他城市。注意原始响应一定要存，字段改名后可以用回放脚本重新解析历史数据。
5.2 城市编码对不上，数据张冠李戴
现象：数据入库后按城市聚合，发现两个城市的数据混在一起，或者某个城市一直没数据。原因：内部city_id和数据源source_code混用，或者同名城市用了不同编码。解决：内部编码和数据源编码分开维护，采集前校验映射表，入库只写内部city_id，同时保留source_code便于追溯。下面这段 SQL 检查映射表里有没有重复或缺失。
-- 检查同一数据源编码是否映射到多个内部城市 SELECT source_name, source_code, COUNT(DISTINCT city_id) AS city_cnt FROM city_mapping GROUP BY source_name, source_code HAVING city_cnt > 1; -- 检查内部城市是否缺少某个数据源的映射 SELECT c.city_id FROM city_dim c LEFT JOIN city_mapping m ON c.city_id = m.city_id AND m.source_name = 'weather' WHERE m.city_id IS NULL;逻辑说明：第一条查询找“一个数据源代码对应多个内部城市”的冲突，第二条找“内部城市没有数据源映射”的缺口。参数说明：source_name区分不同数据源，映射表按数据源分开维护；city_dim是内部城市维度表。常见误用是直接用城市名称做关联，同名不同编码一关联就错，后面很难回查。
5.3 时间字段有时区，增量更新重复
现象：按天增量更新后，同一条观测数据出现两次，或者某天数据少了几小时。原因：采集时间用了本地时间，观测时间用了 UTC，增量条件按字符串比较，跨时区时边界对不上。解决：入库统一 UTC，增量条件用 UTC 时间范围，业务日期单独存一列。下面这段 SQL 按 UTC 时间范围取增量，避免本地时间混入。
SELECT * FROM weather_obs WHERE observe_time_utc >= '2025-06-01T00:00:00Z' AND observe_time_utc < '2025-06-02T00:00:00Z';逻辑说明：增量查询用半开区间，左闭右开，避免边界重复；时间字符串带Z表示 UTC。参数说明：业务日期列business_date只用于展示和按天聚合，不参与增量条件。注意不要用DATE(observe_time)直接比较，数据库时区设置不同会得到不同结果。
5.4 并发开太大，IP 被限流
现象：请求开始返回 429 或 403，重试后短暂恢复，随后又大面积失败。原因：并发数和请求频率超过数据源允许范围，重试叠加上去把瞬时压力放大。解决：降并发，加令牌桶，失败退避，限制单次回捞量。第 3 章的令牌桶可以直接用，另外把Retry的total降到 2 到 3 次，不要无限重试。参数说明：并发数从低往高试，先跑 2 到 4 个并发观察错误率；backoff_factor不要设太小，否则退避没有效果。注意被限流时不要换更多请求去试探，先停一批，等窗口过去再回捞。
5.5 磁盘写满/编码错误，任务半夜挂掉
现象：任务跑到一半报OSError: No space left on device，或者解析时UnicodeDecodeError。原因：原始响应未压缩、按城市存小文件太多，或者响应编码没有显式声明。解决：原始层用 gzip 压缩，按天分目录；解析时用resp.content加显式编码，不要依赖默认编码。下面这段代码在落盘前检查磁盘剩余空间，低于阈值就停止采集。
import shutil def check_disk(path="data", min_free_gb=5): total, used, free = shutil.disk_usage(path) free_gb = free / (1024 ** 3) if free_gb < min_free_gb: raise RuntimeError(f"disk free too low: {free_gb:.2f} GB") return free_gb逻辑说明：每次批量采集前检查磁盘剩余空间，低于阈值直接停，避免写到一半挂掉。参数说明：min_free_gb=5按数据量调整，全国范围采集建议留足空间；path指向数据盘。编码问题用resp.content.decode("utf-8", errors="replace")或按content-type里的 charset 解码，不要直接resp.text猜编码。
6. 进阶用法：用校验和与回放机制把数据集做稳
天气采集做到后面，真正拉开差距的是可回放和可验证。数据源改字段、解析逻辑升级、清洗规则调整，如果没有原始层和校验和，只能重新采一遍。我的习惯是每次采集给响应算内容指纹，原始层按指纹命名，回放脚本从原始层重建数据集。这样字段映射改了，不用重新请求，直接回放历史响应就能验证新逻辑。
6.1 给每次采集算指纹：内容校验和
指纹用响应体的 SHA-256 前 16 位就够，文件名里带上，元数据里也记一份。同一城市同一时刻如果采到两次不同指纹，说明响应内容变了，值得回看。下面这段代码在落盘时算指纹并写元数据，和第 2 章的落盘代码可以合并。
import hashlib def content_fingerprint(body: bytes) -> str: return hashlib.sha256(body).hexdigest()[:16] body = b'{"obsTime":"2025-06-01T00:00:00Z","temp":21.5}' print(content_fingerprint(body))逻辑说明：指纹只用于内容区分和回放索引，不用于安全校验；前 16 位足够区分单日单城市的响应文件。参数说明：指纹算法固定用 SHA-256，不要中途换算法，否则历史文件对不上；元数据里同时存完整文件名和指纹，回放时按指纹找文件。
6.2 回放：用原始响应重建数据集
回放脚本不请求网络，只读原始层和元数据，按当前解析逻辑重新生成清洗层。字段映射改了、单位转换改了，都先跑回放，对比新旧数据集的差异，再决定要不要全量重采。下面这段代码按天遍历原始文件，解析后写回放数据集。
from pathlib import Path import gzip import json def replay_day(day: str, raw_root="data/raw"): rows = [] for meta_file in Path(raw_root, day, "weather").glob("meta.jsonl"): for line in meta_file.read_text(encoding="utf-8").splitlines(): meta = json.loads(line) with gzip.open(meta["raw_path"], "rb") as f: body = f.read() # 这里调用当前解析函数，parse_response 需要能接收 bytes rows.append({"city_id": meta["city_id"], "body_size": len(body)}) return rows逻辑说明：回放只依赖原始层和元数据，不依赖网络；parse_response要改成能接收bytes或text，这样回放和实时采集共用一套解析逻辑。参数说明：day按 UTC 日期取目录；meta["raw_path"]是相对路径时先拼根目录。注意回放不要覆盖原始层，输出写到data/replay/下，方便对比。
6.3 一个具体技巧：把“采集日期”与“业务日期”分开
采集日期是任务跑在哪天，业务日期是观测属于哪天。跨天采集时，凌晨跑的任务可能采到前一天的业务数据。表里至少留collect_date、business_date、fetched_at_utc三个字段。增量更新按business_date取数，状态表按collect_date断点续采，日志按fetched_at_utc排序。这样重跑某天任务不会影响其他天的数据，回放也能按业务日期对齐。
从那以后我每次接天气采集任务，都先跑一遍原始响应回放，再扩城市范围；不先验证回放，后面字段一改就是全量重采。希望帮到你。
本文还有配套的精品资源，点击获取
