📚 课程目录第4章数据清洗实战

4. 数据清洗实战

环境监测数据在采集、传输、入库的每个环节都可能引入“脏数据”：传感器掉线导致整段缺失、量程溢出产生离谱读数、设备重传造成重复记录、CSV 导入把数值列读成字符串。数据清洗的目标不是把数据“改好看”，而是让后续的统计、绘图和超标判定建立在可信的数据基础上。本节以一个典型的空气质量小时数据集为例，完整走一遍从原始表到清洗后数据集的流程。

一、先看清数据：结构与问题诊断

假设我们拿到一份 air_hourly.csv，字段包括站点编号 station、监测时间 ts、PM2.5 浓度 pm25、PM10 浓度 pm10、二氧化硫 so2、温度 temp。先用 pandas 读入并做一次“体检”：

import pandas as pd
import numpy as np

df = pd.read_csv("air_hourly.csv", dtype=str)
print(df.shape)
print(df.dtypes)
print(df.isna().sum())
print(df.duplicated(subset=["station", "ts"]).sum())
print(df["pm25"].head(10).tolist())

这里刻意用 dtype=str 全部按字符串读入，是为了先观察原始形态，避免 pandas 自动推断把 "--"、"缺测"、"1,2.5" 之类的内容悄悄变成 NaN 或报错。诊断阶段要回答四个问题：哪些列有缺失、缺失比例多少；哪些列存在非数值字符；是否存在同一站点同一时刻的重复行；数值分布是否出现明显越界。下面的 SVG 概括了本节清洗流程中各环节的输入、判断与输出关系。

二、统一缺失标记与类型转换

原始数据里缺失可能写成空串、"--"、"缺测"、"NA"。第一步把它们统一成 np.nan，再对数值列做类型转换。字符串清洗的常见动作包括去掉千分位逗号、去掉单位后缀（如 "μg/m³"）、去除首尾空白：

missing_tokens = ["", "--", "缺测", "NA", "N/A", "null"]
df = df.replace(missing_tokens, np.nan)

num_cols = ["pm25", "pm10", "so2", "temp"]
for col in num_cols:
    df[col] = (df[col]
               .str.replace(",", "", regex=False)
               .str.replace("μg/m³", "", regex=False)
               .str.strip())
    df[col] = pd.to_numeric(df[col], errors="coerce")

df["ts"] = pd.to_datetime(df["ts"], errors="coerce")
df["station"] = df["station"].str.strip().str.upper()

关键点是 errors="coerce"：无法解析的内容会变成 NaN 而不是抛异常，这样我们能在转换后统计“有多少条原本是非数值文本”，而不是让程序中途崩溃。ts 转换失败的行同样会变成 NaT，这类记录无法参与时间序列分析，应单独计数后剔除。

三、重复记录：先定义“重复”

环境监测中的重复通常指同一站点、同一监测时刻出现多条记录，可能来自设备重传或多次入库。判断重复的键应选业务主键，而不是整行完全相同：

dup_mask = df.duplicated(subset=["station", "ts"], keep=False)
print("重复行数：", dup_mask.sum())

# 方案A：保留首条
df = df.drop_duplicates(subset=["station", "ts"], keep="first")

# 方案B：同键多条取均值（需先剔除无效值）
# df = (df.groupby(["station", "ts"], as_index=False)[num_cols].mean())

两种方案各有适用场景：若重复来自完全相同的重传，保留首条即可；若同一时刻有多次有效观测（如不同仪器），取均值更合理。选择后要在文档中写明，保证结果可复现。

四、异常值：用量程和物理意义双重判定

“超出量程”的读数是最典型的异常值。例如 PM2.5 传感器量程上限为 1000 μg/m³，出现 65535 往往是寄存器溢出或通信错误；负值在浓度指标上物理上不成立。处理策略是先标记为无效（置 NaN），再决定插补或剔除，而不是直接删除整行——因为同一行的温度可能仍然有效。

valid_range = {
    "pm25": (0, 1000),
    "pm10": (0, 1500),
    "so2":  (0, 2000),
    "temp": (-40, 60),
}

for col, (low, high) in valid_range.items():
    out_of_range = (df[col] < low) | (df[col] > high)
    print(col, "越界条数：", out_of_range.sum())
    df.loc[out_of_range, col] = np.nan

除了量程，还可以用统计方法辅助发现异常，例如按站点计算 pm25 的 99.5% 分位数，或使用“超过均值 ± 3 倍标准差”的规则。但统计规则只能作为提示，最终是否判为异常要结合量程和业务常识，避免把真实的污染过程误删。

五、缺失值处理：插补还是剔除

缺失处理取决于缺失比例和用途。若某站点某指标缺失率超过 30%，通常不建议强行插补，应标记该时段数据不可用；短时缺口（如 1～3 小时）可用时间序列线性插值。pandas 中按站点分组插值可以避免跨站点“借值”：

df = df.sort_values(["station", "ts"])

# 统计缺失率
missing_rate = df[num_cols].isna().mean().round(3)
print(missing_rate)

# 按站点做时间插值，限制最多连续插补 3 个点
df[num_cols] = (df.groupby("station")[num_cols]
                  .apply(lambda g: g.interpolate(method="linear", limit=3))
                  .reset_index(level=0, drop=True))

# 仍缺失的（长缺口）保留 NaN，不伪造数据
df = df.dropna(subset=["ts"])

这里要强调一个原则：插补是“估计”，不是“观测”。清洗后的数据集应保留一列标记，说明哪些值是原始观测、哪些是插补值，便于后续分析时区分。可以用 df["pm25_imputed"] = df["pm25"].isna() 在插补前记录标记。

六、输出与质量校验

清洗完成后不要急着保存，先做一次校验，确认关键指标达到预期：

校验项	检查方法	示例期望结果
数值列类型	`df.dtypes`	pm25/pm10/so2/temp 为 float64
时间列类型	`df["ts"].dtype`	datetime64[ns]
重复记录	`df.duplicated(["station","ts"]).sum()`	0
越界值	按量程重新统计	0（已置 NaN 或剔除）
缺失率	`df[num_cols].isna().mean()`	各列低于设定阈值
时间连续性	相邻 ts 差值是否为 1 小时	缺口数量可解释

df.to_csv("air_clean.csv", index=False, encoding="utf-8-sig")
print(df.shape, df["ts"].min(), df["ts"].max())

使用 utf-8-sig 编码可以让 Excel 直接打开中文列名不乱码。最终交付的 air_clean.csv 应满足：时间列为 datetime、数值列为 float、无重复主键、越界值已处理、缺失情况有记录。至此，原始监测数据才真正变成可以进入统计与可视化环节的可信数据集。