告别数据荒！手把手教你从NCDC和Reliable Prognosis获取中国气象站点数据（附Python处理脚本）

气象数据获取与处理实战指南：从数据荒到高效分析

最近在帮一位地理信息系统专业的研究生处理气象数据时，发现很多研究者都面临同样的困境——要么找不到可靠的数据源，要么数据格式混乱难以直接使用。这让我意识到，气象数据的获取和预处理环节可能比想象中更耗费时间。本文将分享几个实用的解决方案，帮助研究者们快速获取并处理中国区域的气象站点数据。

1. 主流气象数据源深度评测

选择合适的数据源是研究的第一步。经过实际测试和对比，以下三个平台在数据覆盖范围和可用性方面表现突出：

1.1 NOAA国家环境信息中心(NCEI)

作为美国国家海洋和大气管理局的下属机构，NCEI提供了全球范围内的气象数据。其ISD-Lite数据集包含中国400多个气象站点的历史记录，数据字段包括：

• 气温（日最高/最低/平均）
• 降水量（6小时累计）
• 风速和风向
• 露点温度
• 气压测量

数据特点对比表：

指标	NCEI优势	NCEI局限性
时间范围	1901年至今	部分中国站点2013-2019年数据缺失
更新频率	每日更新	数据发布有3个月延迟
数据格式	标准化的CSV	需要转换单位（如降水×0.1）
获取方式	FTP批量下载	需要处理大量小文件

提示：使用NCEI数据时，建议先检查目标站点的数据完整性，特别是2013-2019年期间的中国站点数据。

1.2 Reliable Prognosis平台

这个由俄罗斯团队维护的平台提供了更友好的用户界面，特别适合需要快速获取特定站点数据的场景：

# 示例：通过API获取单站数据 import requests station_id = "545110" # 北京站代码 start_date = "20230101" end_date = "20231231" params = { 'station': station_id, 'start': start_date, 'end': end_date, 'elements': 'temp,precip' # 气温和降水 } response = requests.get("https://rp5.ru/data/archive/download", params=params) with open(f"weather_{station_id}.csv", "wb") as f: f.write(response.content)

平台优势包括：

• 直接计算日/月/年统计值
• 支持多种导出格式
• 提供数据质量标识

但需要注意：

• 仅支持单站查询
• 部分站点数据存在间断
• 降水数据可能存在系统偏差

1.3 青藏高原科学数据中心

对于研究中国西部地区的学者，这个平台提供了独特的价值：

• 1km分辨率格点数据
• 专门针对高原地区的优化算法
• 包含ERA5再分析数据的区域优化版本

2. 数据获取自动化方案

手动下载和处理大量气象数据效率低下。以下是几种自动化方案：

2.1 NCEI数据批量下载

import ftplib import os def download_ncei_data(station_ids, years, output_dir): ftp = ftplib.FTP('ftp.ncdc.noaa.gov') ftp.login() for station in station_ids: for year in years: remote_path = f'/pub/data/noaa/isd-lite/{year}/{station}-99999-{year}.gz' local_path = os.path.join(output_dir, f"{station}_{year}.gz") try: with open(local_path, 'wb') as f: ftp.retrbinary(f"RETR {remote_path}", f.write) print(f"下载成功: {station} {year}") except: print(f"文件不存在: {station} {year}") ftp.quit() # 示例调用 stations = ['545110', '545270'] # 北京和天津站 download_ncei_data(stations, range(2020, 2023), 'weather_data')

2.2 数据质量快速检查

获取数据后，应立即进行基础检查：

import pandas as pd def check_data_quality(filepath): df = pd.read_csv(filepath) # 检查缺失值比例 missing_ratio = df.isnull().mean() # 检查异常值 temp_stats = df['temperature'].describe() precip_stats = df['precipitation'].describe() return { 'missing_data': missing_ratio, 'temperature_range': (temp_stats['min'], temp_stats['max']), 'precipitation_max': precip_stats['max'] }

3. 降水数据处理实战

原始数据中的降水记录往往需要特殊处理。以下是处理NCEI 6小时降水数据的完整方案：

3.1 数据清洗步骤

1. 无效值处理：剔除-9999、0、-1等特殊值
2. 单位转换：原始数据需除以10得到实际毫米数
3. 时间对齐：处理不同时次的观测记录

3.2 日降水量计算逻辑

根据观测频率的不同，采用三种计算策略：

1. 稀疏数据（每日1-3次观测）：

   • 直接取各次观测值的和
   • 或选择特定时次（如08时、20时）的观测

2. 常规数据（每日4次观测）：

   • 标准做法：00Z、06Z、12Z、18Z四个时次的和
   • 替代方案：03Z、09Z、15Z、21Z时次的和（当主要观测时次缺失时）

3. 高密度数据（每小时或每3小时）：

   • 累计24小时内的所有观测
   • 或滑动窗口累计

3.3 Python实现代码

import pandas as pd import numpy as np from pathlib import Path def process_precipitation(data_dir): result = [] for file in Path(data_dir).glob('*.csv'): df = pd.read_csv(file) # 基础清洗 df = df[~df['precip'].isin([-9999, 0, -1])] df['precip'] = df['precip'] / 10 # 单位转换 # 日期解析 df['date'] = pd.to_datetime(df[['year', 'month', 'day']]) # 按日期分组处理 daily = df.groupby('date')['precip'].agg(['count', 'sum']) # 应用计算规则 def calc_daily(row): if row['count'] == 1: return row['sum'] elif row['count'] <= 3: # 获取3、9、15、21时的数据 sub = df[(df['date'] == row.name) & (df['hour'].isin([3, 9, 15, 21]))] return sub['precip'].sum() else: # 排除3、9、15、21时数据后求和 sub = df[(df['date'] == row.name) & (~df['hour'].isin([3, 9, 15, 21]))] return sub['precip'].sum() daily['daily_precip'] = daily.apply(calc_daily, axis=1) result.append(daily.reset_index()) return pd.concat(result) # 使用示例 daily_precip = process_precipitation('path_to_csv_files') daily_precip.to_csv('daily_precipitation.csv', index=False)

4. 数据验证与交叉比对

为确保数据可靠性，建议采用以下验证方法：

4.1 多源数据对比

选择2-3个数据源对关键指标进行交叉验证：

验证指标	可接受偏差范围	常见问题
日最高气温	±1.5°C	时次选择差异
日降水量	±20%	观测方法不同
连续干旱日数	±1天	缺测处理方式

4.2 统计一致性检查

def validate_data(ground_truth, test_data): # 合并数据集 merged = pd.merge( ground_truth, test_data, on=['station_id', 'date'], suffixes=('_ref', '_test') ) # 计算差异 merged['temp_diff'] = merged['temp_ref'] - merged['temp_test'] merged['precip_ratio'] = merged['precip_test'] / merged['precip_ref'] # 分析结果 validation_report = { 'temperature': { 'MAE': merged['temp_diff'].abs().mean(), 'correlation': merged[['temp_ref', 'temp_test']].corr().iloc[0,1] }, 'precipitation': { 'mean_ratio': merged['precip_ratio'].mean(), 'nonzero_match': (merged['precip_ref'] > 0) & (merged['precip_test'] > 0) } } return validation_report

4.3 可视化验证

使用matplotlib或Plotly绘制时间序列对比图，直观检查：

• 季节变化模式是否一致
• 极端事件是否同步记录
• 长期趋势是否相似

5. 高效管理气象数据的技巧

经过多个项目实践，总结出以下提高工作效率的方法：

1. 数据组织规范：

   • 按地区/年份/数据类型分级存储
   • 使用标准化的文件名（如{站号}_{要素}_{起止日期}.csv）
   • 每个数据集附带README说明数据来源和处理记录

2. 元数据管理：

   • 记录各站点的经纬度、海拔高度
   • 标注数据缺失时段和质量问题
   • 保存原始下载文件和清洗后版本

3. 自动化工作流：

   • 使用Airflow或Prefect调度定期数据更新
   • 将常用处理步骤封装为函数或类
   • 建立数据质量检查的自动化报告

class WeatherDataPipeline: def __init__(self, config): self.stations = config['stations'] self.data_dir = config['data_dir'] def run(self): self.download_data() self.process_data() self.generate_report() def download_data(self): # 实现下载逻辑 pass def process_data(self): # 实现处理逻辑 pass def generate_report(self): # 生成数据质量报告 pass # 配置和运行 config = { 'stations': ['545110', '545270'], 'data_dir': 'weather_data', 'years': [2020, 2021, 2022] } pipeline = WeatherDataPipeline(config) pipeline.run()

在处理青藏高原地区的气象数据时，发现站点稀疏地区的格点数据产品往往比站点观测更具代表性。这提醒我们，选择数据源时需要考虑研究区域的特点，有时再分析数据可能比实测数据更适合特定研究需求。