本文目录导读：
- apply() 函数 - 最常用的应用函数
- applymap() 函数 - 对DataFrame的每个元素应用函数
- agg() / aggregate() 函数 - 聚合应用
- transform() 函数 - 转换数据
- 实际业务案例
- 性能优化技巧
- 自定义函数的高级应用
我来详细介绍Pandas中常用的数据应用函数,包括多个实用案例。
apply() 函数 - 最常用的应用函数
基本用法
import pandas as pd
import numpy as np
# 创建示例数据
df = pd.DataFrame({
'A': [1, 2, 3, 4, 5],
'B': [10, 20, 30, 40, 50],
'C': [100, 200, 300, 400, 500]
})
print("原始数据：")
print(df)
# 对单列应用函数
def double_value(x):
return x * 2
df['A_doubled'] = df['A'].apply(double_value)
print("\n对A列应用double函数：")
print(df)
# 使用lambda表达式
df['B_squared'] = df['B'].apply(lambda x: x ** 2)
print("\n对B列应用lambda函数：")
print(df)
# 对多列应用函数
df['A_plus_B'] = df.apply(lambda row: row['A'] + row['B'], axis=1)
print("\n计算A+B：")
print(df)
applymap() 函数 - 对DataFrame的每个元素应用函数
# 创建数据框
df2 = pd.DataFrame({
'Name': ['Alice', 'Bob', 'Charlie'],
'Score1': [85, 92, 78],
'Score2': [90, 88, 95]
})
print("原始数据：")
print(df2)
# 对每个元素应用函数
def grade(value):
if isinstance(value, (int, float)):
if value >= 90:
return '优秀'
elif value >= 80:
return '良好'
elif value >= 70:
return '中等'
else:
return '需要提高'
return value
# 只对数值列应用
df2_numeric = df2.select_dtypes(include=[np.number])
df2_numeric = df2_numeric.applymap(grade)
df2[['Score1', 'Score2']] = df2_numeric
print("\n应用grade函数后：")
print(df2)
agg() / aggregate() 函数 - 聚合应用
# 创建销售数据
sales_df = pd.DataFrame({
'Product': ['A', 'B', 'A', 'B', 'A', 'C'],
'Sales': [100, 200, 150, 180, 120, 300],
'Quantity': [10, 20, 15, 18, 12, 30]
})
print("销售数据：")
print(sales_df)
# 多个聚合函数
result = sales_df.groupby('Product').agg({
'Sales': ['sum', 'mean', 'max', 'min'],
'Quantity': ['sum', 'mean']
})
print("\n按产品分组聚合：")
print(result)
# 自定义聚合函数
def range_func(x):
return x.max() - x.min()
custom_agg = sales_df.groupby('Product').agg(range_func)
print("\n自定义聚合（计算范围）：")
print(custom_agg)
transform() 函数 - 转换数据
# 创建数据
df3 = pd.DataFrame({
'Group': ['A', 'A', 'A', 'B', 'B', 'B'],
'Value': [1, 2, 3, 4, 5, 6]
})
print("原始数据：")
print(df3)
# 标准化每个组内的数据
df3['Normalized'] = df3.groupby('Group')['Value'].transform(
lambda x: (x - x.mean()) / x.std()
)
print("\n组内标准化后：")
print(df3)
# 填充组内缺失值
df3.loc[2, 'Value'] = np.nan
df3['Filled'] = df3.groupby('Group')['Value'].transform(
lambda x: x.fillna(x.mean())
)
print("\n填充缺失值后：")
print(df3)
实际业务案例
案例1：客户数据分析
# 创建客户数据
customers = pd.DataFrame({
'CustomerID': range(1, 11),
'Name': ['Tom', 'Jerry', 'Alice', 'Bob', 'Charlie', 'Diana', 'Eve', 'Frank', 'Grace', 'Henry'],
'Age': [25, 32, 28, 45, 35, 22, 38, 50, 27, 41],
'Purchase_Amount': [500, 1200, 800, 2000, 1500, 300, 1800, 2500, 900, 1600],
'Purchase_Count': [5, 12, 8, 20, 15, 3, 18, 25, 9, 16]
})
# 计算客户价值评分
def customer_score(row):
"""计算客户价值评分"""
amount_score = row['Purchase_Amount'] / 2500 * 40 # 金额评分（最高40分）
count_score = row['Purchase_Count'] / 25 * 30 # 频次评分（最高30分）
age_score = min(row['Age'] / 50 * 30, 30) # 年龄评分（最高30分）
return amount_score + count_score + age_score
customers['Score'] = customers.apply(customer_score, axis=1)
# 客户分层
def customer_tier(score):
if score >= 80:
return 'VIP'
elif score >= 60:
return '黄金'
elif score >= 40:
return '白银'
else:
return '普通'
customers['Tier'] = customers['Score'].apply(customer_tier)
print("客户分析结果：")
print(customers[['Name', 'Purchase_Amount', 'Purchase_Count', 'Score', 'Tier']])
案例2：数据清洗
# 创建包含脏数据的数据集
dirty_data = pd.DataFrame({
'Name': ['John Smith', 'jane doe', 'Bob Johnson', 'ALICE WILLIAMS', ''],
'Email': ['john@email.com', 'jane@email', 'bob@email.com', 'alice@email.com', 'invalid'],
'Salary': [50000, 60000, -1, 75000, 55000],
'Dept': ['IT', 'HR', 'IT', None, 'Finance']
})
print("原始脏数据：")
print(dirty_data)
# 数据清洗函数
def clean_name(name):
"""清洗姓名：首字母大写"""
if pd.isna(name) or name.strip() == '':
return 'Unknown'
return name.strip().title()
def validate_email(email):
"""验证邮箱格式"""
if pd.isna(email) or '@' not in email or '.' not in email:
return 'Invalid'
return email.lower()
def clean_salary(salary):
"""清洗薪资：负数转为平均值"""
if salary < 0:
return np.nan
return salary
# 应用清洗函数
dirty_data['Name'] = dirty_data['Name'].apply(clean_name)
dirty_data['Email'] = dirty_data['Email'].apply(validate_email)
dirty_data['Salary'] = dirty_data['Salary'].apply(clean_salary)
# 填充缺失值
dirty_data['Salary'].fillna(dirty_data['Salary'].mean(), inplace=True)
dirty_data['Dept'].fillna('Unknown', inplace=True)
print("\n清洗后数据：")
print(dirty_data)
性能优化技巧
import time
# 准备大数据集
large_df = pd.DataFrame({
'A': np.random.randn(100000),
'B': np.random.randn(100000),
'C': np.random.choice(['X', 'Y', 'Z'], 100000)
})
# 方法1：使用循环（慢）
start = time.time()
result1 = []
for i in range(len(large_df)):
result1.append(large_df.iloc[i]['A'] + large_df.iloc[i]['B'])
time1 = time.time() - start
print(f"循环方法耗时: {time1:.4f}秒")
# 方法2：使用apply（中等）
start = time.time()
result2 = large_df.apply(lambda row: row['A'] + row['B'], axis=1)
time2 = time.time() - start
print(f"apply方法耗时: {time2:.4f}秒")
# 方法3：使用向量化操作（最快）
start = time.time()
result3 = large_df['A'] + large_df['B']
time3 = time.time() - start
print(f"向量化方法耗时: {time3:.4f}秒")
自定义函数的高级应用
# 复杂业务逻辑函数
def complex_business_logic(row):
"""复杂的业务逻辑判断"""
try:
# 条件判断
if row['Sales'] > 1000 and row['Quantity'] > 10:
return '高价值订单'
elif row['Sales'] > 500:
return '中等价值订单'
elif row['Quantity'] > 0:
return '低价值订单'
else:
return '异常订单'
except Exception as e:
return f'错误: {str(e)}'
# 应用并创建新列
sample_data = pd.DataFrame({
'Sales': [1500, 800, 300, 2000, 0],
'Quantity': [20, 5, 2, 15, -1]
})
sample_data['Order_Type'] = sample_data.apply(complex_business_logic, axis=1)
print("业务逻辑应用结果：")
print(sample_data)
这些案例涵盖了Pandas中数据应用函数的主要用法，从基本的apply到复杂的业务逻辑处理，记得根据实际需求选择合适的方法,尽量使用向量化操作来提高性能。
