企业官网建设流程全解析

1. 先从"等待下载"说起：CDS的请求机制决定了脚本是刚需

去年冬天，我为了跑一次寒潮过程的环境场分析，从CDS上下载ERA5逐小时地面数据。任务提交很顺利，结果排队排了大半天，好不容易轮到自己，下载到一半连接断了。重新提交又等，第二次倒是下完了，zip包解压到一半突然报CRC校验失败。那一刻我意识到，面对ERA5这种动不动几十GB的气象再分析数据集，纯手动操作根本不具备可复现性，必须把它变成一段能自己"出警"的Python脚本。这篇博文想解决的事情也很明确：ERA5数据下载、批量解压、以及全程的Python错误处理，一次讲透，并且给出一套可以直接抄走的脚本。

提示：如果你只需要一小片区域、几个变量的数据，网站手动下载还能忍；但凡是涉及多年份、逐小时、多变量，或者之后要反复跑数据更新，脚本几乎是唯一靠谱的路径。

1.1 手动下载的痛点：不是下载难，是排队难

ERA5是欧洲中期天气预报中心推出的全球再分析数据集，时间跨度从1940年至今，空间分辨率大概30公里，能提供气压层和地面层的几十个变量。很多环境、水文、农业和交通行业的研究都拿它当基础输入。问题在于，CDS（Climate Data Store）这个平台并不像普通文件服务器那样给你一个直链，而是走"提交请求→进入队列→服务器后台组装数据→完成后通知下载"的异步流程。我见过太多人在网站页面上反复刷新，就为了等那个"Ready"按钮出现，高峰时段一等就是半个小时甚至更久，这还只是一次请求。

更麻烦的是ERA5单次请求往往数据量大，下载耗时动辄几十分钟。一旦网络波动或者浏览器休眠导致下载中断，前端页面不会帮你续传，整个流程又要从排队开始重来。这种设计本身是合理的——它针对的是科研用户批量取数的场景，而不是让人坐在浏览器前面盯进度条。所以结论很直接：凡是认真用ERA5的人，早晚都要走向脚本化。脚本可以半夜无人值守地跑，可以在队列里慢慢排，下载断了能重试，下载完能自动校验，这才是面对这种数据平台该有的工作方式。

1.2 API凭证与环境准备：别在第一步就卡住

在写脚本之前，需要先把CDS的API认证准备好。很多人卡在第一步不是代码问题，而是根本没拿到API key。打开CDS网站注册账号后，在个人主页里找到Personal Access Token，那里能看到你的UID和一串密钥。接下来在用户主目录下创建一个文件，Linux和macOS是~/.cdsapi，Windows是%USERPROFILE%\.cdsapi，内容格式非常固定：

url: https://cds.climate.copernicus.eu/api key: 你的UID:你的API密钥

这个文件的作用是让cdsapi库启动时自动读取凭证，不用每次在脚本里硬编码。我见过有人把key直接写在代码里然后随手发给别人，结果对方拿着这个key一顿下载，配额很快就被耗光了。所以强烈建议用.cdsapi文件保存，代码层面只保留对文件的引用。安装客户端库也很简单，一个命令搞定：

pip install cdsapi

如果你用的是虚拟环境（强烈建议），不要直接把包装进系统Python里，否则后面处理netCDF、GRIB时依赖冲突会非常头疼。我通常用miniconda建一个独立的era5环境，然后在这个环境里安装cdsapi和后续要用的xarray、netCDF4、cfgrib。很多人看网上教程直接开写脚本，结果一运行就报ModuleNotFoundError: No module named 'cdsapi'，虽然只是一个小问题，但确实会劝退不少新手——如果你也遇到这种情况，先确认自己当前激活的Python环境是不是你安装包的那个环境。

2. 核心下载函数怎么调：参数、数据格式和批量节奏

CDS API的使用逻辑被cdsapi库封装得很干净，核心就是Client对象的retrieve方法。但很多人第一次用的时候，会被里面那一大串参数搞懵：product_type、variable、year、month、day、time、data_format、download_format……这些参数并不是随便填的，它们直接对应CDS后台的数据集索引。参数写错，轻则请求直接被拒，重则任务提交成功但返回的数据不是你要的。这节我先给一个能跑通的最小示例，再解释批量下载时该怎么组织节奏。

2.1 一个能跑通的最小下载脚本

以ERA5单层资料（reanalysis-era5-single-levels）为例，下载2020年1月前10天的逐6小时2米温度和总降水量，脚本长这样：

import cdsapi client = cdsapi.Client() client.retrieve( "reanalysis-era5-single-levels", { "product_type": "reanalysis", "variable": ["2m_temperature", "total_precipitation"], "year": "2020", "month": "01", "day": [ "01", "02", "03", "04", "05", "06", "07", "08", "09", "10", ], "time": ["00:00", "06:00", "12:00", "18:00"], "data_format": "netcdf", "download_format": "unarchived", }, "era5_202001.nc", )

逐个解释一下关键参数。product_type在ERA5里一般固定为reanalysis，除非你在做前期数据同化对比。variable是变量列表，可以传多个；year、month、day、time这几个是时间维度，day和time都支持列表，方便一次性取多天多个时次。data_format指定返回的文件格式，可选netcdf或grib，绝大多数分析场景用netcdf就够了，因为xarray可以直接读，后续做插值、绘图、统计都非常顺。最后一个download_format很关键，它决定服务端返回给你的文件是不是被打包成zip。

很多人忽略download_format，结果明明只请求一个nc文件，返回的却是一个zip压缩包，还得额外解压一步。设置成unarchived后，如果目标结果只有一个文件，CDS会直接返回文件本体，省掉解压环节。但如果你的请求产生了多个文件（比如同时请求了多个变量、多段时期，服务端按块输出），那么无论怎么设置都只会返回zip。上面那个请求用unarchived是安全的，因为它本质只对应一个nc文件。

2.2 batch拆分：别一口气请求几十年数据

ERA5下载遇到的第一道坎，往往是"请求体太大"或者"任务在队列里卡死"。有一回我图省事，一次性请求20年的逐小时风场数据，结果CDS直接返回了一个400级别的错误，提示请求超出大小限制。这其实是CDS平台的硬性约束：单次请求能组装的数据量是有限的，变量越多、时间跨度越长、区域越大，越容易触发限制。

所以一个非常重要的实践是：把大批量下载拆成多个小请求。我自己的习惯是"按年拆分"，也就是循环遍历年份，每年一个retrieve请求。如果单年的逐小时数据仍然太大，再进一步拆到按月。这样做有双重好处：第一，单个请求体积小，服务器处理快，排队时间明显缩短；第二，中断恢复的时候损失可控——比如2020年下完，2021年还没下，脚本只需要从2021年继续，而不是整个20年请求从头再来。批量循环的代码框架大致如下：

import time years = [2019, 2020, 2021] for year in years: client.retrieve( "reanalysis-era5-single-levels", { "product_type": "reanalysis", "variable": ["2m_temperature"], "year": str(year), "month": "01", "day": ["01", "02", "03", "04", "05"], "time": "00:00", "data_format": "netcdf", "download_format": "unarchived", }, f"era5_{year}.nc", ) print(f"{year} 年下载完成", flush=True) time.sleep(2)

注意这里用time.sleep(2)在每次请求之间留一个间隔。别看只有两秒，它能让你的脚本在短时间内大量请求时显得更"温和"，降低被服务端限流的概率。CDS本身有并发限制和配额机制，如果你一个月内请求太多，账号会被暂时限制，所以节奏控制不是玄学，是真实存在的事情。print里加flush=True也值得养成习惯，这样日志能在重定向到文件时实时写入，不会因为缓冲区挂在那。

2.3 让脚本自己判断"这个文件我是不是已经下过了"

批量下载最大的风险不是某一次失败，而是脚本跑到一半中断后重新跑，前面已经下好的文件又被重复请求。CDS的队列和时间配额都是有限的，重复请求白白消耗配额不说，还可能触发限流。解决办法是在循环里增加文件存在性检查，用一个函数判断目标文件是否已经完整下载：

import os def has_valid_file(path): if not os.path.exists(path): return False if os.path.getsize(path) < 1024 * 10: return False return True

文件存在且大小超过10KB就认为是有效的，这是第一层校验。对于简单场景够用，但如果想更严格，可以增加zip完整性检查，这放到解压章节细说。在循环开头加上判断，如果has_valid_file返回True，直接跳过这次下载。这个逻辑看起来简单，但能让脚本具备真正的"可重入性"——无论你中断多少次，重头再跑它永远只补缺失的部分。对长时间无人值守的下载任务来说，这一条比任何花哨的优化都重要。

3. 收到的是zip不是nc：解压环节的完整处理方案

下载完成并不代表事情结束了。很多情况下，尤其是批量下载GRIB格式或者多时段数据，服务端返回的是一个zip压缩包。把zip安全地解压出来，其实有不少细节需要注意。这节我会讲zipfile的正确打开方式，以及解压之后常见的netCDF和GRIB处理路径。

3.1 安全解压：别让zip包里的文件路径给你挖坑

很多人解压zip的方式简单粗暴：直接zipfile.ZipFile(path).extractall(dest)。这个写法在99%的情况下没问题，但存在一个被反复提及的安全隐患——"Zip Slip"漏洞。恶意构造的zip文件里，条目文件名可能包含../，解压时会跳出你指定的目录，覆盖掉系统里其他文件。ERA5的压缩包虽然来自官方，不太可能被恶意构造，但你的脚本如果还承担着解压其他来源压缩包的功能，最好从一开始就养成安全解压的习惯。

还有一个更实际的问题：CRC校验失败。extractall并不会保证每个文件都完整体验证，有些损坏的zip会在解压中途抛异常，留下半截文件。保险的做法是先跑一遍testzip()，它会遍历zip内所有条目并校验CRC，如果发现损坏返回第一个坏文件名，没有损坏返回None。把这两个点合到一起，就是下面这个安全解压函数：

import os import zipfile def safe_extract(zip_path, dest_dir): os.makedirs(dest_dir, exist_ok=True) with zipfile.ZipFile(zip_path) as zf: # CRC 校验 bad_file = zf.testzip() if bad_file is not None: raise RuntimeError(f"Zip 校验失败，损坏文件: {bad_file}") for member in zf.infolist(): # 防 Zip Slip target_path = os.path.abspath(os.path.join(dest_dir, member.filename)) if not target_path.startswith(os.path.abspath(dest_dir)): raise RuntimeError(f"非法解压路径: {member.filename}") zf.extract(member, dest_dir) print(f"已解压到 {dest_dir}")

有人会问，为什么不直接extractall然后自己处理？因为extractall会在解压途中做遍历，一旦中途遇到异常，你很难知道已经解出来了哪些文件、哪些是残缺的。逐条infolist处理虽然稍微啰嗦一点，但能让你在执行过程中对每个文件有完全的掌控。配合前面的CRC校验，解压环节出错时你能立刻知道是哪个文件出了问题，而不是面对一堆半成品目录发呆。

3.2 解压后处理：netCDF读取、GRIB依赖和多文件合并

解压出来的东西，大概率是.nc结尾的netCDF文件。用xarray读取是最舒服的路径：

import xarray as xr ds = xr.open_dataset("era5_2020.nc") print(ds)

xarray会自动处理时间维、经纬度坐标、单位等信息，打印出来的数据结构一目了然。如果你下载了多年份的多个文件，想合并成一个数据集，可以使用open_mfdataset：

ds_all = xr.open_mfdataset( "era5_*.nc", combine="by_coords", )

combine="by_coords"会让xarray按时间坐标自动拼接，这是处理多年数据最常用的方式。有个小坑是，如果不同文件里的时间坐标有重叠，合并时会报错或导致奇怪的结果，所以文件命名保持一致、时间段不重叠非常重要。这也是为什么我习惯把文件名起成era5_2020.nc、era5_2021.nc这种带年份的形式，一目了然也好管理。

GRIB格式的情况稍微麻烦一点。GRIB是气象领域的老牌二进制格式，读取它需要cfgrib库，而cfgrib底层依赖eccodes。Linux环境通常一条conda install -c conda-forge cfgrib就能搞定，Windows下则经常出现DLL找不到的错误，非常劝退。如果你不是要跑WRF等数值模式，只想做数据分析，我的建议很直接：下载时指定data_format: "netcdf"，从一开始就别碰GRIB。这样可以规避掉整个eccodes的安装陷阱，把注意力集中在数据分析本身。

4. 错误处理实战：什么时候该重试，什么时候该认输

ERA5下载的体验用一个词概括就是"一波三折"。排队慢、下载慢、偶尔还断线。错误处理的本质不是把每个异常都吞掉，而是分辨哪些错误重试有用、哪些错误重试一万次也没用。这节是这篇文章的精华，也是我在实际使用中积累的最多经验的地方。

4.1 高频错误码速查：每个错误对应的正确操作

CDS的API错误不算特别复杂，但不同错误码背后的含义完全不同。我整理了一个速查表，方便你对照排查：

这里最需要澄清的是HTTP 400和HTTP 503的区别。400是"你错了"，无论重试多少次都一样，因为请求本身不合法，应该去检查变量名、日期范围、数据集名称这些参数。503则是"服务器暂时扛不住"，过一会儿可能就好了，这类错误非常适合自动重试。我见过不少人在循环里不管什么错误都反复重试，结果一个小时里全在打400的无效请求，白白浪费配额和时间。写错误处理前，先判断这个错误类别是不是可恢复的。

4.2 三层重试：请求层面、文件层面、人工层面

我对ERA5下载的设计习惯是三层防护，分别对应请求发出、文件落地、任务调度三个环节。

第一层是cdsapi客户端自带的请求重试。实例化Client时可以传入几个重要参数：

client = cdsapi.Client( retry=20, wait_until_complete=True, timeout=600, sleep_max=60, )

retry=20表示在请求提交阶段连续失败20次后才放弃；wait_until_complete=True会让脚本在请求提交后一直等待服务器把数据处理完成，而不是提交完就跑了；timeout=600是单个HTTP请求的超时时间，单位是秒，如果你下载的数据比较大，这个值可以适当调大；sleep_max则控制每次重试之间的最大休眠时间，避免重试太频繁。

第二层是文件层面校验，也就是前面说的has_valid_file和zip完整性检查。文件下完了不等于真的下好了，我见过太多次下载报告"成功"，但本地文件打不开的情况。所以文件落地后必须做大小检查和格式校验，不合格就删掉重下。这个步骤看起来多花了几秒钟，实际上是在给整个流程兜底。

第三层是任务调度层面的人工兜底。所有自动重试都失败之后，脚本应该把错误日志记录下来，发送一个明显的失败标记，而不是静默退出。我的习惯是让脚本在最后把成功和失败的文件列表分别打印出来，方便一眼定位哪一年哪个月出了问题。自动重试处理80%的偶发问题，剩下20%需要人来看，区分清楚才能真正省心。

4.3 wait_until_complete、超时与长任务的耐心问题

使用wait_until_complete=True会带来一个现象：脚本提交请求后会在retrieve内部长时间阻塞，表面看起来像卡死了。其实它是在等CDS那边组装数据，这是正常行为。如果脚本运行在你自己的电脑上，中途千万不要因为"没反应"就按Ctrl+C杀掉进程，否则任务白排队了。正确做法是让脚本后台运行，例如在Linux或macOS下用nohup：

nohup python era5_download.py > download.log 2>&1 &

这样即使终端关闭，进程也会继续跑。Windows用户则可以用PowerShell的Start-Process或者直接开着命令行窗口别关。日志里建议带时间戳，方便判断当前处于哪个阶段：

import datetime def log(msg): print(f"[{datetime.datetime.now():%Y-%m-%d %H:%M:%S}] {msg}", flush=True) log(f"开始提交 {year} 年请求") client.retrieve(...) log(f"{year} 年下载完成")

有了时间戳，你就能清楚地看到每次请求排队花了多久、下载花了多久，心里对耗时有个底。另外提醒一下，nohup模式下的输出会缓存在系统缓冲区，适当使用flush=True或给Python设置-u参数（python -u script.py）能保证日志实时刷新，排查问题时不会看到缺失的最后几行。

5. 完整一键脚本和三个我亲历的翻车现场

前面几节把下载、批量、解压、错误处理的关键点都拆开讲了。这一节我把所有思路整合成一个完整的一键脚本，然后复盘三个我实际踩过的坑。你可以直接照着用，也可以根据自己需求改。

5.1 脚本全文：下载+校验+安全解压一条龙

这个脚本支持按年循环下载ERA5单层netCDF数据，自动跳过已有文件，下载完成后自动解压，并对所有zip做CRC校验。它不像某些单行代码那样炫技，但胜在稳定、可重入、适合真正跑长期任务：

import cdsapi import os import time import zipfile import argparse BASE_DIR = os.path.expanduser("~/era5_data") def log(msg): print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}", flush=True) def safe_extract(zip_path, dest_dir): os.makedirs(dest_dir, exist_ok=True) with zipfile.ZipFile(zip_path) as zf: bad_file = zf.testzip() if bad_file is not None: raise RuntimeError(f"Zip 校验失败，损坏文件: {bad_file}") for member in zf.infolist(): target_path = os.path.abspath(os.path.join(dest_dir, member.filename)) if not target_path.startswith(os.path.abspath(dest_dir)): raise RuntimeError(f"非法解压路径: {member.filename}") zf.extract(member, dest_dir) def has_valid_file(path): if not os.path.exists(path): return False if os.path.getsize(path) < 1024 * 10: return False if path.endswith(".zip"): try: with zipfile.ZipFile(path) as zf: if zf.testzip() is not None: return False except zipfile.BadZipFile: return False return True def download_year(client, year, download_format="zip"): dest_dir = os.path.join(BASE_DIR, f"era5_{year}") zip_path = os.path.join(BASE_DIR, f"era5_{year}.zip") # 如果目录里已经有解压结果，跳过整个年份 nc_files = [f for f in os.listdir(dest_dir) if f.endswith(".nc")] if os.path.isdir(dest_dir) else [] if nc_files: log(f"{year} 已存在 {len(nc_files)} 个 nc 文件，跳过") return True # 如果 zip 有效，直接解压后跳过下载 if has_valid_file(zip_path): log(f"{year} 的 zip 已存在且完整，开始解压") safe_extract(zip_path, dest_dir) return True log(f"开始提交 {year} 年请求") client.retrieve( "reanalysis-era5-single-levels", { "product_type": "reanalysis", "variable": ["2m_temperature", "total_precipitation"], "year": str(year), "month": [f"{m:02d}" for m in range(1, 13)], "day": [f"{d:02d}" for d in range(1, 32)], "time": ["00:00", "06:00", "12:00", "18:00"], "data_format": "netcdf", "download_format": download_format, }, zip_path, ) if not has_valid_file(zip_path): raise RuntimeError(f"{year} 下载文件校验失败") safe_extract(zip_path, dest_dir) log(f"{year} 下载并解压完成") return True def main(): parser = argparse.ArgumentParser(description="ERA5 批量下载脚本") parser.add_argument("--start", type=int, required=True, help="起始年份") parser.add_argument("--end", type=int, required=True, help="结束年份") args = parser.parse_args() os.makedirs(BASE_DIR, exist_ok=True) client = cdsapi.Client(retry=20, wait_until_complete=True, timeout=600) for year in range(args.start, args.end + 1): try: download_year(client, year) except Exception as e: log(f"{year} 失败: {e}") log("继续下一个年份，全部结束后请检查失败列表") time.sleep(3) log("全部任务执行结束") if __name__ == "__main__": main()

使用方式很简单：

python era5_download.py --start 2019 --end 2021

这个脚本有两个设计细节值得说明。第一，判断"是否已经下载过"时，优先检查解压后的nc文件有没有存在，而不是只看zip有没有下载过。因为实际使用中经常出现zip下完了但还没解压的情况，如果只看zip存在就直接跳过解压，数据等于白下。第二，每下载一个zip就立刻解压，解压完不急着删zip，而是先留着。这样做的好处是，万一解压后的nc在后续处理中被误删，原始zip还能再解一次，不用重新排队下载。等整个项目彻底结束、数据确认无误，再手动清理这些zip也不迟。

5.2 翻车现场复盘：有些坑只有跑过才知道

第一次翻车是用一次性请求下载20年的数据。我当时天真地以为，ERA5的请求就和文件下载一样，无非是一次请求、一堆数据。结果CDS直接甩回来一个错误，提示请求体太大了。后来才知道每个请求都有时间范围和变量数量的隐形上限。从那以后我彻底改成了按年循环，数据量大的时候按月循环，再没遇到过因为请求过大而直接被拒的情况。

第二次翻车发生在解压环节。我下载40GB的zip到本地磁盘，解压到一半磁盘满了，进程被系统杀掉，留下了一堆残缺的nc文件和半解压的临时目录。最气人的是，因为zip本身没有损坏，重新解压时又要从头开始，占用同样多的磁盘空间。从那以后，我在解压前都会检查磁盘余量，并且优先把压缩包放在数据盘而不是系统盘。你可以用下面的代码快速看磁盘空间：

import shutil total, used, free = shutil.disk_usage(BASE_DIR) print(f"剩余可用空间: {free / 1024**3:.2f} GB")

如果剩余空间不足，宁可先删掉一些中间产物，也不要硬着头皮解压。数据文件在空间不足时会非常快地损坏，而且有时候文件大小看起来正常，实际内容已经写坏了，这种问题最难排查。

第三次翻车是一个特别容易被忽略的小问题：文件名冲突。我在循环里下载不同年份的数据时，用了固定的文件名download.nc，结果所有年份的数据互相覆盖，最后只剩下一年的文件。这种错误属于典型的"低级但致命"，完全可以通过在目标文件名里拼接年份来避免。上面脚本里我把文件名设计成era5_{year}.zip，就是从这次教训里得来的。你如果自己改脚本，一定仔细检查每一轮循环写入的文件名是不是唯一的，不然批量任务跑完一看，进度全假的，心态直接爆炸。