2. 多站点监测数据合并与站点元数据关联

任务拆解：从“多文件”到“一张可分析的长表”

本节要解决的核心问题是：多个站点各自导出的监测 CSV 文件，字段结构一致但站点信息只体现在文件名或某一列中；同时另有一张站点元数据表，记录站点编码、名称、经纬度、海拔。目标是先把多文件纵向拼接成统一长表，再按站点编码左连接元数据，最后验证两件事：每个站点的记录数是否与源文件一致，以及元数据中站点编码是否唯一（否则连接会放大行数）。

准备示例数据与读取

为便于复现，先构造三个站点文件与一张元数据表。实际项目中文件来自不同站点导出，这里用字典模拟“文件名 → DataFrame”的读取结果，字段与真实监测数据一致。

import pandas as pd
import numpy as np

# 模拟三个站点源文件（实际中由 pd.read_csv 逐个读入）
def make_station(code, n, base):
    rng = pd.date_range("2024-01-01", periods=n, freq="h")
    return pd.DataFrame({
        "station_code": code,
        "time": rng,
        "pm25": np.round(base + np.random.randn(n) * 5, 1),
        "pm10": np.round(base * 1.6 + np.random.randn(n) * 8, 1),
        "no2": np.round(base * 0.5 + np.random.randn(n) * 3, 1),
    })

files = {
    "ST001_2024.csv": make_station("ST001", 24, 35),
    "ST002_2024.csv": make_station("ST002", 20, 42),
    "ST003_2024.csv": make_station("ST003", 18, 28),
}

# 站点元数据：station_code 为唯一键
meta_df = pd.DataFrame({
    "station_code": ["ST001", "ST002", "ST003"],
    "station_name": ["城东站", "城西站", "城北站"],
    "latitude": [30.12, 30.20, 30.31],
    "longitude": [120.34, 120.10, 120.22],
    "altitude": [18, 25, 12],
})

纵向合并：多文件拼成统一长表

合并时保留 station_code 列作为站点标识，并记录每个源文件的原始行数，供后续核对。使用 ignore_index=True 重建行索引，避免多文件索引重复。

source_counts = {}
frames = []
for fname, df in files.items():
    source_counts[df["station_code"].iloc[0]] = len(df)
    frames.append(df)

long_df = pd.concat(frames, ignore_index=True)
print("合并后总行数:", len(long_df))
print("各站源文件行数:", source_counts)

预期输出：合并后总行数为 24+20+18=62，各站源文件行数分别为 24、20、18。这一步只做纵向堆叠，不做任何去重或过滤，保证后续核对有基准。

元数据唯一性检查与关联

左连接前必须先确认 meta_df 中 station_code 唯一。若存在重复，merge 会产生笛卡尔放大，导致记录数虚增。检查通过后再执行左连接，保证长表每一行都能匹配到且仅匹配到一条元数据。

# 1) 唯一性检查
dup = meta_df["station_code"].duplicated().sum()
assert dup == 0, f"元数据站点编码存在 {dup} 条重复，需先清洗"

# 2) 左连接：以长表为主，保留全部监测记录
merged_df = long_df.merge(
    meta_df,
    on="station_code",
    how="left",
    validate="many_to_one",   # 显式声明多对一，重复键会直接报错
)

print("关联后行数:", len(merged_df))
print("未匹配元数据的行数:", merged_df["station_name"].isna().sum())

validate="many_to_one" 是关键防护：它要求右侧 meta_df 的键唯一，一旦元数据重复会立即抛出 MergeError，比事后比对行数更早暴露问题。

关联结果验证

验证分两步：先按站点统计关联后记录数，与源文件行数逐一比对；再确认没有未匹配行。两者都通过，才说明合并没有丢行、也没有被放大。

# 各站关联后记录数
merged_counts = merged_df.groupby("station_code").size().to_dict()
print("关联后各站记录数:", merged_counts)

# 与源文件行数比对
for code, n in source_counts.items():
    assert merged_counts[code] == n, f"{code} 记录数不一致: {merged_counts[code]} vs {n}"

# 未匹配检查
assert merged_df["station_name"].isna().sum() == 0, "存在未匹配元数据的站点"
print("验证通过：行数一致且元数据全部匹配")

可复用的核对清单