Python案例如何用Pandas做数据Pipe管道

wen python案例 2026-07-15 1

本文目录导读：

基础Pipe管道使用
定义管道函数
使用pipe()方法进行管道操作
复杂数据处理管道
实用的数据清洗管道
带参数的管道函数
实际业务场景案例
高级管道技巧
性能优化建议

我来详细讲解如何使用Pandas进行数据管道（Pipe）操作,包含多个实用案例。

基础Pipe管道使用

import pandas as pd
import numpy as np
# 创建示例数据
df = pd.DataFrame({
    'name': ['Alice', 'Bob', 'Charlie', 'David', 'Eve'],
    'age': [25, 30, 35, 28, 32],
    'salary': [50000, 60000, 70000, 55000, 65000],
    'department': ['IT', 'HR', 'IT', 'Finance', 'HR']
})
print("原始数据:")
print(df)

定义管道函数

def filter_adults(df, min_age=25):
    """过滤年龄大于等于指定值的行"""
    return df[df['age'] >= min_age]
def add_tax_column(df, tax_rate=0.2):
    """添加税后工资列"""
    df = df.copy()
    df['tax'] = df['salary'] * tax_rate
    df['after_tax_salary'] = df['salary'] - df['tax']
    return df
def sort_by_salary(df, ascending=False):
    """按工资排序"""
    return df.sort_values('salary', ascending=ascending)
def group_by_department(df):
    """按部门分组并计算平均值"""
    return df.groupby('department').agg({
        'salary': 'mean',
        'age': 'mean',
        'tax': 'mean',
        'after_tax_salary': 'mean'
    }).round(2)

使用pipe()方法进行管道操作

# 方法1：链式调用pipe
result1 = (df.pipe(filter_adults, min_age=26)
             .pipe(add_tax_column, tax_rate=0.15)
             .pipe(sort_by_salary, ascending=False))
print("管道处理结果1:")
print(result1)
# 方法2：使用lambda表达式
result2 = (df.pipe(lambda x: x[x['department'].isin(['IT', 'HR'])])
             .pipe(add_tax_column, tax_rate=0.2)
             .pipe(group_by_department))
print("\n管道处理结果2:")
print(result2)

复杂数据处理管道

# 创建更复杂的数据集
df_complex = pd.DataFrame({
    'date': pd.date_range('2023-01-01', periods=100, freq='D'),
    'product': np.random.choice(['A', 'B', 'C'], 100),
    'category': np.random.choice(['Electronics', 'Clothing', 'Food'], 100),
    'quantity': np.random.randint(1, 100, 100),
    'price': np.random.uniform(10, 1000, 100).round(2),
    'region': np.random.choice(['North', 'South', 'East', 'West'], 100)
})
def add_revenue_column(df):
    """添加营收列"""
    df = df.copy()
    df['revenue'] = df['quantity'] * df['price']
    return df
def add_price_category(df):
    """添加价格分类"""
    df = df.copy()
    df['price_category'] = pd.cut(df['price'], 
                                  bins=[0, 100, 500, 1000],
                                  labels=['Low', 'Medium', 'High'])
    return df
def remove_outliers(df, column, std_threshold=3):
    """移除异常值"""
    mean = df[column].mean()
    std = df[column].std()
    return df[(df[column] >= mean - std_threshold * std) & 
              (df[column] <= mean + std_threshold * std)]
def aggregate_monthly(df):
    """按月聚合"""
    df = df.copy()
    df['month'] = df['date'].dt.month
    return df.groupby(['month', 'category']).agg({
        'revenue': 'sum',
        'quantity': 'sum',
        'price': 'mean'
    }).round(2)
# 复杂管道处理
complex_result = (df_complex.pipe(add_revenue_column)
                           .pipe(add_price_category)
                           .pipe(remove_outliers, 'revenue', 2)
                           .pipe(aggregate_monthly))
print("复杂管道处理结果:")
print(complex_result.head())

实用的数据清洗管道

# 创建包含脏数据的示例
df_dirty = pd.DataFrame({
    'id': [1, 2, 3, 4, 5, 6],
    'name': ['Alice', 'Bob', 'Charlie', 'David', 'Eve', None],
    'age': [25, 30, -5, 28, 150, 35],  # 包含无效年龄
    'email': ['alice@email.com', 'bob@email', 'charlie@email.com', 
              'david@email.com', None, 'eve@email.com'],
    'salary': [50000, 60000, 70000, None, 65000, 80000]
})
def clean_names(df, name_col='name'):
    """清理名称列"""
    df = df.copy()
    df[name_col] = df[name_col].fillna('Unknown').str.strip().str.title()
    return df
def clean_age(df, age_col='age', min_age=0, max_age=120):
    """清理年龄列"""
    df = df.copy()
    df[age_col] = df[age_col].clip(min_age, max_age)
    return df
def validate_email(df, email_col='email'):
    """验证邮箱格式"""
    df = df.copy()
    df['email_valid'] = df[email_col].str.contains(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', na=False)
    return df
def fill_missing_salary(df, salary_col='salary'):
    """填充缺失工资"""
    df = df.copy()
    df[salary_col] = df[salary_col].fillna(df[salary_col].median())
    return df
# 数据清洗管道
cleaned_df = (df_dirty.pipe(clean_names)
                      .pipe(clean_age)
                      .pipe(validate_email)
                      .pipe(fill_missing_salary))
print("清洗后的数据:")
print(cleaned_df)

带参数的管道函数

def apply_multiple_transformations(df):
    """应用多个转换并处理参数"""
    def standardize_date(df, date_cols=None):
        """标准化日期格式"""
        df = df.copy()
        if date_cols:
            for col in date_cols:
                df[col] = pd.to_datetime(df[col])
        return df
    def create_features(df, feature_config=None):
        """创建特征"""
        df = df.copy()
        if feature_config:
            for new_col, func in feature_config.items():
                df[new_col] = func(df)
        return df
    def normalize_columns(df, columns_to_normalize=None):
        """标准化数值列"""
        df = df.copy()
        if columns_to_normalize:
            for col in columns_to_normalize:
                df[f'{col}_normalized'] = (df[col] - df[col].mean()) / df[col].std()
        return df
    return df.pipe(standardize_date, date_cols=['date'])

实际业务场景案例

# 销售数据分析管道
sales_data = pd.DataFrame({
    'order_date': pd.date_range('2023-01-01', periods=200, freq='D'),
    'customer_id': np.random.randint(1000, 1100, 200),
    'product_category': np.random.choice(['A', 'B', 'C'], 200),
    'quantity': np.random.randint(1, 50, 200),
    'unit_price': np.random.uniform(10, 200, 200).round(2),
    'discount': np.random.uniform(0, 0.3, 200).round(2)
})
def calculate_net_revenue(df):
    """计算净收入"""
    df = df.copy()
    df['gross_revenue'] = df['quantity'] * df['unit_price']
    df['discount_amount'] = df['gross_revenue'] * df['discount']
    df['net_revenue'] = df['gross_revenue'] - df['discount_amount']
    return df
def customer_segmentation(df):
    """客户分群"""
    df = df.copy()
    df['order_month'] = df['order_date'].dt.month
    customer_stats = df.groupby('customer_id').agg({
        'net_revenue': 'sum',
        'order_date': 'count'
    }).rename(columns={'order_date': 'order_count'})
    # 简单分群逻辑
    customer_stats['segment'] = pd.cut(
        customer_stats['net_revenue'],
        bins=[0, 1000, 5000, float('inf')],
        labels=['Low Value', 'Medium Value', 'High Value']
    )
    return df.merge(customer_stats[['segment']], left_on='customer_id', right_index=True)
def sales_summary(df):
    """生成销售摘要"""
    return df.groupby(['product_category', 'segment']).agg({
        'net_revenue': ['sum', 'mean'],
        'quantity': 'sum',
        'customer_id': 'nunique'
    }).round(2)
# 执行完整销售分析
sales_analysis = (sales_data.pipe(calculate_net_revenue)
                           .pipe(customer_segmentation)
                           .pipe(sales_summary))
print("销售分析结果:")
print(sales_analysis)

高级管道技巧

# 条件管道执行
def conditional_pipe(df, condition_func, true_func, false_func):
    """根据条件执行不同的管道函数"""
    if condition_func(df):
        return true_func(df)
    else:
        return false_func(df)
# 缓存中间结果
class PipeCache:
    def __init__(self):
        self.cache = {}
    def cached_pipe(self, df, func, cache_key):
        if cache_key not in self.cache:
            self.cache[cache_key] = func(df)
        return self.cache[cache_key]
# 使用示例
cache = PipeCache()
def expensive_operation(df):
    """模拟耗时的操作"""
    import time
    time.sleep(0.1)
    return df.copy()
# 使用缓存
result_cached = (df.pipe(lambda x: cache.cached_pipe(x, expensive_operation, 'step1'))
                    .pipe(add_tax_column))

性能优化建议

# 使用@pipe装饰器
from functools import wraps
def pipeable(func):
    """将函数转换为可管道使用的函数"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper
@pipeable
def custom_transform(df, column):
    df = df.copy()
    df[f'{column}_transformed'] = df[column] * 2
    return df
# 链式调用自定义函数
result_custom = df.pipe(custom_transform, 'salary')

Pandas的pipe管道用法提供了：

代码可读性：清晰的转换流程
可重用性：函数可以重复使用
调试方便：可以在任意步骤检查数据
灵活性：支持复杂的条件逻辑
模块化：将复杂操作分解为小函数

建议在数据清洗和特征工程等需要多个步骤的场景中,优先使用管道方式组织代码。