data-cleaning

数据清洗

此技能可帮助 AI Agent 系统地清理和预处理原始数据集，使其达到可供分析的状态。Agent 会处理缺失值、重复记录、数据类型不匹配、格式不一致、异常值和归一化问题。它还可以强制执行验证模式，确保持续的数据质量。主要工具链是 pandas，并借助 pyjanitor 和 great_expectations 执行高级验证。

工作流

1. 摄取原始数据并进行数据概况分析。 加载数据集后，立即生成质量报告：统计每列的空值数量、识别重复行、检查数据类型是否符合预期模式，并标记包含混合类型的列。此概况将指导后续所有清洗决策。

2. 处理缺失值。 根据数据类型和缺失模式，为每列应用相应策略。缺失值少于 5% 的数值列使用中位数填充。分类列使用众数或专门的“未知”类别。对于缺失值超过 40% 的列，将其标记为可能需要删除，并在删除前咨询用户。

3. 删除重复项并解决冲突。 识别完全重复项和近似重复项（例如，仅空格或大小写不同的行）。对于完全重复项，保留首次出现的记录。对于近似重复项，使用可配置的相似度阈值进行模糊匹配，并根据时间新近程度或完整性合并冲突值。

4. 修正数据类型并统一格式。 将列转换为预期类型——将日期字符串解析为 datetime 对象、将数字字符串转换为浮点数，并将分类值规范为标准形式。统一电话号码、邮政编码和货币表示等格式。

5. 检测并处理异常值。 对称分布使用 IQR 方法（1.5x），正态分布数据使用 z 分数。提供三种处理选项：将值限制在边界值（缩尾处理）、替换为空值以便后续填充，或仅标记模式——添加标注但保留原始值。

6. 验证清洗后的输出。 根据验证规则检查清洗后的数据集，包括非空约束、范围检查、唯一性约束和引用完整性。报告所有剩余违规项，并保存清洗后的数据集以及清洗日志，记录应用的每项转换。

支持的技术

• pandas — 核心数据处理和类型转换
• pyjanitor — 为清洗操作提供便捷的方法链
• great_expectations — 模式验证和数据质量检查
• fuzzywuzzy — 用于检测近似重复项的模糊字符串匹配
• numpy — 用于异常值检测的数值运算

用法

向 Agent 提供原始数据集的文件路径，也可以选择提供模式定义，指定预期列类型、有效范围和唯一性约束。Agent 将生成清洗后的文件和转换日志。

示例

示例 1：使用 pandas 清理杂乱的 CSV

import pandas as pd
import numpy as np

# Load raw data
df = pd.read_csv("messy_orders.csv")
print(f"Raw shape: {df.shape}")  # (2340, 8)
print(df.isnull().sum())
# order_id         0
# customer_name   12
# email           45
# order_date      18
# amount          23
# status           0
# region          67
# discount         0

# 1. Fix data types — order_date has mixed formats
df["order_date"] = pd.to_datetime(df["order_date"], format="mixed", dayfirst=False)
df["amount"] = pd.to_numeric(df["amount"], errors="coerce")

# 2. Handle missing values
df["customer_name"] = df["customer_name"].fillna("Unknown")
df["email"] = df["email"].fillna("[email protected]")
df["amount"] = df["amount"].fillna(df["amount"].median())
df["region"] = df["region"].fillna(df["region"].mode()[0])
df["order_date"] = df["order_date"].fillna(method="ffill")

# 3. Remove duplicates
before = len(df)
df = df.drop_duplicates(subset=["order_id"], keep="first")
print(f"Removed {before - len(df)} duplicate orders")  # Removed 34 duplicate orders

# 4. Standardize categorical values
df["status"] = df["status"].str.strip().str.lower().replace({
    "shipped": "shipped", "ship": "shipped",
    "cancelled": "cancelled", "canceled": "cancelled",
    "pending": "pending", "pend": "pending"
})
df["region"] = df["region"].str.strip().str.title()

# 5. Outlier treatment — cap amounts at IQR bounds
Q1 = df["amount"].quantile(0.25)
Q3 = df["amount"].quantile(0.75)
IQR = Q3 - Q1
lower, upper = Q1 - 1.5 * IQR, Q3 + 1.5 * IQR
df["amount"] = df["amount"].clip(lower=lower, upper=upper)

print(f"Clean shape: {df.shape}")  # (2306, 8)
df.to_csv("clean_orders.csv", index=False)

示例 2：通过模式强制执行的数据验证流水线

import great_expectations as gx

context = gx.get_context()

# Define a validation suite
suite = context.add_expectation_suite("orders_validation")

# Add expectations
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="amount", min_value=0.01, max_value=50000.00
    )
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeInSet(
        column="status", value_set=["pending", "shipped", "delivered", "cancelled"]
    )
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="order_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToMatchRegex(
        column="email", regex=r"^[^@]+@[^@]+\.[^@]+$"
    )
)

# Run validation against cleaned data
results = context.run_validation(suite, batch=gx.read_csv("clean_orders.csv"))

print(f"Success: {results.success}")
print(f"Passed: {results.statistics['successful_expectations']}/{results.statistics['evaluated_expectations']}")
# Success: True
# Passed: 5/5

最佳实践

• 在进行任何转换前，始终保存一份原始数据副本——清洗应当可复现，而非破坏性操作。
• 记录每项转换及其数量（例如，“使用中位数 312.45 填充了 amount 列中的 23 个空值”），以建立可审计的清洗记录。
• 优先采用结合领域知识的填充方法，而不是机械地使用默认值；如果某列的缺失模式并非随机（MNAR），请咨询用户。
• 尽早并频繁地进行验证——每个清洗阶段后都运行模式检查，而不只是最后检查。
• 将清洗视为迭代过程：第一轮会发现明显问题，但后续分析经常会暴露新问题。
• 对 pd.to_numeric 和 pd.to_datetime 使用 errors="coerce"，将转换失败显示为 NaN，而不是导致程序崩溃。

边缘情况

• 完全为空的列。 如果某列的值 100% 都是 null，则自动删除并记录警告，而不是尝试对没有信息的列进行填补。
• 重复的列名。 Pandas 允许重复的列名而不会发出提示。加载时检测重复列名，并在执行任何操作前添加后缀（_1、_2）进行重命名。
• 编码问题。 如果 read_csv 引发 UnicodeDecodeError，则依次尝试使用 encoding="latin-1" 和 encoding="cp1252" 重新读取，并记录成功使用的编码。
• 包含多种格式的日期列。 如果同一列中包含 "2024-01-15"、"01/15/2024" 和 "Jan 15, 2024"，则使用 pd.to_datetime(col, format="mixed")，并通过抽样检查验证解析结果。
• 以字符串存储且带有货币符号的数值列。 类型转换前，先去除 $、€、逗号和空白：df["price"].str.replace(r"[$€,\s]", "", regex=True).astype(float)。