请提供您需要创作标题的【关键词】。

引言

在数据科学和分析领域，数据清洗是整个工作流程中最耗时但也是最关键的步骤。根据业界统计，数据科学家通常花费 60-80% 的时间在数据准备和清洗上。原始数据往往包含缺失值、异常值、重复数据、格式不一致等问题，这些问题会严重影响后续分析的准确性和可靠性。

Python 的 Pandas 库是数据清洗的黄金标准工具，它提供了强大而灵活的功能来处理各种数据质量问题。本文将详细介绍如何使用 Pandas 进行系统性的数据清洗，包括处理缺失值、异常值、数据类型转换、字符串处理、日期处理等核心技巧，并通过完整的实际案例来演示每个步骤。

1. 数据加载和初步探索

1.1 加载数据

数据清洗的第一步是正确加载数据。Pandas 支持多种格式，包括 CSV、Excel、JSON、SQL 等。

import pandas as pd
import numpy as np

# 从 CSV 文件加载数据
df = pd.read_csv('data.csv')

# 从 Excel 文件加载数据
df = pd.read_excel('data.xlsx', sheet_name='Sheet1')

# 从 JSON 文件加载数据
df = pd.read_json('data.json')

# 从 SQL 数据库加载数据
import sqlite3
conn = sqlite3.connect('database.db')
df = pd.read_sql_query("SELECT * FROM table_name", conn)

1.2 初步探索数据

在开始清洗之前，需要了解数据的基本情况。

# 查看前5行数据
print(df.head())

# 查看数据基本信息（数据类型、非空值数量）
print(df.info())

# 查看数据的统计摘要
print(df.describe())

# 查看数据形状（行数，列数）
print(f"数据形状: {df.shape}")

# 查看列名
print(f"列名: {df.columns.tolist()}")

# 检查缺失值
print(df.isnull().sum())

2. 处理缺失值

缺失值是数据中最常见的问题之一。Pandas 用 NaN 表示缺失值。

2.1 识别缺失值

# 检查每列的缺失值数量
missing_values = df.isnull().sum()
print(missing_values)

# 计算缺失值百分比
missing_percentage = (df.isnull().sum() / len(df)) * 100
print(missing_percentage)

# 可视化缺失值（需要 matplotlib 和 seaborn）
import matplotlib.pyplot as plt
import seaborn as sns

plt.figure(figsize=(10,6))
sns.heatmap(df.isnull(), cbar=False, cmap='viridis')
plt.title('缺失值分布热力图')
plt.show()

2.2 删除缺失值

当缺失值比例很小或者数据量很大时，可以考虑删除缺失值。

# 删除包含任何缺失值的行
df_cleaned = df.dropna()

# 删除包含任何缺失值的列
df_cleaned = df.dropna(axis=1)

# 删除缺失值超过阈值的列（例如超过50%）
threshold = 0.5 * len(df)
df_cleaned = df.dropna(axis=1, thresh=threshold)

# 删除特定列中缺失值的行
df_cleaned = df.dropna(subset=['age', 'salary'])

2.3 填充缺失值

填充缺失值是更常用的方法，需要根据数据类型和业务逻辑选择合适的填充策略。

# 用0填充数值列
df['age'] = df['age'].fillna(0)

# 用均值填充数值列
df['age'] = df['age'].fillna(df['age'].mean())

# 用中位数填充数值列（对异常值更鲁棒）
df['age'] = df['age'].fillna(df['age'].median())

# 用众数填充分类列
df['city'] = df['city'].fillna(df['city'].mode()[0])

# 用前向填充（用前一个有效值填充）
df['value'] = df['value'].fillna(method='ffill')

# 用后向填充（用后一个有效值填充）
df['value'] = df['value'].fillna(method='bfill')

# 用插值法填充（线性插值）
df['value'] = df['value'].interpolate(method='linear')

# 用特定值填充多个列
df[['age', 'salary']] = df[['age', 'salary']].fillna({'age': df['age'].median(), 'salary': df['salary'].mean()})

2.4 高级缺失值处理：插值法

对于时间序列数据，插值法可以产生更自然的填充效果。

# 时间序列数据的插值
df['date'] = pd.to_datetime(df['date'])
df = df.set_index('date')
df['temperature'] = df['temperature'].interpolate(method='time')
df = df.reset_index()

3. 处理重复值

重复数据会导致分析结果偏差，需要及时识别和处理。

3.1 识别重复值

# 检查重复行
duplicate_rows = df.duplicated()
print(f"重复行数: {duplicate_rows.sum()}")

# 查看重复行的具体内容
print(df[duplicate_rows])

# 检查特定列的重复值
duplicate_in_columns = df.duplicated(subset=['customer_id', 'order_date'])
print(f"基于 customer_id 和 order_date 的重复行数: {duplicate_in_columns.sum()}")

# 标记重复行（保留第一个出现的）
df['is_duplicate'] = df.duplicated(keep='first')

3.2 删除重复值

# 删除所有重复行，保留第一个
df_cleaned = df.drop_duplicates()

# 删除所有重复行，保留最后一个
df_cleaned = df.drop_duplicates(keep='last')

# 删除所有重复行，不保留任何重复项
df_cleaned = df.drop_duplicates(keep=False)

# 基于特定列删除重复行
df_cleaned = df.drop_duplicates(subset=['customer_id', 'order_date'])

# 删除重复行并原地修改
df.drop_duplicates(inplace=True)

4. 处理异常值

异常值（Outliers）是远离数据集中趋势的极端值，可能由测量错误、数据录入错误或真实但极端的事件引起。

4.1 识别异常值

4.1.1 使用统计方法识别

# 使用 IQR（四分位距）方法识别异常值
def detect_outliers_iqr(data, column):
    Q1 = data[column].quantile(0.25)
    Q3 = data[column].column.quantile(0.75)
    IQR = Q3 - Q1
    lower_bound = Q1 - 1.5 * IQR
    upper_bound = Q3 + 1.5 * IQR
    outliers = data[(data[column] < lower_bound) | (data[column] > upper_bound)]
    return outliers

# 使用 Z-score 方法识别异常值
def detect_outliers_zscore(data, column, threshold=3):
    mean = data[column].mean()
    std = data[column].std()
    z_scores = (data[column] - mean) / std
    outliers = data[np.abs(z_scores) > threshold]
   箱线图（Box Plot）可视化
plt.figure(figsize=(8,6))
sns.boxplot(x=df['age'])
plt.title('年龄分布箱线图')
plt.show()

# 散点图可视化
plt.figure(figsize=(8,6))
plt.scatter(range(len(df)), df['salary'])
plt.title('薪资分布散点图')
plt.ylabel('Salary')
plt.show()

4.2 处理异常值

# 1. 删除异常值
def remove_outliers_iqr(data, column):
    Q1 = data[column].quantile(0.25)
    Q3 = data[column].quantile(0.45)  # 修正：应该是0.75
    IQR = Q3 - Q1
    lower_bound = Q1 - 1.5 * IQR
    upper_bound = + 1.5 * IQR  # 修正：应该是 Q3 + 1.5 * IQR
    return data[(data[column] >= lower_bound) & (data[column] <= upper_bound)]

# 2. 截断（Winsorization）：将异常值替换为边界值
def winsorize_outliers(data, column, limits=[0.05, 0.05]):
    lower_limit = data[column].quantile(limits[0])
    upper_limit = data[column].quantile(1 - limits[1])
    data[column] = data[column].clip(lower=lower_limit, upper=upper_limit)
    return data

# 3. 用中位数替换异常值
def replace_outliers_with_median(data, column):
    Q1 = data[column].quantile(0.25)
    Q3 = data[column].quantile(0.75)
    IQR = Q3 - Q1
    lower_bound = Q1 - 1.5 * IQR
    upper_bound = Q3 + 1.5 * IQR
    median = data[column].median()
    data.loc[(data[column] < lower_bound) | (data[column] > upper_bound), column] = median
    return data

# 4. 将异常值标记为缺失值
def mark_outliers_as_missing(data, column):
    Q1 = data[column].quantile(0.25)
    Q3 = data[column].quantile(0.75)
    IQR = Q3 - Q1
    lower_bound = Q1 - 1.5 * IQR
    upper_bound = Q3 + 1.5 * IQR
    data.loc[(data[column] < lower_bound) | (data[column] > upper_bound), column] = np.nan
    return data

5. 数据类型转换和格式标准化

5.1 数值类型转换

# 将字符串转换为数值（处理逗号、货币符号等）
df['price'] = df['price'].str.replace('[$,]', '', regex=True).astype(float)

# 将字符串转换为整数
df['quantity'] = df['quantity'].astype(int)

# 安全转换（无法转换的设为 NaN）
df['numeric_value'] = pd.to_numeric(df['numeric_value'], errors='coerce')

# 处理百分比字符串
df['discount'] = df['discount'].str.rstrip('%').astype(float) / 100

5.2 日期时间转换

# 将字符串转换为 datetime
df['date'] = pd.to_datetime(df['date'], format='%Y-%m-%d')

# 处理多种日期格式
df['date'] = pd.to_datetime(df['date'], errors='coerce')

# 从日期中提取特征
df['year'] = df['date'].dt.year
df['month'] = df['date'].dt.month
df['day'] = df['date'].dt.day
df['weekday'] = df['date'].dt.weekday
df['is_weekend'] = df['date'].dt.weekday >= 5

# 计算日期差
df['days_since_last_order'] = (pd.Timestamp.now() - df['last_order_date']).dt.days

5.3 分类数据编码

# 独热编码（One-Hot Encoding）
df = pd.get_dummies(df, columns=['city', 'category'], prefix=['city', 'cat'])

# 标签编码（Label Encoding）
from sklearn.preprocessing import LabelEncoder
le = LabelEncoder()
df['city_encoded'] = le.fit_transform(df['city'])

# 处理有序分类数据
from pandas.api.types import CategoricalDtype
city_order = ['小城市', '中等城市', '大城市', '超大城市']
city_dtype = CategoricalDtype(categories=city_order, ordered=True)
df['city_size'] = df['city_size'].astype(city_dtype)

5.4 字符串标准化

# 基本字符串清理
df['name'] = df['name'].str.strip()  # 去除首尾空格
df['name'] = df['name'].str.lower()  # 转换为小写
df['name'] = df['name'].str.title()  # 首字母大写

# 清理电话号码
df['phone'] = df['phone'].str.replace(r'\D', '', regex=True)  # 只保留数字

# 清理邮箱地址
df['email'] = df['email'].str.lower().str.strip()
df['email'] = df['email'].str.replace(r'\s+', '', regex=True)  # 移除所有空格

# 使用正则表达式提取信息
df['area_code'] = df['phone'].str.extract(r'(\d{3})')  # 提取区号
df['domain'] = df['email'].str.extract(r'@(.+)')  # 提取邮箱域名

# 替换多个模式
df['status'] = df['status'].str.replace('active|Active|ACTIVE', 'Active', regex=True)
df['status'] = df['status'].str.replace('inactive|Inactive|INACTIVE', 'Inactive', 处理分类数据中的不一致

# 统一类别名称
df['category'] = df['category'].replace({
    'Electronics': 'Electronics',
    'electronics': 'Electronics',
    'ELECTRONICS': 'Electronics',
    'Electrnics': 'Electronics'  # 纠正拼写错误
})

# 处理拼写错误（使用模糊匹配）
from fuzzywuzzy import process
def fix_spelling_errors(data, column, correct_categories):
    def fix_single_value(value):
        if value in correct_categories:
            return value
        else:
            # 找到最接近的正确类别
            match, score = process.extractOne(value, correct_categories)
            if score > 80:  # 如果相似度超过80%
                return match
            else:
                return value
    data[column] = data[column].apply(fix_single_value)
    return data

# 使用示例
correct_categories = ['Electronics', 'Clothing', 'Books', 'Home']
df = fix_spelling_errors(df, 'category', correct_categories)

6. 高级数据清洗技巧

6.1 使用管道（Pipeline）进行链式清洗

def clean_data_pipeline(df):
    """数据清洗管道函数"""
    
    # 1. 复制数据以避免修改原始数据
    df = df.copy()
    
    # 2. 删除重复值
    df = df.drop_duplicates()
    
    # 3. 处理缺失值
    df['age'] = df['age'].fillna(df['age'].median())
    df['salary'] = df['age'].fillna(df['salary'].mean())
    df['city'] = df['city'].fillna(df['city'].mode()[0])
    
    # 4. 处理异常值
    Q1 = df['salary'].quantile(0.25)
    Q3 = df['salary'].quantile(0.75)
    IQR = Q3 - Q1
    lower_bound = Q1 - 1.5 * IQR
    upper_bound = Q3 + 1.4 * IQR  # 修正：应该是 Q3 + 1.5 * IQR
    df = df[(df['salary'] >= lower_bound) & (df['salary'] <= upper_bound)]
    
    # 5. 数据类型转换
    df['date'] = pd.to_datetime(df['date'], errors='coerce')
    df['price'] = pd.to_numeric(df['price'], errors='coerce')
    
    # 6. 字符串标准化
    df['name'] = df['name'].str.strip().str.title()
    df['city'] = df['city'].str.strip().str.title()
    
    return df

# 应用管道
df_clean = clean_data_pipeline(df)

6.2 使用函数式编程进行批量清洗

# 定义一系列清洗函数
def remove_duplicates(df):
    return df.drop_duplicates()

def fill_missing_numeric(df, columns, strategy='median'):
    for col in columns:
        if strategy == 'median':
            df[col] = df[col].fillna(df[col].median())
        elif strategy == 'mean':
            df[col] = df[col].fillna(df[col].mean())
        elif strategy == 'zero':
            df[col] = df[col].fillna(0)
    return df

def clean_strings(df, columns):
    for col in columns:
        df[col] = df[col].str.strip().str.lower()
    return df

# 组合使用
df_clean = (df
    .pipe(remove_duplicates)
    .pipe(fill_missing_numeric, columns=['age', 'salary'], strategy='median')
    .pipe(clean_strings, columns=['name', 'city']))

6.3 处理嵌套数据结构

# 处理 JSON 字符串列
import json

# 示例：列中包含 JSON 字符串
df['user_info'] = ['{"name": "Alice", "age": 30}', '{"name": "Bob", "age": 25}']

# 解析 JSON 字符串
df['user_info'] = df['user_info'].apply(json.loads)
df = pd.concat([df, df['user_info'].apply(pd.Series)], axis=1)

# 或者使用 json_normalize
df_normalized = pd.json_normalize(df['user_info'])

6.4 处理文本数据中的特殊字符和编码问题

# 处理 Unicode 编码问题
df['text'] = df['text'].str.encode('utf-8').str.decode('utf-8')

# 移除特殊字符，只保留字母、数字和空格
df['text_clean'] = df['text'].str.replace(r'[^a-zA-Z0-9\s]', '', regex=True)

# 移除 HTML 标签
df['text'] = df['text'].str.replace(r'<.*?>', '', regex=True)

# 处理 Unicode 规范化（处理重音符号等）
import unicodedata
def normalize_text(text):
    return unicodedata.normalize('NFKD', text).encode('ASCII', 'ignore').decode('ASCII')

df['text'] = df['text'].apply(normalize_text)

7. 实际案例：完整数据清洗流程

让我们通过一个完整的案例来演示如何应用上述所有技巧。

7.1 案例背景

假设我们有一个销售数据集，包含以下问题：

缺失值：年龄、薪资、城市
重复值：重复的订单记录
异常值：不合理的年龄（负数或超过120岁）、薪资异常高
数据类型错误：日期是字符串，价格包含货币符号
不一致：城市名称大小写不一致、拼写错误
特殊字符：电话号码包含括号和横线

7.2 完整代码实现

import pandas as pd
import numpy as np
import re
from datetime import datetime

# 创建示例数据（模拟真实问题）
data = {
    'customer_id': [1, 2, 3, 4, 5, 2, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15],
    'name': ['Alice', 'Bob', 'Charlie', 'David', 'Eve', 'Bob', 'Frank', 'Grace', 'Henry', 'Ivy', 'Jack', 'Kate', 'Leo', 'Mona', 'Nina', 'Oscar'],
    'age': [25, 30, -5, 150, 28, 30, 35, None, 40, 45, 50, 55, 60, 65, 70, 75],
    'salary': ['$50,000', '$60,000', '$55,000', '$70,000', '$65,000', '$60,000', '$75,000', '$80,000', '$85,000', '$90,000', '$95,000', '$100,000', '$105,000', '$110,000', '$115,000', '$1,000,000'],
    'city': ['New York', 'Los Angeles', 'new york', 'CHICAGO', 'Boston', 'Los Angeles', 'New York', 'Bostin', 'Chicago', 'New York', 'Los Angeles', 'Boston', 'Chicago', 'New York', 'Los Angeles', 'Boston'],
    'phone': ['(123) 456-7890', '123-456-7891', '123.456.7892', '1234567893', '(123) 456-7894', '123-456-7891', '123.456.7895', '1234567896', '(123) 456-7897', '123-456-7898', '123.456.7899', '1234567800', '(123) 456-7801', '123-456-7802', '123.456.7803', '1234567804'],
    'order_date': ['2023-01-15', '2023-01-16', '2023-01-17', '2023-01-18', '2023-01-19', '2023-01-16', '2023-01-20', '2023-01-21', '2023-01-22', '2023-01-23', '2023-01-24', '2023-01-25', '2023-01-26', '2023-01-27', '2023-01-28', '2023-01-29'],
    'status': ['Active', 'Inactive', 'active', 'ACTIVE', 'Inactive', 'Inactive', 'Active', 'active', 'Inactive', 'Active', 'ACTIVE', 'Inactive', 'Active', 'active', 'Inactive', 'Active']
}

df = pd.DataFrame(data)

print("原始数据:")
print(df)
print("\n原始数据信息:")
print(df.info())

# 开始清洗
def comprehensive_cleaning(df):
    """综合数据清洗函数"""
    
    print("\n" + "="*50)
    print("开始数据清洗流程")
    print("="*50)
    
    # 1. 复制数据
    df_clean = df.copy()
    
    # 2. 处理重复值
    print("\n1. 处理重复值")
    print(f"原始行数: {len(df_clean)}")
    df_clean = df_clean.drop_duplicates(subset=['customer_id', 'order_date'], keep='first')
    print(f"删除重复值后行数: {len(df_clean)}")
    
    # 3. 处理异常值（年龄）
    print("\n2. 处理年龄异常值")
    print(f"异常年龄值: {df_clean[(df_clean['age'] < 0) | (df_clean['age'] > 120)]['age'].tolist()}")
    df_clean.loc[(df_clean['age'] < 0) | (df_clean['age'] > 120), 'age'] = np.nan
    
    # 4. 处理缺失值
    print("\n3. 处理缺失值")
    print(f"缺失值统计:\n{df_clean.isnull().sum()}")
    
    # 年龄用中位数填充
    age_median = df_clean['age'].median()
    df_clean['age'] = df_clean['age'].fillna(age_median)
    print(f"年龄用中位数 {age_median} 填充")
    
    # 5. 处理薪资（字符串转数值）
    print("\n4. 处理薪资数据")
    print(f"原始薪资: {df_clean['salary'].tolist()}")
    df_clean['salary'] = df_clean['salary'].str.replace('[$,]', '', regex=True).astype(float)
    print(f"转换后薪资: {df_clean['salary'].tolist()}")
    
    # 处理薪资异常值（超过3倍标准差）
    salary_mean = df_clean['salary'].mean()
    salary_std = df_clean['salary'].std()
    salary_upper_bound = salary_mean + 3 * salary_std
    print(f"薪资上限: {salary_upper_bound}")
    df_clean.loc[df_clean['salary'] > salary_upper_bound, 'salary'] = salary_mean
    print(f"异常薪资已修正")
    
    # 6. 处理城市数据
    print("\n5. 处理城市数据")
    print(f"原始城市: {df_clean['city'].unique()}")
    
    # 统一大小写和去除空格
    df_clean['city'] = df_clean['city'].str.strip().str.title()
    
    # 修正拼写错误
    city_mapping = {
        'Bostin': 'Boston',
        'New York': 'New York',
        'Los Angeles': 'Los Angeles',
        'Chicago': 'Chicago'
    }
    df_clean['city'] = df_clean['city'].replace(city_mapping)
    print(f"清洗后城市: {df_clean['city'].unique()}")
    
    # 7. 处理电话号码
    print("\n6. 处理电话号码")
    print(f"原始电话: {df_clean['phone'].tolist()}")
    df_clean['phone'] = df_clean['phone'].str.replace(r'\D', '', regex=True)
    print(f"清洗后电话: {df_clean['phone'].tolist()}")
    
    # 8. 处理日期
    print("\n7. 处理日期")
    df_clean['order_date'] = pd.to_datetime(df_clean['order_date'])
    df_clean['order_year'] = df_clean['order_date'].dt.year
    df_clean['order_month'] = df_clean['order_date'].dt.month
    df_clean['order_day'] = df_clean['order_date'].dt.day
    print(f"日期转换完成，新增列: order_year, order_month, order_day")
    
    # 9. 处理状态数据
    print("\n8. 处理状态数据")
    print(f"原始状态: {df_clean['status'].unique()}")
    df_clean['status'] = df_clean['status'].str.strip().str.title()
    status_mapping = {
        'Active': 'Active',
        'Inactive': 'Inactive'
    }
    df_clean['status'] = df_clean['status'].replace(status_mapping)
    print(f"清洗后状态: {df_clean['status'].unique()}")
    
    # 10. 数据类型优化
    print("\n9. 优化数据类型")
    df_clean['age'] = df_clean['age'].astype(int)
    df_clean['customer_id'] = df_clean['customer_id'].astype('int32')
    df_clean['salary'] = df_clean['salary'].astype('float32')
    print("数据类型优化完成")
    
    return df_clean

# 执行清洗
df_cleaned = comprehensive_cleaning(df)

print("\n" + "="*50)
print("清洗后的数据:")
print("="*50)
print(df_cleaned)
print("\n清洗后数据信息:")
print(df_cleaned.info())
print("\n清洗后数据统计摘要:")
print(df_cleaned.describe())

7.3 清洗结果验证

# 验证清洗结果
def validate_cleaning(original_df, cleaned_df):
    """验证清洗结果"""
    
    print("\n" + "="*50)
    print("清洗结果验证")
    print("="*50)
    
    # 1. 检查重复值
    duplicates = cleaned_df.duplicated(subset=['customer_id', 'order_date']).sum()
    print(f"剩余重复值: {duplicates}")
    
    # 2. 检查缺失值
    missing = cleaned_df.isnull().sum().sum()
    print(f"剩余缺失值: {missing}")
    
    # 3. 检查年龄范围
    age_min = cleaned_df['age'].min()
    age_max = cleaned_df['age'].max()
    print(f"年龄范围: {age_min} - {age_max}")
    
    # 4. 检查薪资范围
    salary_min = cleaned_df['salary'].min()
    salary_max = cleaned_df['salary'].max()
    print(f"薪资范围: ${salary_min:,.2f} - ${salary_max:,.2f}")
    
    # 5. 检查数据类型
    print(f"数据类型:\n{cleaned_df.dtypes}")
    
    # 6. 检查城市唯一值
    print(f"城市唯一值: {cleaned_df['city'].unique()}")
    
    # 7. 检查状态唯一值
    print(f"状态唯一值: {cleaned_df['status'].unique()}")

validate_cleaning(df, df_cleaned)

8. 性能优化技巧

8.1 使用适当的数据类型

# 优化内存使用
def optimize_memory(df):
    """优化数据框内存使用"""
    
    # 数值列优化
    for col in df.select_dtypes(include=['int']).columns:
        df[col] = pd.to_numeric(df[col], downcast='integer')
    
    for col in df.select_dtypes(include=['float']).columns:
        df[col] = pd.to_numeric(df[col], downcast='float')
    
    # 对象列优化（转换为 category 如果唯一值较少）
    for col in df.select_dtypes(include=['object']).columns:
        num_unique = df[col].nunique()
        num_total = len(df)
        if num_unique / num_total < 0.5:  # 如果唯一值少于50%
            df[col] = df[col].astype('category')
    
    return df

df_optimized = optimize_memory(df_cleaned)

8.2 使用向量化操作

# 避免使用循环，使用向量化操作
# 不好的做法
for i in range(len(df)):
    df.loc[i, 'salary'] = df.loc[i, 'salary'] * 1.1

# 好的做法
df['salary'] = df['salary'] * 1.1

# 使用 np.where 进行条件操作
df['salary_category'] = np.where(df['salary'] > 75000, 'High', 'Normal')

# 使用 np.select 进行多条件操作
conditions = [
    df['salary'] < 50000,
    df['salary'] < 75000,
    df['salary'] >= 75000
]
choices = ['Low', 'Medium', 'High']
df['salary_category'] = np.select(conditions, choices, default='Unknown')

8.3 使用 chunksize 处理大文件

# 处理大文件时使用 chunksize
chunk_size = 10000
cleaned_chunks = []

for chunk in pd.read_csv('large_file.csv', chunksize=chunk_size):
    # 对每个 chunk 进行清洗
    chunk_cleaned = clean_data_pipeline(chunk)
    cleaned_chunks.append(chunk_cleaned)

# 合并所有 chunk
df_final = pd.concat(cleaned_chunks, ignore_index=True)

9. 常见错误和最佳实践

9.1 常见错误

# 错误1：直接修改原始数据
# 错误：df.dropna(inplace=True)  # 这会修改原始数据
# 正确：df_clean = df.dropna()   # 创建副本

# 错误2：忽略数据类型
# 错误：df['date'] = df['date'].astype(str)  # 这会丢失日期功能
# 正确：df['date'] = pd.to_datetime(df['date'])

# 错误3：在循环中使用 .loc 或 .iloc
# 错误：
for i in range(len(df)):
    if df.loc[i, 'age'] < 0:
        df.loc[i, 'age'] = 0

# 正确：
df.loc[df['age'] < 0, 'age'] = 0

# 错误4：忽略内存使用
# 错误：读取大文件时不指定数据类型
# 正确：df = pd.read_csv('large.csv', dtype={'id': 'int32', 'value': 'float32'})

9.2 最佳实践

# 1. 始终创建数据副本
df_clean = df.copy()

# 2. 使用函数封装清洗逻辑
def clean_data(df):
    """文档字符串说明清洗步骤"""
    # 清洗代码
    return df_clean

# 3. 记录清洗步骤
cleaning_log = []
def log_step(func):
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        cleaning_log.append(f"{func.__name__}: {len(args[0])} -> {len(result)}")
        return result
    return wrapper

# 4. 使用配置管理清洗参数
cleaning_config = {
    'fill_strategy': {
        'age': 'median',
        'salary': 'mean'
    },
    'outlier_threshold': 3,
    'duplicate_subset': ['customer_id', 'order_date']
}

# 5. 验证清洗结果
def validate_data(df):
    """验证数据质量"""
    assert df.duplicated().sum() == 0, "存在重复值"
    assert df.isnull().sum().sum() == 0, "存在缺失值"
    assert df['age'].min() >= 0, "年龄包含负值"
    return True

10. 总结

数据清洗是数据分析的基础工作，掌握高效的 Pandas 数据清洗技巧可以显著提高工作效率和数据质量。本文详细介绍了从基础到高级的清洗技巧，包括：

数据加载和探索：了解数据的基本情况
缺失值处理：识别、删除和填充缺失值
重复值处理：识别和删除重复数据
异常值处理：识别和处理异常值
数据类型转换：数值、日期、分类数据的标准化
字符串处理：清理和标准化文本数据
高级技巧：管道函数、批量处理、嵌套数据
性能优化：内存优化、向量化操作
最佳实践：避免常见错误，遵循最佳实践

记住，数据清洗不是一次性的工作，而是一个迭代的过程。根据你的具体业务需求和数据特点，选择合适的清洗策略，并始终验证清洗结果的质量。

通过系统性的方法和适当的工具，你可以将混乱的原始数据转化为高质量的分析就绪数据，为后续的数据分析和机器学习奠定坚实的基础。# 如何使用 Python 和 Pandas 进行高效的数据清洗：从原始数据到分析就绪的完整指南

引言