Python案例如何用Pandas做数据应用函数

wen python案例 2026-07-15 193

本文目录导读：

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到复杂的业务逻辑处理，记得根据实际需求选择合适的方法,尽量使用向量化操作来提高性能。

本文地址： https://www.bmjup.com/post/21325.html