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

wen python案例 2026-07-15 192

本文目录导读：

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

我来介绍如何使用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())

关键要点

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

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

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