去年跑一个区域气候分析项目需要在 ECMWF 上下载 40 年的 ERA5 逐小时气压层数据算下来有 240GB 左右。一开始我是手动在网页上选参数、点提交、等下载一个变量一个月就要折腾几分钟。跑了三天下载量连零头都不够整个人直接裂开。后来实在忍不了花了一个周末把基于 Python API 的 ECMWF 大气数据批量下载与自动化处理这套流程彻底捋顺了从环境配置到批量提交、断点续传、自动重试再到下载后的数据校验和裁剪一条线全部串起来。这篇文章就是把这几天的实践完整记录下来包括代码、参数计算、踩过的坑以及怎么让下载任务无人值守地跑完。这套内容适合谁如果你正在做气象、水文、环境、新能源功率预测相关的项目需要批量使用 ERA5、ERA-Interim 或 CAMS 等 ECMWF 数据又不想在网页上一个一个点下载那这篇文章就是给你准备的。我尽量用最直接的方式讲清楚每一步包括为什么这么做而不只是给出代码让你抄。1. 整体设计思路手动下载为什么走不通API 方案怎么选1.1 手动下载的真实痛点量一大就失控ECMWF 的数据服务按理说是很好用的。网页交互界面做得很成熟选区域、选变量、选时间提交之后后台排队生成下载链接浏览器一点就能下载。听起来没问题但实际用起来全是坑。首先是任务数量问题。ERA5 单次请求有限制我这边实测下来一次性请求 40 年逐小时数据基本是不可能的后台会直接拒绝或者排队排到天荒地老。更常见的方式是把任务拆成小块比如每个变量每十年一块甚至每个变量每个月一块。这样一来一个项目少说几十个任务多则上千个任务。手动操作意味着你要反复填写同一套参数点几十上百次提交再去盯每个任务的状态确认下载链接有没有生成、有没有过期。其次是断点问题。浏览器下载一个大文件只要网络抖动一下或者电脑休眠下载就断了。ECMWF 给的是临时下载链接过期时间有限断了之后又要回去重新提交任务全部重来。那种几百个任务手动重提交的绝望感我想经历过的人都能懂。第三个问题是数据组织混乱。手动下载的文件名由 ECMWF 服务端生成带有很长一串 ID跟真实的时间范围、变量名对不上。下载完还要自己重新改文件名再分目录归档不然一年以后你根本不知道这个文件里装的是什么数据。当时我就做了一个决定要么写脚本自动化要么这个项目不做了。显然脚本是唯一解。1.2 工具选型官方 Python API、FTP 还是通用爬虫动手之前我简单调研过三条技术路线。第一条是用 ECMWF 官方提供的 Python API 库也就是 cdsapi。这个库封装了和 CDSClimate Data Store交互的所有细节你只需要按照 Key-Value 的格式提交请求参数然后等待结果。它的底层本质上是 HTTP 请求只是把认证、任务提交、状态轮询这些重复劳动全都包好了。第二条是走 ECMWF 的 FTP 服务器。ERA5 的一部分数据集可以通过 FTP 访问适合老手但问题在于数据集覆盖范围有限不是所有产品和变量都挂在 FTP 上而且没有统一的检索界面找数据本身就很费劲。第三条路是直接分析网页接口模拟浏览器请求也就是所谓的爬虫方案。这条路最灵活但维护成本最高。ECMWF 的网页结构只要一改版你的爬虫就得跟着改况且官方本身提供接口没必要自己造轮子。我的结论是用官方 cdsapi 库理由很简单它是 ECMWF 主推的方式数据服务更新时会同步更新 API兼容性最好请求参数和网页界面完全一致不用额外学习而且官方支持持续开发遇到问题社区答案也多。1.3 理解 CDS API 的工作机制用 cdsapi 前先理解一下它背后发生了什么。从本质上讲CDS API 是个异步任务系统。你提交一个数据请求服务端不会立刻把文件传给你而是先创建一个任务然后把任务放进队列。接下来你的客户程序需要反复轮询服务端问“我的任务好了吗”。任务完成后服务端会返回一个下载 URL这时候客户端再去拉取数据文件。这个流程其实跟你在网上冲印照片非常像。你上传照片、提交订单、商家开始制作你隔一会儿刷新一下订单状态状态变成“已完成”后你才能点击下载。你不可能提交完订单就立刻拿到成品。理解了这一层机制你会发现很多事情豁然开朗为什么请求可能排队很久因为服务端资源有限队列前面还有其他任务。为什么偶尔会出现请求失败因为网络连接可能在长时间轮询中断开。为什么下载链接有有效期因为服务端不可能无限期存储所有任务的结果文件。有了这个基础认知后续的容错、重试、批量调度写起来才不至于跑偏。2. 环境准备与 API 初始化先把地基打稳2.1 Python 环境搭建与依赖安装我用的是 Python 3.9 版本实际上 3.8 以上都行官方库对版本的要求没那么苛刻。如果你电脑里还没装 Python我的建议是直接从官网下载安装包安装时记得勾选“Add Python to PATH”不然后面在命令行里敲 python 会提示找不到命令。不过如果你跟我一样做数据分析比较多我更推荐直接装 Anaconda。它不是单纯的 Python而是打包了一整套科学计算环境里面已经预装了 conda 包管理器后面安装任何库都会方便很多。关键是它自带 Python 解释器省掉了手动配置环境变量这一步对新手极其友好。在环境方面你需要装这几类库cdsapiECMWF 官方下载客户端。xarray 和 netCDF4读取和操作 netCDF 格式的气象数据。pandas处理时间序列和元数据。pyyaml如果你的配置文件用的是 YAML 格式。tqdm显示下载进度条方便观察批量任务的状态。安装命令很直接pip install cdsapi xarray netCDF4 pandas pyyaml tqdm如果你用的是 conda也可以这样装conda install -c conda-forge cdsapi xarray netCDF4 pandas pyyaml tqdm我实测下来 conda-forge 源里的 cdsapi 版本比 PyPI 同步得稍晚一点但 LTS 使用没差别看你习惯哪个包管理器就行。2.2 注册账号、获取 API Key 与配置 .cdsapirc这是整个流程里最容易迷糊的一步因为 ECMWF 的账号体系有点绕。你需要去 CDS 网站注册一个账号注意不是 ECMWF 主站账号而是 Climate Data Store 的账号。注册完成后登录网站把页面右上角个人信息点开里面有一个“API key”栏目。你会看到两个关键信息一个是你注册时填写的邮箱或者用户 ID另一串是 URL 编码的 API Key大概长这样的格式uid: 12345 apikey: xxxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx接下来要做的是把这组信息写到本地配置文件里。在用户主目录下创建一个名为.cdsapirc的文件内容如下url: https://cds.climate.copernicus.eu/api key: 12345:xxxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx注意格式不能错URL 和你自己的 uid 用冒号连接 API Key。如果你用的是新版 CDS 的认证方式有些情况下还需要下载一个 API Key 文件里面可能是更长的字符串格式略有不同但核心逻辑一致就是把认证信息喂给客户端。为什么这个文件要放在用户主目录因为 cdsapi 客户端默认就去这里找配置省得你每次写代码都传一遍 key。文件名为.cdsapirc是隐藏文件Windows 下要注意显示隐藏文件才能看到Linux 和 macOS 下用ls -a就能看到。2.3 连通性验证先来一个最小请求测试环境配置好之后我先跑了一个最小请求目的是验证账号、API Key、网络链路全部通畅。这一步别急着一口气下载大任务不然真出问题光排查就够你折腾的。最小请求我选的是 2015 年 6 月 1 日单个日期的 2m 气温数据空间范围只取中国区域外的一个小格子。代码大概长这样import cdsapi c cdsapi.Client() c.retrieve( reanalysis-era5-single-levels, { product_type: reanalysis, variable: 2m_temperature, year: 2015, month: 06, day: 01, time: 12:00, area: [50, -5, 48, -3], format: netcdf }, test_download.nc )跑完以后检查本地是否生成了test_download.nc文件然后用 xarray 打开看一眼import xarray as xr ds xr.open_dataset(test_download.nc) print(ds)如果能正常打印出数据结构说明整条链路是通的。这个时候再去写批量下载脚本心里就有底了。3. 核心环节实现从单次请求到批量下载脚本3.1 拆解请求参数年份、月份、变量、区域的规划我的项目需要 1980 到 2020 年共 41 年的逐小时 2m 温度、10m 风场和地表气压数据使用的是 ERA5 单层数据集。要一次性请求这 41 年数据不用试都知道服务端肯定不干。因此我按“年份、月份”为粒度把任务切分。每一块任务对应一个变量、一年、一个月。这样算下来变量 3 个年份 41 个月份 12 个每个变量年月的请求数是41 x 12 492三个变量加起来就是 1476 个任务。听上去挺吓人的但脚本处理 1500 个任务和处理 10 个任务本质上没区别只是挂在后台跑的时间长短不同。关键是在构造请求参数时就固定好一个模板循环体里只替换年、月、变量这三个字段variables_map { t2m: 2m_temperature, ws10m: 10m_wind_speed, sp: surface_pressure } for var_key, var_name in variables_map.items(): for year in range(1980, 2021): for month in range(1, 13): request_params { product_type: reanalysis, variable: var_name, year: str(year), month: f{month:02d}, day: [f{d:02d} for d in range(1, 32)], time: [f{h:02d}:00 for h in range(0, 24, 6)], data_format: netcdf, download_format: unarchived }这里有一个需要强调的点time参数我按每 6 小时一条选了四个时次也就是 00 点、06 点、12 点、18 点。如果你要逐小时数据直接把range(0, 24, 1)填进去文件大小会显著增大下载耗时也会翻倍。另一个容易踩坑的地方是day参数。如果你直接写 1 到 31那么在 2 月、4 月、6 月、9 月、11 月这些短月份服务端会不会报错实测下来不会ECMWF 会自动忽略不存在的日期但保险起见我还是用了一个小工具函数根据年份判断每个月实际有多少天import calendar def days_in_month(year, month): return calendar.monthrange(year, month)[1]3.2 批量下载主循环任务提交、状态轮询与文件保存批量下载的核心循环其实不复杂真正复杂的是怎么应对各种异常情况。先给一版能跑的代码再一点点加防护。import cdsapi import os import time from tqdm import tqdm c cdsapi.Client() save_dir ./ecmwf_data os.makedirs(save_dir, exist_okTrue) def download_one(var_name, year, month): filename fera5_{var_name}_{year}_{month:02d}.nc filepath os.path.join(save_dir, filename) if os.path.exists(filepath) and os.path.getsize(filepath) 1024 * 1024: print(f跳过已存在文件: {filename}) return try: c.retrieve( reanalysis-era5-single-levels, { product_type: reanalysis, variable: var_name, year: str(year), month: f{month:02d}, day: [f{d:02d} for d in range(1, days_in_month(year, month) 1)], time: [00:00, 06:00, 12:00, 18:00], data_format: netcdf }, filepath ) except Exception as e: print(f下载失败 {filename}: {e}) for var_key, var_name in variables_map.items(): for year in range(1980, 2021): for month in range(1, 13): download_one(var_name, year, month)这个版本已经能解决 90% 的问题。它做了什么第一个是文件存在性检查。每次进入任务前先看本地有没有对应文件名且文件大小大于 1MB就跳过。这相当于实现了一个粗糙的断点续传机制。哪怕脚本因为网络原因中途退出下次重新运行已经下载成功的文件不会再被重复下载。第二个是异常捕获。每个任务单独包在 try-except 里个别任务失败不会导致整个脚本崩溃。这个设计非常重要因为 1500 个任务跑下来中间必然会有几个请求因为网络抖动、服务端排队超时等各种原因失败。3.3 并发下载并行提速的正确姿势起初我用的是单线程循环一个任务下载完才轮到下一个。实测一个任务大概 2 到 10 分钟遇到服务端排队高峰期可能更久。1500 个任务串行跑可能要连续跑好几天。后来我改成多线程并发下载速度提升非常明显。但这里有个前提ECMWF 服务端对单个账号的并发连接数和请求频率是有限制的。我看到网上有人开了 20 个线程结果被服务端识别为恶意请求直接封了一段时间的 API。我实测下来合理的并发放到 3 到 5 个线程比较稳妥。既能明显提速又不会触发服务端风控。用 Python 自带的concurrent.futures模块就能轻松实现from concurrent.futures import ThreadPoolExecutor, as_completed task_list [] for var_key, var_name in variables_map.items(): for year in range(1980, 2021): for month in range(1, 13): task_list.append((var_name, year, month)) with ThreadPoolExecutor(max_workers4) as executor: future_map { executor.submit(download_one, var_name, year, month): (var_name, year, month) for var_name, year, month in task_list } for future in tqdm(as_completed(future_map), totallen(task_list)): var_name, year, month future_map[future] try: future.result() except Exception as e: print(f任务异常 {year}-{month}: {e})3.4 请求被拒时的重试策略气象数据下载经常会遇到Request timed out或者Client.Error这类异常。第一次遇到时我以为是账号被限制了后来仔细查了日志才发现是服务端排队时间过长HTTP 连接断开了。我的解决办法是自定义一个带重试机制的下载函数遇到异常时等待一段时间再试。重点在于等待时间不能是固定值最好是递增的。第一次失败等 30 秒第二次等 60 秒第三次等 120 秒按 2 倍指数增长最多重试 5 次。这个策略借鉴了网络请求里的指数退避算法目的是避免短时间密集重试被服务端当成攻击。import time def download_with_retry(c, dataset, params, filepath, retries5): for attempt in range(1, retries 1): try: c.retrieve(dataset, params, filepath) return True except Exception as e: print(f第 {attempt} 次尝试失败: {e}) if attempt retries: raise e wait_time 30 * (2 ** (attempt - 1)) print(f等待 {wait_time} 秒后重试...) time.sleep(wait_time)为什么是 2 倍递增因为服务端在任务队列繁忙时短时间内的压力是持续性的。如果你等 10 秒就重试大概率还是失败。等 30 秒、60 秒、120 秒给服务端留出缓冲时间重试成功率会高很多。4. 自动化处理让下载任务无人值守4.1 增量更新只下载缺失的数据批量任务跑起来之后最希望看到的结果是“一键执行、全部搞定”。但因为网络、磁盘、服务端状态等不可控因素实际运行中总会出现意外。增量更新是解决这个问题最直接的办法。核心逻辑就一句话下载前先检查目标文件是否已经存在并且文件大小是否合理。如果存在且大小正常就跳过否则就重新下载。我在实际项目中做得更细一些除了判断文件大小还会尝试用 xarray 打开文件做一次轻量级验证。如果文件被截断netCDF 格式是打不开的这时候即使文件大小接近正常也需要删除重建。def is_file_valid(filepath): if not os.path.exists(filepath): return False if os.path.getsize(filepath) 1024 * 1024: return False try: with xr.open_dataset(filepath) as ds: _ ds.variables return True except Exception: return False这个方法在下载任务因网络中断而留下坏文件时特别管用。我第一次跑批量任务时凌晨断网导致七八个文件只有几十 KB那时候我还没写校验逻辑结果后处理阶段才发现问题只能回头重下白白浪费了时间。4.2 定时调度三种方案对比批量下载脚本写好后剩下的问题是“怎么让它自动开始、自动循环”。尤其是项目时间跨度长数据是持续更新的不能每天手动跑一次。我试过三种方式第一种是 cron。这是 Linux 和 macOS 上最经典的定时任务工具。在 crontab 里添加一行即可运行脚本0 3 * * * cd /path/to/project /usr/bin/python3 download_script.py意思是每天凌晨 3 点执行下载脚本。这个方案最轻量不依赖任何额外服务。缺点是 cron 对 Windows 用户不友好而且脚本一旦报错没有自动重跑机制。第二种是 APScheduler。它是 Python 生态里的定时任务库可以在脚本内部定义调度器from apscheduler.schedulers.blocking import BlockingScheduler scheduler BlockingScheduler() scheduler.add_job( funcmain_download_job, triggercron, hour3, minute0, iddaily_ecmwf_download ) scheduler.start()这种方案的优点是逻辑都在一个进程里方便与其他处理模块集成。缺点是调度器本身挂了任务就没了。需要配合 systemd 或者 nohup 常驻运行。第三种是云平台托管比如阿里云函数计算、AWS Lambda 或者 GitHub Actions。这些平台都支持定时触发把脚本部署上去跑完自动释放资源。优点是高可用缺点是环境配置稍微麻烦点尤其是涉及大文件下载时临时存储空间要预留够。我的选择本地机器加 cron外加强制日志记录。因为项目跑在实验室的一台 Windows 服务器上我用计划任务替代 cron效果一样。关键点在于脚本本身要写成“可重入”的也就是每次启动都会检查已有文件已经下载的自动跳过这样哪怕偶发中断下一次定时执行也能接着补漏。4.3 日志记录排查问题的第一现场没人愿意在脚本跑了三天后发现数据不对然后对着空空的命令行窗口判断哪里出了错。日志记录是自动化任务里最重要、也最容易被忽略的环节。我的做法是重定向 print 输出到日志文件每一行都带时间戳import logging logging.basicConfig( filenameecmwf_download.log, levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s ) logging.info(f开始下载 {year}-{month:02d} 的 {var_name})这看起来很简单但在排错时价值巨大。你能够看到每个任务的开始时间、结束时间、失败原因。如果某个文件下载特别慢你能判断是服务端排队还是网络问题。如果某个任务反复失败你能看到失败的类型然后针对性地调整参数。另外建议加一个“本轮任务汇总”的逻辑所有任务跑完之后统计成功、失败、跳过的文件数量。这样第二天早上打开日志一眼就知道昨晚跑得怎么样不用手动数文件名。5. 下载后的数据自动化处理从原始文件到可用数据5.1 数据完整性校验确保每个文件都可用下载完成不等于任务结束。ERA5 的下载文件偶尔会有问题最典型的是文件未完整写入、文件损坏。我在下载完成后加了一道校验环节。校验分三层第一层是文件大小检查过小的文件直接判为异常第二层是用 xarray 或 netCDF4 打开文件验证能不能正常解析第三层是对关键维度做检查比如时间维度的长度是否符合预期范围。一个月 6 小时一次的数据应该有4 x 天数值个时次如果时间维数不对说明数据不完整。def validate_netcdf(filepath): try: ds xr.open_dataset(filepath) expected_times days_in_month * 4 actual_times ds.sizes[time] ds.close() return actual_times expected_times except Exception: return False5.2 常见后处理区域裁剪、时间重采样与单位换算下载来的原始数据是全球范围的可能包含的网格点数量非常大。实际业务往往只需要某个区域、某些时间层级。这时候做后处理能显著缩小数据体积提高后续分析效率。区域裁剪最简单的方法是用 xarray 的.sel()方法按经纬度范围切片ds_sub ds.sel( latitudeslice(50, 30), longitudeslice(105, 125) )时间重采样最常用的场景是把 6 小时数据聚合成日平均ds_daily ds.resample(time1D).mean(dimtime)单位换算是一个容易忽略的环节。ERA5 原始数据中2m 气温的单位是开尔文 K而业务上我们通常习惯使用摄氏度。风速的单位是 m/s压强是 Pa。如果直接把 K 当成摄氏度用结果会差 273.15这个坑我见过不止一次。ds[t2m_celsius] ds[t2m] - 273.155.3 多文件合并与标准化输出当所有按月下载的文件准备好之后我们往往需要把它们合并成多年连续的数据集。xarray 的open_mfdataset是最顺手的工具import xarray as xr ds_all xr.open_mfdataset(era5_t2m_*.nc, combineby_coords) ds_all.to_netcdf(era5_t2m_1980_2020.nc)这里有一个性能提示如果文件数量非常大建议先关闭decode_cf以外的多余解码或者直接用chunk参数控制载入内存的块大小否则机器内存容易被撑爆。ds_all xr.open_mfdataset( era5_t2m_*.nc, combineby_coords, chunks{time: 365 * 4} )如果你有 CDOClimate Data Operations这个工具合并会更高效它是纯 C 写的处理大文件不占额外内存。但考虑到脚本闭环我主推 xarray 方案因为一套 Python 环境就能搞定不用额外安装外部程序。6. 常见问题排查与避坑锦囊6.1 CDS API 错误码速查表与应对策略我在工作中整理过一份 ECMWF 下载常见的错误码对照表按实际发生频率排序错误现象可能原因应对方式HTTP 400 Bad Request请求参数格式错误检查 dataset 名称、变量名、时间格式是否匹配HTTP 401 UnauthorizedAPI Key 无效或过期重新复制 .cdsapirc 中的 key确认冒号分隔格式HTTP 403 Forbidden数据集访问权限不足确认该数据集是否对你所属用户组开放HTTP 429 Too Many Requests请求频率过快降低并发数延长重试等待时间Request timed out服务端排队太久连接断开用指数退避重试必要时拆小任务块文件大小远小于预期下载中断或服务端处理异常删除文件重新下载用校验函数识别遇到报错不要慌第一步去看日志里的具体错误码第二步对应上面的表去调参数。大多数问题都不是代码逻辑错误而是请求参数或者账号权限问题。6.2 网络中断和磁盘爆满问题批量下载最怕的不是慢而是跑到一半断网或者磁盘满了。断网可以用重试机制缓解但磁盘爆满是硬伤几乎没法自动解决。我在脚本里加了一个磁盘剩余空间预检import shutil def check_disk_space(min_free_gb20): usage shutil.disk_usage(save_dir) free_gb usage.free / (1024 ** 3) if free_gb min_free_gb: raise RuntimeError(f磁盘空间不足剩余 {free_gb:.1f} GB)在每一轮批量任务开始前先检查一次空间不够就直接停住避免下载到一半因为磁盘满而写出大量残缺文件。6.3 容易被忽视的坑日志里没有失败记录的失败还有一类问题很隐蔽程序没有报错退出码为 0但下载结果就是不对。我遇到过两种情况。一种是 cdsapi 在retrieve时如果单次请求的数据量太大服务端可能返回一个不完整文件客户端不会报错但文件里只有几个时次的数据。这个问题很难从日志中发现必须靠数据校验环节去识别。另一种是内存泄露。批量循环中频繁打开和关闭 netCDF 文件Python 本身有垃圾回收机制但 xarray 和 netCDF4 在极端情况下会持有文件句柄导致磁盘空间被占用新文件写不进去。我的解决办法是尽量在任务函数内部用with语句管理文件对象及时释放资源。7. 写在最后几个让我彻底省心的细节整个流程跑通之后我最大的体会是批量下载 ECMWF 数据这件事代码本身不难写难的是把所有边界情况都考虑进去。你不需要写出多高深的代码但必须对“任务失败怎么办”“文件损坏怎么办”“服务端排队怎么办”这些问题提前设计好应对方案。我现在的实践流程大概是这样脚本启动时先校验本地已有文件正常则跳过异常则重下并发数设为 4每个请求最多重试 5 次等待时间按指数增长每下载完一个文件就用 xarray 验证一次验证不过就删除重来所有 print 都写入带时间戳的日志整个流程每周用定时任务自动触发一次增量补齐最新数据。运行到现在连续三个月没有人工干预数据完整率 100%。如果你也准备开始做这件事我建议从最小请求开始先跑通一年的数据下载确认整个过程没有坑再放开到全量批量。别一上来就想一次跑完几十年那样一旦出问题排查成本会让你怀疑人生。先小步快跑再稳步放量这才是最靠谱的节奏。