Python计算两个CSV数据集同位置时间的气象参数差值

1. 前置校验：避免文件错配 #

• 先固定两个数据集的根目录路径，处理过程中不要改动文件存放位置
• 给每个CSV生成唯一匹配键：如果文件名本身包含位置、时间信息，直接用正则提取这两个字段拼接成键；如果文件名是随机命名，就先读每个CSV的表头行，拿表内记录的位置、时间字段当键
• 把两个数据集的匹配键分别存成集合，筛出两边都存在的键组成待处理文件对，两边独有的文件直接记到日志里后续人工核对，这步不能省，几千上万个文件场景下，很容易出现个别文件命名错、漏传、损坏的情况，不做校验直接按文件名排序配对大概率出错

2. 单文件对计算逻辑 #

• 用pandas读CSV，提前确认好文件编码、分隔符，避免读出来乱码
• 读入后先做行对齐校验：因为所有参数是随海拔变化的，必须保证两个文件的海拔列数值、顺序完全一致，对不上的直接记异常跳过，不要硬算
• 区分维度列和参数列：位置、时间、海拔属于维度信息，不参与差值计算；剩下的温度、气压、湿度这类测量列，直接做列级减法就行，pandas的列运算比逐行循环快1-2个数量级，大文件也能扛住
• 输出结果保留所有维度列，每个参数的差值列命名统一加_diff后缀，比如温度差值列叫temperature_diff，方便后续分析

3. 批量处理性能优化 #

• 不要单进程串行跑，用Python标准库的concurrent.futures.ProcessPoolExecutor开多进程，进程数设为CPU核心数减1就行，不会把系统资源占满，处理速度能提升3-6倍
• 如果单个CSV体积超过100MB，读的时候加chunksize参数分块加载，每算完一块就写入结果文件，避免内存溢出
• 输出目录的结构尽量和原数据集保持一致，后续找文件不用反复翻

4. 异常兜底 #

• 所有单文件处理逻辑外层加异常捕获，出错的文件路径、错误原因统一写到日志文件里，不要因为单个文件损坏导致整个批量任务中断
• 每处理一定数量的文件就打印一次进度，方便随时掌握处理情况，不用傻等

注意：以下示例代码里的正则规则、列名需要根据你自己的文件实际格式调整，直接复制运行大概率不符合你的数据格式

import os
import re
import pandas as pd
from concurrent.futures import ProcessPoolExecutor
import logging

# 日志配置，出错信息全存在这里
logging.basicConfig(
    filename='process_error.log',
    level=logging.WARNING,
    format='%(asctime)s - %(message)s',
    encoding='utf-8'
)

# 路径配置，改成自己的实际路径
DATASET_A_ROOT = "./dataset1"
DATASET_B_ROOT = "./dataset2"
OUTPUT_ROOT = "./diff_result"
# 维度列列表，这些列不参与差值计算，根据自己的表结构改
DIMENSION_COLS = ["location", "record_time", "altitude"]

def gen_file_match_key(file_path):
    """从文件名提取位置+时间生成匹配键，正则规则按自己的文件名格式改"""
    file_name = os.path.basename(file_path)
    # 示例正则适配：位置_时间_其他信息.csv 格式，比如 海淀_2024060112_100m.csv
    pattern = r"([\u4e00-\u9fa5a-zA-Z0-9]+)_(\d{12})_.*\.csv$"
    match_obj = re.match(pattern, file_name)
    if not match_obj:
        return None
    location, record_time = match_obj.groups()
    return f"{location}_{record_time}"

def single_pair_process(file_pair_info):
    """处理单个匹配的文件对"""
    a_file_path, b_file_path = file_pair_info
    try:
        # 读两个文件
        df_a = pd.read_csv(a_file_path)
        df_b = pd.read_csv(b_file_path)
        # 校验海拔序列完全对齐
        if not df_a['altitude'].equals(df_b['altitude']):
            logging.warning(f"海拔序列不匹配，跳过：{a_file_path} | {b_file_path}")
            return
        # 构建结果表
        df_result = df_a[DIMENSION_COLS].copy()
        # 遍历所有参数列算差值
        param_cols = [col for col in df_a.columns if col not in DIMENSION_COLS]
        for col in param_cols:
            df_result[f"{col}_diff"] = df_a[col] - df_b[col]
        # 写入结果，保持原文件名
        output_file_path = os.path.join(OUTPUT_ROOT, os.path.basename(a_file_path))
        df_result.to_csv(output_file_path, index=False, encoding='utf-8')
    except Exception as e:
        logging.warning(f"文件处理失败：{a_file_path} | {b_file_path}，错误信息：{str(e)}")

if __name__ == "__main__":
    os.makedirs(OUTPUT_ROOT, exist_ok=True)
    # 收集A数据集所有csv的匹配键和路径
    a_file_map = {}
    for cur_root, _, cur_files in os.walk(DATASET_A_ROOT):
        for single_file in cur_files:
            if single_file.endswith(".csv"):
                full_path = os.path.join(cur_root, single_file)
                match_key = gen_file_match_key(full_path)
                if match_key:
                    a_file_map[match_key] = full_path
    # 收集B数据集所有csv的匹配键和路径
    b_file_map = {}
    for cur_root, _, cur_files in os.walk(DATASET_B_ROOT):
        for single_file in cur_files:
            if single_file.endswith(".csv"):
                full_path = os.path.join(cur_root, single_file)
                match_key = gen_file_match_key(full_path)
                if match_key:
                    b_file_map[match_key] = full_path
    # 筛选两边都存在的匹配键，组成文件对
    valid_keys = a_file_map.keys() & b_file_map.keys()
    process_pairs = [(a_file_map[k], b_file_map[k]) for k in valid_keys]
    print(f"共找到匹配文件对{len(process_pairs)}组，开始处理")
    # 多进程批量处理
    with ProcessPoolExecutor(max_workers=os.cpu_count() - 1) as executor:
        list(executor.map(single_pair_process, process_pairs))
    print("处理完成，异常信息请查看process_error.log")

内容的提问来源于stack exchange，提问作者mohamed hatem