Python案例如何用Pandas做数据分组替换

wen python案例 2026-07-14 192

本文目录导读：

基本概念准备
创建示例数据
分组替换的多种方法
实战案例：销售数据分析
高级技巧：滚动替换
性能优化技巧
完整实战：数据清洗案例

我来详细讲解如何使用Pandas进行数据分组替换,包含多个实用案例：

基本概念准备

import pandas as pd
import numpy as np

创建示例数据

# 创建示例数据集
df = pd.DataFrame({
    'department': ['销售部', '技术部', '销售部', '技术部', '市场部', '市场部'],
    'employee': ['张三', '李四', '王五', '赵六', '孙七', '周八'],
    'salary': [8000, 12000, 9500, 15000, 7000, 8500],
    'years': [3, 5, 2, 6, 1, 4]
})
print("原始数据：")
print(df)

分组替换的多种方法

方法1：使用groupby + transform

# 案例1：替换为组内平均值
print("=== 替换为各部门平均薪资 ===")
df['salary_avg_dept'] = df.groupby('department')['salary'].transform('mean')
print(df)
# 案例2：替换为组内排名
print("\n=== 替换为各部门薪资排名（从低到高）===")
df['salary_rank'] = df.groupby('department')['salary'].rank()
print(df)
# 案例3：标准化组内数据
print("\n=== 各部门薪资标准化（Z-score）===")
def z_score(x):
    return (x - x.mean()) / x.std()
df['salary_zscore'] = df.groupby('department')['salary'].transform(z_score)
print(df)

方法2：使用自定义替换规则

# 案例4：条件分组替换
print("=== 根据部门和工作年限调整薪资 ===")
def adjust_salary(group):
    if group.name == '销售部':
        group['salary'] = group['salary'] * 1.1  # 销售部涨薪10%
    elif group.name == '技术部':
        group['salary'] = group['salary'] * 1.2  # 技术部涨薪20%
    elif group.name == '市场部':
        # 市场部按年限调整
        group['salary'] = np.where(group['years'] >= 3, 
                                   group['salary'] * 1.15,  # 老员工涨15%
                                   group['salary'] * 1.05)  # 新员工涨5%
    return group
df_adjusted = df.groupby('department', group_keys=False).apply(adjust_salary)
print(df_adjusted[['department', 'employee', 'salary']])

方法3：使用map/dict映射

# 案例5：基于分组的映射替换
print("=== 基于部门的分级替换 ===")
# 创建替换映射
dept_level = {
    '销售部': 'A级',
    '技术部': 'A级',
    '市场部': 'B级'
}
df['dept_level'] = df['department'].map(dept_level)
print(df)
# 案例6：多条件分组替换
print("\n=== 复合条件替换 ===")
def classify_employee(row):
    if row['department'] == '技术部' and row['salary'] > 12000:
        return '高级技术'
    elif row['department'] == '销售部' and row['salary'] > 8000:
        return '高级销售'
    elif row['years'] >= 5:
        return '资深员工'
    else:
        return '普通员工'
df['employee_level'] = df.apply(classify_employee, axis=1)
print(df)

方法4：使用cut/qcut进行分组

# 案例7：按薪资分箱（等距）
print("=== 各部门薪资分层 ===")
for dept in df['department'].unique():
    dept_data = df[df['department'] == dept].copy()
    dept_data['salary_lvl'] = pd.cut(dept_data['salary'], 
                                      bins=3, 
                                      labels=['低', '中', '高'])
    print(f"\n{dept}部门:")
    print(dept_data[['employee', 'salary', 'salary_lvl']])
# 案例8：按百分位分组
print("\n=== 按百分位分组 ===")
def assign_percentile(group):
    group['salary_percentile'] = pd.qcut(group['salary'], 
                                         q=4, 
                                         labels=['Q1', 'Q2', 'Q3', 'Q4'])
    return group
df_percentile = df.groupby('department', group_keys=False).apply(assign_percentile)
print(df_percentile)

实战案例：销售数据分析

# 创建更实际的销售数据
np.random.seed(42)
sales_df = pd.DataFrame({
    'date': pd.date_range('2024-01-01', periods=100, freq='D'),
    'product': np.random.choice(['手机', '电脑', '平板'], 100),
    'region': np.random.choice(['华东', '华南', '华北'], 100),
    'sales': np.random.randint(10000, 50000, 100),
    'quantity': np.random.randint(10, 100, 100)
})
print("销售数据预览：")
print(sales_df.head())
# 案例9：按产品和地区分组替换异常值
print("\n=== 按组替换异常值 ===")
def remove_outliers(group):
    Q1 = group['sales'].quantile(0.25)
    Q3 = group['sales'].quantile(0.75)
    IQR = Q3 - Q1
    # 将异常值替换为中位数
    median = group['sales'].median()
    group['sales_cleaned'] = group['sales'].clip(Q1 - 1.5*IQR, Q3 + 1.5*IQR)
    group['sales_cleaned'] = np.where(
        (group['sales'] < Q1 - 1.5*IQR) | (group['sales'] > Q3 + 1.5*IQR),
        median, group['sales']
    )
    return group
sales_cleaned = sales_df.groupby(['product', 'region'], group_keys=False).apply(remove_outliers)
print("清洗前后对比：")
print(sales_cleaned[sales_cleaned['sales'] != sales_cleaned['sales_cleaned']])

高级技巧：滚动替换

# 案例10：按时间分组滚动计算
print("=== 按产品分组滚动平均 ===")
sales_df['rolling_avg_7d'] = (
    sales_df.groupby('product')['sales']
    .rolling(window=7, min_periods=1)
    .mean()
    .reset_index(level=0, drop=True)
)
# 案例11：分组累计替换
print("\n=== 各产品累计销售额 ===")
sales_df['cumulative_sales'] = (
    sales_df.groupby('product')['sales']
    .cumsum()
)
print(sales_df[sales_df['product'] == '手机'].head(10))

性能优化技巧

# 方法5：使用agg聚合后合并（大数据量时更高效）
print("=== 高效分组替换方式 ===")
# 计算统计值
dept_stats = df.groupby('department')['salary'].agg(['mean', 'median', 'std']).reset_index()
print("部门统计：")
print(dept_stats)
# 合并回原数据
df_efficient = df.merge(dept_stats, on='department', how='left')
print("\n合并后数据：")
print(df_efficient)
# 案例12：并行分组处理（大数据场景）
from multiprocessing import Pool
def process_group(group):
    # 处理每个组
    group['processed_salary'] = group['salary'] * 1.1
    return group
# 注意：实际使用时需要根据数据量选择是否并行

完整实战：数据清洗案例

# 案例13：完整的数据清洗流程
print("=== 数据清洗完整流程 ===")
# 创建带有缺失值和异常值的数据
dirty_data = pd.DataFrame({
    'category': ['A', 'A', 'A', 'B', 'B', 'B', 'C', 'C', 'C'],
    'value': [100, 200, 1000, 150, 175, 50, 300, 350, 25]  # 1000和25可能异常
})
print("原始脏数据：")
print(dirty_data)
# 完整清洗流程
def clean_data(df, group_col, value_col, threshold=1.5):
    def clean_group(group):
        Q1 = group[value_col].quantile(0.25)
        Q3 = group[value_col].quantile(0.75)
        IQR = Q3 - Q1
        lower_bound = Q1 - threshold * IQR
        upper_bound = Q3 + threshold * IQR
        # 标记异常值
        group['is_outlier'] = ((group[value_col] < lower_bound) | 
                              (group[value_col] > upper_bound))
        # 替换异常值为组内中位数
        median_val = group[value_col].median()
        group[value_col + '_cleaned'] = np.where(
            group['is_outlier'],
            median_val,
            group[value_col]
        )
        return group
    return df.groupby(group_col, group_keys=False).apply(clean_group)
# 执行清洗
cleaned_data = clean_data(dirty_data, 'category', 'value')
print("\n清洗后数据：")
print(cleaned_data)

主要方法：