Python 下载 CFSv2 数据：目录规则、断点续传与并发优化

简介：这是一份用于自动化下载CFSv2气象数据的Python脚本资源，面向气象科研人员、气候模型开发者以及需要批量获取NCEP再分析产品的学习者。脚本通过调用UCAR数据接口完成身份认证与数据下载，使用者只需在代码中替换账号密码占位符即可运行，能够有效摆脱手动登录网页逐条下载的繁琐流程。压缩包内包含1个py脚本，整体大小约1KB，轻量简洁，适合具有一定Python基础、熟悉网络请求与文件操作的读者直接使用。目前已有600人学习下载。阅读此脚本可以掌握CFSv2数据接口的调用方法、HTTP认证流程、分块下载与异常重试等常见处理技巧；同时脚本中预留了文件系列ID的修改位置，便于按需抓取不同时间范围或指定参数的数据，为气候分析、模式验证和科研工作提供可靠的数据支撑。

1. 用 Python 下载 CFSv2 数据，先搞懂目录再写循环

做气象、水文或者新能源功率预测的人，迟早要面对 CFSv2 这组数据。NCEP 的气候预报系统第二版提供从 2011 年至今的历史回算和实时预报，时间分辨率 6 小时、水平分辨率约 0.5 度，覆盖全球。真要把一段时间的风场、温度、辐射拿全，按每天 4 个起报时刻计算，单变量单月份就是 120 个文件，手动从 NOMADS 网页一个个点显然不现实。

更反直觉的一点是：CFSv2 的文件目录结构看起来层级很深、命名很乱，但文件名规则实际上极其统一。这意味着你不需要先上 xarray、siphon 这类重量级库，只用 Python 标准库加 requests 就能写出一个稳定、可断点续传的下载脚本。这篇文章就把 URL 构造规则、并发下载姿势、文件完整性校验和 OPeNDAP 按变量读取这几件事一次讲透，新手能照着改成自己的下载器，老手也能看到一些容易翻车的边界情况。

2. CFSv2 在 NOMADS 上的目录结构与文件名规则

2.1 数据集编号 ds094.0 与两类数据入口

NOMADS 把不同模式输出按数据集编号组织，CFSv2 对应的编号就是ds094.0。进入https://nomads.ncep.noaa.gov/pub/data/nccf/com/cfs/prod/cfs/能看到按日期组织的目录，这是实时预报产品的落地位置；而历史回算数据则放在https://nomads.ncep.noaa.gov/pub/data/nccf/com/cfs/prod/cfs.20110101/这类带日期的路径下。注意ds094.0在 NOMADS 的数据清单页面里标识的是 CFSv2 整体，真正下载时路径里不会直接出现ds094.0这个字符串。

CFSv2 输出分为两大类：一类是time-series格式，把所有变量按时间维排进同一个文件，适合直接做时间序列分析；另一类是grib2格式的逐时效文件，文件名以pgb开头，每个文件只含某个起报时刻、某个预报时效的全球场，适合做 WRF 初始场或者自己控制变量范围。绝大多数下载脚本都围绕 grib2 这组来做，因为它的文件边界清晰，下载失败后补起来也容易。

2.2 从时间参数推算文件名的 Python 工具函数

grib2 文件的名字由两部分拼出来：前缀加时间戳。以 6 小时间隔的预报文件为例，典型文件名长这样：

pgbf06.gdas.2011010100.f000 pgbf06.gdas.2011010100.f006 pgbf06.gdas.2011010100.f012

pgbf06中的f06表示 0.5 度网格（另一个常见前缀是pgbf00，对应 1 度网格），gdas在这个语境下只是历史回算产品的固定标识，后面的2011010100是起报时刻，f000表示预报时效 0 小时，也就是分析场。一天有 00、06、12、18 四个起报时刻，每个起报时刻最长预报到 384 小时（16 天），但很多研究只用到前 120 小时。

写下载脚本时，最核心的转换函数就是给定一个datetime和预报时效，生成对应的文件名和 URL。代码里我一般这么写：

from datetime import datetime, timedelta def cfsv2_filename(base_time: datetime, forecast_hour: int) -> str: """根据起报时刻与预报时效生成 CFSv2 文件名。""" prefix = "pgbf06" # 0.5 度网格产品 ymdh = base_time.strftime("%Y%m%d%H") return f"{prefix}.gdas.{ymdh}.f{forecast_hour:03d}" def cfsv2_url(base_time: datetime, forecast_hour: int) -> str: """构造成完整的 NOMADS 下载 URL。""" year = base_time.strftime("%Y") ymdh = base_time.strftime("%Y%m%d%H") filename = cfsv2_filename(base_time, forecast_hour) return ( f"https://nomads.ncep.noaa.gov/pub/data/nccf/com/cfs/prod/" f"cfs.{ymdh}/{year}/{ymdh}/" f"time_grib2/{filename}" )

这段代码里值得注意的地方有三个。第一，strftime("%Y%m%d%H")直接拼接出 10 位时间戳，比手写字符串加法更不容易出错，起报时刻永远取整点，不会有分钟和秒。第二，URL 路径里同一个ymdh出现了两次，一次在cfs.前缀后面表示产品目录，一次在year子目录下面表示文件所在目录，少一层都会 404。第三，f{forecast_hour:03d}把时效补成三位数，f000、f006这种格式不能写成f0或f6，否则服务器直接返回找不到文件。

2.3 下载前必须确认的 3 个参数

写循环之前把下面三项确认清楚，能省掉大量试错时间。第一是时间范围：历史回算数据从 2011 年 1 月 1 日开始（再早之前是 CFSR，数据集不同），如果你要 2010 年的数据，这个 URL 模板的目录结构不适用。第二是变量文件类型：需要全变量就下pgbf06，只想要某几个变量可以下pgbf00然后用 wgrib2 裁剪，或者走后文要讲的 OPeNDAP。第三是预报时效步长：CFSv2 前 9 天是逐 6 小时输出，9 到 16 天变成逐 12 小时，如果你做的是短期预报，直接取[0, 6, 12, 18, 24]这个列表就行；要拉满 16 天，得在生成时效时加一个条件判断。

另外提醒一点，CFSv2 的文件体积不算小，0.5 度全变量的 grib2 单文件大约在 60 到 120 MB 之间（不同时效略有差异），先估算总量再决定一次拉多少。一个月 120 个文件就是 10 GB 量级，磁盘规划别忽略了。

3. 用 Python 标准库实现可断点续传的 CFSv2 下载脚本

3.1 requests 下载与分块写盘的基本流程

CFSv2 数据下载最朴素的实现是用requests.get把整个响应读进内存再写文件。这种做法在文件只有几 MB 时没问题，但面对上百 MB 的 grib2 文件，一次性r.content会占用大量内存，而且一旦连接中断，前面下载的进度全部作废。所以我习惯用stream=True配合分块写入，代码长这样：

import requests from pathlib import Path def download_file(url: str, dest: Path, chunk_size: int = 1024 * 256): """分块下载单个文件，chunk_size 控制每次写入的字节数。""" dest.parent.mkdir(parents=True, exist_ok=True) with requests.get(url, stream=True, timeout=(10, 60)) as r: r.raise_for_status() with open(dest, "wb") as f: for chunk in r.iter_content(chunk_size=chunk_size): if chunk: f.write(chunk)

这里有两个参数值得细说。timeout=(10, 60)是连接超时和读取超时的二元组，NOMADS 服务器在高峰期响应慢，连接超时设 10 秒是合理的，读取超时 60 秒保证长时间没有新数据到达时能报错而不是无限挂起。chunk_size取 256 KB 是个折中：太小会导致频繁磁盘写入、CPU 占用高，太大则内存压力上升，实测在网络稳定的环境下这个值对吞吐量影响不大，反而是重试逻辑对整体体验影响最大。

3.2 让脚本扛住网络抖动：重试与 Range 续传

气象数据的下载场景通常在实验室或公司内网，偶尔会碰到代理超时、DNS 抖动、服务器限流。一个健壮的下载脚本必须考虑两件事：失败后重试，以及已下载一半的文件不要从头再来。这两件事靠requests的headers参数就能实现，核心是利用 HTTP 的Range头。

import time import requests from pathlib import Path def download_with_resume(url: str, dest: Path, max_retries: int = 5): """带断点续传和指数退避重试的下载函数。""" dest.parent.mkdir(parents=True, exist_ok=True) existing_size = dest.stat().st_size if dest.exists() else 0 for attempt in range(max_retries): headers = {"Range": f"bytes={existing_size}-"} try: with requests.get(url, stream=True, timeout=(10, 120), headers=headers) as r: if r.status_code == 416: print(f"文件已完整，跳过 {dest.name}") return True r.raise_for_status() mode = "ab" if existing_size > 0 else "wb" with open(dest, mode) as f: for chunk in r.iter_content(chunk_size=1024 * 256): if chunk: f.write(chunk) existing_size += len(chunk) return True except (requests.RequestException, ConnectionError) as e: wait = 2 ** attempt print(f"第 {attempt + 1} 次下载失败: {e}，{wait} 秒后重试") time.sleep(wait) return False

这段代码的逻辑比初版多了两个关键点。第一，请求头里的Range: bytes=...告诉服务器从指定字节开始返回，这样脚本中断后再次运行时先取本地文件大小，接着往下传。第二，416状态码表示请求的范围超出文件长度，也就是文件其实已经下完整了，这种情况直接返回成功，避免每次都因为「文件已存在」而重复下载。

一个常见的坑是：不少服务器不支持 Range，会忽略这个头直接返回 200 和全量内容。NOMADS 目前支持 Range，但如果你把这段代码换成别的数据源，最好先打印r.status_code和r.headers.get("Content-Length")确认一下。另外，断点续传依赖本地文件大小的准确性，如果之前下载过程中文件被截断但恰好和服务器上某一段长度相同，续传后会导致文件尾部错位。稳妥的做法是下载完成后校验一次文件大小，逻辑放到第 4 章讲。

3.3 主循环：按时间网格生成任务并逐一下载

有了 URL 生成函数和带重试的下载函数，主循环就非常直接了。把起始日期、结束日期、每日起报时刻、预报时效列表依次遍历，逐个调用下载函数。

from datetime import datetime, timedelta start = datetime(2023, 1, 1, 0) end = datetime(2023, 1, 3, 0) forecast_hours = [0, 6, 12, 18, 24] current = start while current <= end: for fh in forecast_hours: url = cfsv2_url(current, fh) filename = cfsv2_filename(current, fh) dest = Path("cfsv2_data") / filename ok = download_with_resume(url, dest) if not ok: print(f"最终失败: {url}") current += timedelta(hours=6)

timedelta(hours=6)在这里是核心，它保证了起报时刻一定按照 00、06、12、18 的顺序推进，不会出现字符串拼接导致的进位错误。两组日期之间如果出现闰年或者跨月，datetime的日期运算会自动处理，不需要单独判断。下载目录cfsv2_data保持扁平结构，文件名本身已经包含了完整的起报时间和时效信息，后续处理时按文件名解析即可。

4. 把下载速度提上去：并发、校验与按需取变量

4.1 用 ThreadPoolExecutor 做并发下载的合适并发数

单线程逐个下载 120 个文件，每个文件按 50 MB 计算，在 10 MB/s 的网络下要 10 分钟。CFSv2 数据下载是典型的 IO 密集型任务，瓶颈在网络延迟和服务器带宽，不是 CPU，所以用多线程能显著提高吞吐量。concurrent.futures.ThreadPoolExecutor比手动开threading.Thread简单得多，失败重试逻辑可以原样复用。

from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from datetime import datetime, timedelta def download_task(base_time: datetime, fh: int, out_dir: Path): url = cfsv2_url(base_time, fh) dest = out_dir / cfsv2_filename(base_time, fh) ok = download_with_resume(url, dest) return url, ok with ThreadPoolExecutor(max_workers=6) as executor: futures = [] current = datetime(2023, 1, 1, 0) end = datetime(2023, 1, 31, 0) while current <= end: for fh in [0, 6, 12, 18, 24]: futures.append(executor.submit(download_task, current, fh, Path("cfsv2_data"))) current += timedelta(hours=6) for future in as_completed(futures): url, ok = future.result() if not ok: print(f"需手动补下: {url}")

max_workers=6是我在常规网络环境下常用的值。设太小浪费带宽，设太大会触发 NOMADS 的并发连接限制，表现为大量连接被重置或直接超时。如果你在内网有代理或者镜像，可以尝试 8 到 10；直连外网时 6 是一个稳妥起点。另外要注意，download_with_resume里的重试逻辑在线程池中依然有效，但多个线程同时打印日志会交错，可读性变差，可以给print加上线程 id 前缀，这里不展开。

4.2 校验文件是否完整：Content-Length 对齐与 grib2 抽样检查

下载完成后最重要的动作是验证文件完整性，否则 grib2 文件尾部缺失会在后续解码时产生各种诡异报错。最轻量的校验是用requests.head获取服务器的Content-Length和本地文件大小做对比。

import requests from pathlib import Path def verify_file(url: str, dest: Path) -> bool: """对比服务器 Content-Length 与本地文件大小。""" if not dest.exists(): return False try: r = requests.head(url, timeout=10) remote_size = int(r.headers.get("Content-Length", 0)) local_size = dest.stat().st_size return remote_size == local_size except requests.RequestException: return False missing = [] for p in Path("cfsv2_data").glob("pgbf06.*"): url = cfsv2_url(datetime.strptime(p.stem.split(".")[2], "%Y%m%d%H"), int(p.stem[-3:])) if not verify_file(url, p): missing.append(p.name) print("不完整的文件:", missing)

这里有两个细节容易忽略。第一，requests.head对 NOMADS 这类静态文件服务通常是有效的，但有些 CDN 对 HEAD 请求不返回Content-Length，这时需要退回到 GET 请求且stream=True再读取响应头。第二，文件名解析用了两次split，第一次取点号分隔后的第三段得到时间戳，第二次取后缀的后三位得到时效，这种解析方式依赖文件名格式严格统一，CFSv2 的文件名满足这个约束，但如果哪天换成别的数据源就要重新写解析逻辑。

如果系统里装了 wgrib2 或 grib_ls，我建议再做一个抽样检查：随机挑几个下载完的文件，用grib_ls -p forecastTime,centre查看 GRIB 消息头里的预报时效和中心编号。这一步能同时验证文件不是 HTML 错误页——有些代理会拦截下载请求并返回一个 200 的 HTML 页面，扩展名虽然是.f000，但内容完全不是 grib 数据。单靠文件大小对比发现不了这类问题。

4.3 只想用某几个变量：用 xarray 走 OPeNDAP 按需读取

CFSv2 的 grib2 文件包含几十个变量，如果你只想要 2 米温度或者 10 米风场，整个文件下载下来既浪费带宽又占磁盘。NOMADS 为 CFSv2 提供了 OPeNDAP 端点，xarray 可以直接按变量、按经纬度范围远程切片，只下载真正用到的数据。

import xarray as xr url = ("https://nomads.ncep.noaa.gov:9090/dods/cfs/cfs.20230101/" "cfs.t00z.pgrbf06") ds = xr.open_dataset(url) # 查看变量名后按 name 筛选 t2m = ds["tmp2m"].isel(time=0).sel(lat=slice(30, 20), lon=slice(110, 130)) value = t2m.values

这段代码里open_dataset并不会把整个数据集拉进内存，只有在isel、sel之后触发values时才会通过网络读取切片数据，所以可以先放心查看ds.variables再决定取哪些变量。OPeNDAP 的 URL 端口是 9090，路径结构和 HTTP 下载目录不完全一样，确定具体的 ncss 或 dods 端点后最好先在浏览器里访问一次确认字段名。

值得注意的是：如果你不光要分析数据本身，还想拿 CFSv2 当 WRF 的初始场和侧边界，那就必须下载完整 grib2 文件，因为 WPS 的 ungrib 工具要读取的是完整 GRIB 消息而不是变量切片。所以这一节和上一节的取舍是用途决定的：做统计分析和画图，走 OPeNDAP；做动力模式驱动，老老实实下整个文件。

5. 易踩的坑与验证下载结果的三个技巧

5.1 时区与文件名解析的边界条件

CFSv2 所有时间都是 UTC，脚本里一旦混入本地时间，最直接的后果是current这个迭代变量偏移 8 个小时，文件名里的%H变成 08 而不是 00，服务器上根本不存在这个目录。如果你在服务器上跑脚本且服务器时区不是 UTC，建议在文件开头统一os.environ["TZ"] = "UTC"并至少把datetime.utcnow()作为默认结束时间，不要让脚本依赖系统时区配置。

5.2 跨月和闰年场景先算一遍期望文件数

按第 3 章的时间循环逻辑，2024 年 2 月会自然生成 29 日的日期序列，这个逻辑本身没错，但 2024 年 2 月 29 日当天的 CFSv2 数据是否已经归档到 NOMADS，取决于当前日期和服务器同步策略。更常见的坑是跨年月时目录结构变化，比如 2022 年 12 月 31 日 18 UTC 起报的 f024 时效，物理时间是 2023 年 1 月 1 日 18 UTC，但文件路径仍然在cfs.2022123118下面。用本文第 2.2 节的 URL 生成函数就不会出错，因为路径始终绑定起报时刻，而不是有效时刻。

验证脚本是否覆盖了完整网格，可以用一个离线文件清单对比：先跑一遍脚本生成所有 URL 存到manifest.txt，再去下载循环里对每个 URL 检查对应文件是否存在。这样跨月、跨年的目录遍历是否正确，一眼就能看出来。

5.3 磁盘已满时的 fail-fast 处理

100 个 grib2 文件落盘前先算总大小：120 个文件乘以单文件平均 80 MB，约 9.6 GB。如果目标磁盘只剩 5 GB，写满之后文件系统报错，前面的所有下载文件都要挪走才能继续。我一般在脚本开头加一个总大小预估函数，遍历待下载列表，把requests.head拿到的Content-Length累加，然后和shutil.disk_usage的剩余空间做比较，不够就提前退出。

最后留一个最简单也最有效的验证技巧：改起始时间跑一遍 2011 年 1 月 1 日 00 时这一个起报时刻的 f000 和 f006 两个文件，用 wgrib2 打印它们的forecastTime，确认一个是 0 一个是 6，再对比文件大小是否都落在合理区间。跑通这两个文件，整条 URL 构造、下载、校验的链路就都验证过了，后续再放开时间范围就不会出意外。

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