本文目录导读：
我来分享几个Python数据聚合的典型案例,涵盖不同场景和常用库。
Pandas数据聚合案例
案例1：销售数据汇总
import pandas as pd
import numpy as np
# 创建示例销售数据
sales_data = pd.DataFrame({
'日期': ['2024-01-01', '2024-01-01', '2024-01-02', '2024-01-02', '2024-01-03'],
'城市': ['北京', '上海', '北京', '上海', '广州'],
'产品': ['A', 'B', 'A', 'B', 'A'],
'销售额': [1000, 1500, 1200, 1800, 900],
'数量': [10, 15, 12, 18, 9]
})
# 按城市汇总销售额和数量
city_summary = sales_data.groupby('城市').agg({
'销售额': ['sum', 'mean', 'max'],
'数量': ['sum', 'mean']
}).round(2)
print("=== 按城市汇总 ===")
print(city_summary)
print("\n")
# 按日期和城市汇总
daily_city = sales_data.groupby(['日期', '城市']).agg({
'销售额': 'sum',
'数量': 'sum'
}).reset_index()
print("=== 按日期和城市汇总 ===")
print(daily_city)
案例2：多维度统计分析
# 创建更复杂的数据集
import pandas as pd
import numpy as np
np.random.seed(42)
df = pd.DataFrame({
'部门': np.random.choice(['销售部', '市场部', '技术部', '人事部'], 100),
'级别': np.random.choice(['初级', '中级', '高级'], 100),
'薪资': np.random.randint(5000, 30000, 100),
'绩效分': np.random.randint(60, 100, 100),
'项目数': np.random.randint(1, 10, 100)
})
# 多维度聚合
summary = df.groupby(['部门', '级别']).agg({
'薪资': ['mean', 'std', 'min', 'max'],
'绩效分': ['mean', 'std'],
'项目数': ['sum', 'mean']
}).round(2)
print("=== 多维度统计分析 ===")
print(summary.head(10))
SQL风格聚合（pandasql）
from pandasql import sqldf
import pandas as pd
# 使用SQL语法进行聚合
sales_df = pd.DataFrame({
'region': ['华北', '华东', '华南', '华北', '华东'],
'product': ['A', 'A', 'B', 'B', 'A'],
'revenue': [100, 200, 150, 120, 180],
'quantity': [10, 20, 15, 12, 18]
})
query = """
SELECT
region,
COUNT(*) as order_count,
SUM(revenue) as total_revenue,
AVG(revenue) as avg_revenue,
SUM(quantity) as total_quantity,
MAX(revenue) as max_revenue
FROM sales_df
GROUP BY region
ORDER BY total_revenue DESC
"""
result = sqldf(query, locals())
print("=== SQL风格聚合 ===")
print(result)
实战案例：电商订单分析
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
# 生成模拟订单数据
np.random.seed(123)
dates = pd.date_range('2024-01-01', '2024-01-10', freq='H')
orders = pd.DataFrame({
'order_id': range(1, 1001),
'customer_id': np.random.randint(1000, 2000, 1000),
'category': np.random.choice(['电子产品', '服装', '食品', '日用品'], 1000),
'amount': np.random.uniform(10, 500, 1000).round(2),
'quantity': np.random.randint(1, 10, 1000),
'order_date': np.random.choice(dates, 1000)
})
# 1. 按类别汇总销售额
category_sales = orders.groupby('category').agg({
'amount': ['sum', 'mean', 'count'],
'quantity': 'sum'
}).round(2)
category_sales.columns = ['总销售额', '平均订单额', '订单数', '总数量']
print("=== 类别销售汇总 ===")
print(category_sales)
print("\n")
# 2. 按日期聚合（每日销售额趋势）
daily_sales = orders.groupby(orders['order_date'].dt.date).agg({
'amount': 'sum',
'order_id': 'count'
}).reset_index()
daily_sales.columns = ['日期', '日销售额', '订单数']
print("=== 日销售趋势（前10天）===")
print(daily_sales.head(10))
print("\n")
# 3. RFM分析（客户价值分析）
from datetime import datetime
current_date = max(orders['order_date'])
rfm = orders.groupby('customer_id').agg({
'order_date': lambda x: (current_date - x.max()).days, # 最近购买时间
'order_id': 'count', # 购买频率
'amount': 'sum' # 总消费金额
}).reset_index()
rfm.columns = ['客户ID', '最近购买天数', '购买次数', '总消费']
print("=== RFM客户分析（前10名）===")
print(rfm.head(10))
使用字典进行灵活聚合
# 灵活定义聚合函数
agg_functions = {
'销售额': ['sum', 'mean', 'std', 'count'],
'数量': ['sum', 'mean'],
'利润': lambda x: x.sum() * 0.3 # 自定义函数
}
# 添加利润列
sales_data['利润'] = sales_data['销售额'] * 0.3
# 使用自定义聚合
result = sales_data.groupby('城市').agg(agg_functions)
print("=== 灵活聚合 ===")
print(result)
实际业务统计案例
# 假设这是某公司的员工绩效数据
performance = pd.DataFrame({
'employee_id': range(1, 101),
'department': np.random.choice(['技术', '产品', '销售', '运营'], 100),
'level': np.random.choice(['P3', 'P4', 'P5', 'P6', 'P7'], 100),
'salary': np.random.randint(15000, 50000, 100),
'performance_score': np.random.randint(60, 100, 100),
'projects_completed': np.random.randint(1, 20, 100),
'satisfaction': np.random.uniform(3, 5, 100).round(2)
})
# 综合绩效分析
def performance_analysis(data):
"""综合绩效分析函数"""
# 1. 部门维度统计
dept_stats = data.groupby('department').agg({
'salary': ['mean', 'median', 'std'],
'performance_score': ['mean', 'min', 'max'],
'projects_completed': 'sum',
'satisfaction': 'mean'
}).round(2)
# 2. 等级维度统计
level_stats = data.groupby('level').agg({
'salary': 'mean',
'performance_score': 'mean',
'projects_completed': 'mean',
'employee_id': 'count'
}).round(2)
# 3. 绩效等级分布（自定义区间）
def performance_level(score):
if score >= 90: return 'A'
elif score >= 80: return 'B'
elif score >= 70: return 'C'
else: return 'D'
data['绩效等级'] = data['performance_score'].apply(performance_level)
grade_dist = data.groupby('绩效等级').agg({
'employee_id': 'count',
'salary': 'mean',
'projects_completed': 'mean'
}).round(2)
return dept_stats, level_stats, grade_dist
dept_stats, level_stats, grade_dist = performance_analysis(performance)
print("=== 部门绩效统计 ===")
print(dept_stats)
print("\n=== 等级绩效统计 ===")
print(level_stats)
print("\n=== 绩效等级分布 ===")
print(grade_dist)
可视化汇总
import matplotlib.pyplot as plt
import seaborn as sns
# 创建可视化数据汇总
fig, axes = plt.subplots(2, 2, figsize=(15, 10))
# 1. 部门薪资分布
dept_salary = performance.groupby('department')['salary'].mean().sort_values()
axes[0, 0].barh(dept_salary.index, dept_salary.values)
axes[0, 0].set_title('部门平均薪资')
# 2. 等级分布
level_count = performance['level'].value_counts()
axes[0, 1].pie(level_count.values, labels=level_count.index, autopct='%1.1f%%')
axes[0, 1].set_title('员工等级分布')
# 3. 绩效与薪资关系
axes[1, 0].scatter(performance['performance_score'], performance['salary'])
axes[1, 0].set_xlabel('绩效分')
axes[1, 0].set_ylabel('薪资')
axes[1, 0].set_title('绩效-薪资关系')
plt.tight_layout()
plt.show()
最佳实践建议
-
选择合适的方法：
- 简单统计用
describe() - 分组汇总用
groupby().agg() - 复杂分析用
pivot_table()
- 简单统计用
-
性能优化：
- 使用
agg()代替多次apply() - 大数据集考虑使用
dask或polars
- 使用
-
数据质量：
- 聚合前先处理缺失值
- 使用
dropna()或fillna()
这些案例涵盖了日常工作中80%以上的数据聚合场景，可以根据实际需求灵活调整使用。
