CSV数据清洗全攻略：10个Python Pandas技巧处理脏数据

import pandas as pd
import numpy as np

df = pd.read_csv('users.csv')
print(df.head())
print(df.info())
print(df.isnull().sum())

# 删除完全重复的行
df = df.drop_duplicates()

# 如果只想基于某些列去重（比如user_id），保留第一条
df = df.drop_duplicates(subset=['user_id'], keep='first')

print(f"去重前：{len(df_before)} 行，去重后：{len(df)} 行")

# 策略1：删除含缺失值的行（数据量大的时候用）
df = df.dropna()

# 策略2：用均值/中位数填充数值列
df['age'].fillna(df['age'].median(), inplace=True)

# 策略3：用众数填充分类列
df['city'].fillna(df['city'].mode()[0], inplace=True)

# 策略4：给缺失值打标签（适合缺失有含义的场景）
df['email'].fillna('no_email', inplace=True)

# 列名全部小写，空格替换为下划线
df.columns = df.columns.str.lower().str.replace(' ', '_')
df.columns = df.columns.str.replace('-', '_')

# 如果需要，还可以用映射表自定义列名
rename_map = {'user_name': 'username', 'signup_date': 'register_date'}
df.rename(columns=rename_map, inplace=True)

# 把多种日期格式统一为 YYYY-MM-DD
df['register_date'] = pd.to_datetime(
    df['register_date'],
    errors='coerce',  # 无法解析的转为 NaT
    infer_datetime_format=True
)

# 检查有没有解析失败的日期
bad_dates = df[df['register_date'].isna()]
print(f"有 {len(bad_dates)} 行日期格式无法解析")

# 提取日期中的年月
df['year'] = df['register_date'].dt.year
df['month'] = df['register_date'].dt.month

# 状态列统一为大写
df['status'] = df['status'].str.upper()

# 用户名字段首字母大写
df['username'] = df['username'].str.title()

# 去掉前后空格（非常常见的问题）
for col in ['username', 'email', 'city']:
    if col in df.columns:
        df[col] = df[col].astype(str).str.strip()

# 金额列转数值，去掉千分位逗号和货币符号
df['amount'] = df['amount'].astype(str).str.replace('[$,]', '', regex=True)
df['amount'] = pd.to_numeric(df['amount'], errors='coerce')

# 手机号验证——只保留11位数字
df['phone'] = df['phone'].astype(str).str.extract(r'(\d{11})')

# 布尔列转换
df['is_vip'] = df['is_vip'].map({'是': True, '否': False, 'Yes': True, 'No': False})

# 用 IQR（四分位距）法检测数值异常值
Q1 = df['age'].quantile(0.25)
Q3 = df['age'].quantile(0.75)
IQR = Q3 - Q1
lower = Q1 - 1.5 * IQR
upper = Q3 + 1.5 * IQR

# 标记并移除异常值
df = df[(df['age'] >= lower) & (df['age'] <= upper)]

# 另一种方法：直接用分位数截断
df['age'] = df['age'].clip(df['age'].quantile(0.01), df['age'].quantile(0.99))

# 拆分：把"张三/男/25岁"拆成三列
df[['name', 'gender', 'age_str']] = df['info'].str.split('/', expand=True)

# 合并：把分开的年月日合并成一个日期列
df['full_date'] = df['year'].astype(str) + '-' + df['month'].astype(str).str.zfill(2) + '-01'
df['full_date'] = pd.to_datetime(df['full_date'])

# 合并文本列
df['full_name'] = df['first_name'] + ' ' + df['last_name']

# 城市名称可能有多种写法
city_mapping = {
    '北京': '北京',
    '北京市': '北京',
    '上海': '上海',
    '上海市': '上海',
    'shanghai': '上海',
    'SZ': '深圳',
    '深圳': '深圳',
}
df['city'] = df['city'].map(city_mapping).fillna(df['city'])

# 或者用模糊匹配处理（适合写法很多的情况）
from fuzzywuzzy import fuzz
def fuzzy_match(value, mapping, threshold=80):
    if value in mapping:
        return mapping[value]
    best_match = max(mapping.keys(), key=lambda k: fuzz.ratio(value.lower(), k.lower()))
    if fuzz.ratio(value.lower(), best_match.lower()) > threshold:
        return mapping[best_match]
    return value

df['city'] = df['city'].apply(fuzzy_match, mapping=city_mapping)

# 保存为新的CSV，加上时间戳
from datetime import datetime
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
df.to_csv(f'users_cleaned_{timestamp}.csv', index=False, encoding='utf-8-sig')

# 保存为Parquet格式（推荐，文件更小、读取更快）
df.to_parquet(f'users_cleaned_{timestamp}.parquet', index=False)

# 备份原始数据
df_original = pd.read_csv('users.csv')
df_original.to_csv(f'users_original_backup_{timestamp}.csv', index=False)

import pandas as pd
import numpy as np

# 1. 读取数据
df = pd.read_csv('data.csv')
print(f"原始数据：{df.shape}")

# 2. 去重
df = df.drop_duplicates()
print(f"去重后：{df.shape}")

# 3. 列名标准化
df.columns = df.columns.str.lower().str.replace(' ', '_')

# 4. 处理缺失值
df = df.dropna(subset=['user_id'])  # 核心字段不能为空
df['email'].fillna('unknown@example.com', inplace=True)

# 5. 日期统一
df['register_date'] = pd.to_datetime(df['register_date'], errors='coerce')

# 6. 文本统一
for col in ['username', 'city']:
    if col in df.columns:
        df[col] = df[col].astype(str).str.strip()

# 7. 异常值过滤
df = df[(df['age'] > 0) & (df['age'] < 120)]

# 8. 保存
df.to_csv('data_cleaned.csv', index=False, encoding='utf-8-sig')
print("清洗完成！")

chunk_size = 100000
for chunk in pd.read_csv('big_data.csv', chunksize=chunk_size):
    chunk = chunk.drop_duplicates()
    # 继续处理...