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

本文目录导读：

1. 基础数据管道结构
2. 数据清理管道
3. 特征工程管道
4. 数据转换管道
5. 完整的数据处理管道类
6. 链式管道函数
7. 异步数据管道
8. 实际应用案例：销售数据分析管道
9. 关键要点

我来介绍如何使用Pandas构建数据管道函数,通过多个实践案例来说明。

基础数据管道结构

import pandas as pd
import numpy as np
from typing import List, Dict, Any, Callable
# 创建示例数据
def create_sample_data():
    np.random.seed(42)
    dates = pd.date_range('2023-01-01', periods=100, freq='D')
    data = {
        'date': dates,
        'product': np.random.choice(['A', 'B', 'C', 'D'], 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),
        'customer_id': np.random.randint(1000, 2000, 100)
    }
    # 添加一些缺失值
    data['quantity'][np.random.choice(100, 5)] = np.nan
    data['price'][np.random.choice(100, 3)] = np.nan
    return pd.DataFrame(data)

数据清理管道

def clean_pipeline(df: pd.DataFrame) -> pd.DataFrame:
    """
    数据清理管道
    """
    def remove_duplicates(data):
        """去重"""
        return data.drop_duplicates()
    def handle_missing_values(data):
        """处理缺失值"""
        data = data.copy()
        # 数值列用中位数填充
        numeric_cols = data.select_dtypes(include=[np.number]).columns
        for col in numeric_cols:
            data[col] = data[col].fillna(data[col].median())
        return data
    def remove_outliers(data, columns: List[str], threshold: float = 3):
        """移除异常值"""
        data = data.copy()
        for col in columns:
            z_scores = np.abs((data[col] - data[col].mean()) / data[col].std())
            data = data[z_scores < threshold]
        return data
    return (df
            .pipe(remove_duplicates)
            .pipe(handle_missing_values)
            .pipe(remove_outliers, columns=['quantity', 'price']))

特征工程管道

def feature_engineering_pipeline(df: pd.DataFrame) -> pd.DataFrame:
    """
    特征工程管道
    """
    def create_date_features(data):
        """创建日期特征"""
        data = data.copy()
        if 'date' in data.columns:
            data['year'] = data['date'].dt.year
            data['month'] = data['date'].dt.month
            data['day'] = data['date'].dt.day
            data['day_of_week'] = data['date'].dt.dayofweek
            data['is_weekend'] = data['day_of_week'].isin([5, 6]).astype(int)
        return data
    def create_aggregate_features(data):
        """创建聚合特征"""
        data = data.copy()
        # 对每个产品计算历史统计
        product_stats = data.groupby('product').agg({
            'quantity': ['mean', 'std', 'max'],
            'price': ['mean', 'std']
        }).round(2)
        # 展平列名
        product_stats.columns = ['_'.join(col).strip() for col in product_stats.columns.values]
        product_stats = product_stats.reset_index()
        # 合并回原数据
        data = data.merge(product_stats, on='product', how='left')
        return data
    def create_interaction_features(data):
        """创建交叉特征"""
        data = data.copy()
        data['total_value'] = data['quantity'] * data['price']
        data['log_total_value'] = np.log1p(data['total_value'])
        data['price_per_unit'] = data['price'] / data['quantity']
        return data
    return (df
            .pipe(create_date_features)
            .pipe(create_aggregate_features)
            .pipe(create_interaction_features))

数据转换管道

def transformation_pipeline(df: pd.DataFrame) -> pd.DataFrame:
    """
    数据转换管道
    """
    def encode_categorical(data, method: str = 'onehot'):
        """编码分类变量"""
        import warnings
        warnings.filterwarnings('ignore')
        data = data.copy()
        categorical_cols = data.select_dtypes(include=['object']).columns
        if method == 'label':
            from sklearn.preprocessing import LabelEncoder
            le = LabelEncoder()
            for col in categorical_cols:
                data[f'{col}_encoded'] = le.fit_transform(data[col])
        elif method == 'onehot':
            data = pd.get_dummies(data, columns=categorical_cols, prefix=categorical_cols)
        return data
    def normalize_numeric(data, method: str = 'minmax'):
        """标准化数值"""
        from sklearn.preprocessing import MinMaxScaler, StandardScaler
        data = data.copy()
        numeric_cols = data.select_dtypes(include=[np.number]).columns
        scaler = MinMaxScaler() if method == 'minmax' else StandardScaler()
        data[numeric_cols] = scaler.fit_transform(data[numeric_cols])
        return data
    return (df
            .pipe(encode_categorical, method='onehot')
            .pipe(normalize_numeric, method='standardize'))

完整的数据处理管道类

class DataPipeline:
    """
    完整的数据管道类
    """
    def __init__(self):
        self.steps = []
        self.logs = []
    def add_step(self, name: str, func: Callable, **kwargs):
        """添加管道步骤"""
        self.steps.append({
            'name': name,
            'function': func,
            'kwargs': kwargs
        })
        return self
    def run(self, df: pd.DataFrame) -> pd.DataFrame:
        """执行管道"""
        result = df.copy()
        for step in self.steps:
            start_shape = result.shape
            result = step['function'](result, **step['kwargs'])
            end_shape = result.shape
            self.logs.append({
                'step': step['name'],
                'start_shape': start_shape,
                'end_shape': end_shape,
                'changes': f"{(start_shape[0]-end_shape[0])} rows, {(start_shape[1]-end_shape[1])} cols"
            })
        return result
    def summarize(self):
        """管道摘要"""
        return pd.DataFrame(self.logs)
# 使用示例
def pipeline_example():
    pipeline = DataPipeline()
    pipeline.add_step('clean', clean_pipeline)
    pipeline.add_step('features', feature_engineering_pipeline)
    pipeline.add_step('transform', transformation_pipeline)
    df = create_sample_data()
    result = pipeline.run(df)
    print("原始数据形状:", df.shape)
    print("处理后数据形状:", result.shape)
    print("\n管道执行日志:")
    print(pipeline.summarize())
    return result
# 执行管道
result = pipeline_example()

链式管道函数

class ChainPipe:
    """
    链式管道实现
    """
    def __init__(self, data: pd.DataFrame):
        self.data = data
    def filter(self, condition: Callable):
        """过滤数据"""
        self.data = self.data[condition(self.data)]
        return self
    def select(self, columns: List[str]):
        """选择列"""
        self.data = self.data[columns]
        return self
    def mutate(self, **kwargs):
        """创建新列"""
        self.data = self.data.assign(**kwargs)
        return self
    def group_aggregate(self, group_cols: List[str], agg_dict: Dict):
        """分组聚合"""
        self.data = self.data.groupby(group_cols).agg(agg_dict).reset_index()
        return self
    def sort(self, by: List[str], ascending: bool = True):
        """排序"""
        self.data = self.data.sort_values(by=by, ascending=ascending)
        return self
    def get_data(self):
        """获取数据"""
        return self.data
# 使用示例
def chain_pipeline_example():
    df = create_sample_data()
    pipeline = (ChainPipe(df)
                .filter(lambda x: x['quantity'] > 10)
                .filter(lambda x: x['price'] > 50)
                .mutate(total = lambda x: x['quantity'] * x['price'])
                .sort(by=['total'], ascending=False)
                .select(['date', 'product', 'total'])
                .get_data())
    print("链式管道处理结果:")
    print(pipeline.head(10))
    return pipeline
# 执行
chain_pipeline_example()

异步数据管道

import asyncio
from concurrent.futures import ThreadPoolExecutor
class AsyncDataPipeline:
    """
    异步数据管道
    """
    def __init__(self, max_workers=4):
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
    async def process_chunk(self, chunk: pd.DataFrame, functions: List[Callable]):
        """异步处理数据块"""
        for func in functions:
            chunk = await asyncio.get_event_loop().run_in_executor(
                self.executor, func, chunk
            )
        return chunk
    async def parallel_process(self, df: pd.DataFrame, functions: List[Callable], n_chunks: int = 4):
        """并行处理大数据集"""
        chunks = np.array_split(df, n_chunks)
        tasks = [self.process_chunk(chunk, functions) for chunk in chunks]
        results = await asyncio.gather(*tasks)
        return pd.concat(results, ignore_index=True)
# 异步管道使用
async def async_pipeline_example():
    df = create_sample_data()
    pipeline = AsyncDataPipeline()
    functions = [clean_pipeline, feature_engineering_pipeline]
    result = await pipeline.parallel_process(df, functions, n_chunks=2)
    print(f"异步处理结果形状: {result.shape}")
    return result
# 运行异步管道
async_pipeline_example()

实际应用案例：销售数据分析管道

def sales_analysis_pipeline(df: pd.DataFrame) -> Dict[str, Any]:
    """
    销售分析完整管道
    """
    def calculate_revenue(data):
        """计算收入"""
        data['revenue'] = data['quantity'] * data['price']
        return data
    def analyze_performance(data):
        """分析表现"""
        # 产品表现
        product_performance = data.groupby('product').agg({
            'revenue': 'sum',
            'quantity': 'sum',
            'customer_id': 'nunique'
        }).round(2)
        product_performance['avg_order_value'] = (
            product_performance['revenue'] / product_performance['customer_id']
        ).round(2)
        # 类别表现
        category_performance = data.groupby('category').agg({
            'revenue': ['sum', 'mean'],
            'quantity': 'sum'
        }).round(2)
        return product_performance, category_performance
    def find_top_customers(data, n: int = 10):
        """找出最佳客户"""
        top_customers = (data.groupby('customer_id')
                        .agg({
                            'revenue': 'sum',
                            'quantity': 'sum',
                            'product': lambda x: x.mode().iloc[0] if not x.mode().empty else None
                        })
                        .nlargest(n, 'revenue')
                        .reset_index())
        return top_customers
    # 执行管道
    df_clean = df.pipe(calculate_revenue)
    product_perf, category_perf = analyze_performance(df_clean)
    top_customers = find_top_customers(df_clean)
    return {
        'processed_data': df_clean,
        'product_performance': product_perf,
        'category_performance': category_perf,
        'top_customers': top_customers
    }
# 运行销售分析
df = create_sample_data()
results = sales_analysis_pipeline(df)
print("产品表现:")
print(results['product_performance'].head())
print("\n客户表现:")
print(results['top_customers'].head())

关键要点

1. 管道模式优势：代码可复用、易于测试、流程清晰
2. 函数组合：使用.pipe()方法串联处理步骤
3. 灵活配置：通过参数化函数适应不同需求
4. 错误处理：添加适当的异常处理和日志记录
5. 性能优化：大数据集考虑使用Dask或并行处理

这些管道函数可以根据具体需求进行调整和扩展,形成完整的数据处理工作流。