Python案例如何用Pandas做数据并行处理

wen python案例 2026-07-15 1

本文目录导读：

使用 multiprocessing 进行并行处理
使用 swifter 库（推荐）
使用 pandas groupby 配合并行
使用 joblib 进行并行处理
使用 Dask 进行大规模并行处理
自定义并行循环
性能比较和最佳实践
注意事项

我来详细介绍几种使用Pandas进行数据并行处理的方法：

使用 multiprocessing 进行并行处理

基础示例：并行处理DataFrame

import pandas as pd
import numpy as np
from multiprocessing import Pool, cpu_count
import time
# 创建示例数据
def create_large_dataset(n_rows=100000):
    df = pd.DataFrame({
        'id': range(n_rows),
        'value1': np.random.randn(n_rows),
        'value2': np.random.randn(n_rows),
        'category': np.random.choice(['A', 'B', 'C', 'D'], n_rows)
    })
    return df
# 耗时处理函数
def complex_processing(row):
    # 模拟复杂计算
    result = np.sin(row['value1']) * np.cos(row['value2']) + np.sqrt(abs(row['value1']))
    time.sleep(0.001)  # 模拟耗时操作
    return result
# 并行处理版本
def parallel_processing(df, n_cores=None):
    if n_cores is None:
        n_cores = cpu_count() - 1  # 保留一个核心
    # 分割DataFrame
    df_split = np.array_split(df, n_cores)
    # 创建进程池并执行
    with Pool(n_cores) as pool:
        results = pool.map(complex_process_chunk, df_split)
    # 合并结果
    return pd.concat(results, axis=0)
def complex_process_chunk(chunk):
    # 处理数据块
    chunk['processed'] = chunk.apply(complex_processing, axis=1)
    return chunk
# 测试性能
def compare_performance():
    df = create_large_dataset(5000)
    # 串行处理
    start = time.time()
    result_serial = df.copy()
    result_serial['processed'] = result_serial.apply(complex_processing, axis=1)
    serial_time = time.time() - start
    # 并行处理
    start = time.time()
    result_parallel = parallel_processing(df)
    parallel_time = time.time() - start
    print(f"串行处理时间: {serial_time:.2f}秒")
    print(f"并行处理时间: {parallel_time:.2f}秒")
    print(f"速度提升: {serial_time/parallel_time:.2f}倍")
# 运行对比
compare_performance()

使用 swifter 库（推荐）

import swifter
import pandas as pd
import numpy as np
# 安装: pip install swifter
# 创建数据
df = pd.DataFrame({
    'A': np.random.randn(10000),
    'B': np.random.randn(10000)
})
# 复杂计算函数
def complex_calculation(x, y):
    # 模拟复杂计算
    result = 0
    for i in range(100):
        result += np.sin(x * i) * np.cos(y * i)
    return result
# 使用 swifter 自动并行
# swifter 会自动选择串行或并行
df['result'] = df.swifter.apply(
    lambda row: complex_calculation(row['A'], row['B']), 
    axis=1
)
# 也可以指定使用多核
df['result'] = df.swifter.progress_bar(False).apply(
    lambda row: complex_calculation(row['A'], row['B']), 
    axis=1
)

使用 pandas groupby 配合并行

import pandas as pd
import numpy as np
from multiprocessing import Pool
def process_group(group_df):
    """处理每个分组"""
    return group_df.groupby('category').agg({
        'value1': ['mean', 'std'],
        'value2': ['sum', 'count']
    })
# 创建数据
df = pd.DataFrame({
    'category': np.random.choice(['A', 'B', 'C', 'D', 'E'], 100000),
    'value1': np.random.randn(100000),
    'value2': np.random.randn(100000),
    'value3': np.random.randn(100000)
})
def parallel_groupby(df, group_col, n_cores=None):
    """并行分组处理"""
    if n_cores is None:
        n_cores = cpu_count() - 1
    # 获取唯一值并分割
    unique_groups = df[group_col].unique()
    group_chunks = np.array_split(unique_groups, n_cores)
    def process_combinations(groups):
        mask = df[group_col].isin(groups)
        subset = df[mask]
        return subset.groupby(group_col).agg({
            'value1': ['mean', 'std'],
            'value2': ['sum', 'count']
        })
    with Pool(n_cores) as pool:
        results = pool.map(process_combinations, group_chunks)
    return pd.concat(results)
# 执行并行groupby
result = parallel_groupby(df, 'category')

使用 joblib 进行并行处理

from joblib import Parallel, delayed
import pandas as pd
import numpy as np
# 创建数据
df = pd.DataFrame({
    'id': range(1000),
    'data': [np.random.randn(1000) for _ in range(1000)]
})
# 处理函数
def process_single(row):
    """处理单行数据"""
    data = row['data']
    return {
        'id': row['id'],
        'mean': np.mean(data),
        'std': np.std(data),
        'skew': pd.Series(data).skew(),
        'kurtosis': pd.Series(data).kurtosis()
    }
# 并行处理
def parallel_apply_joblib(df_func, n_jobs=-1):
    """使用joblib并行处理"""
    results = Parallel(n_jobs=n_jobs)(
        delayed(df_func)(row) for _, row in df.iterrows()
    )
    return pd.DataFrame(results)
# 执行并行处理
result_df = parallel_apply_joblib(process_single)

使用 Dask 进行大规模并行处理

import dask.dataframe as dd
import pandas as pd
import numpy as np
# 创建大规模数据
def create_big_data():
    n = 1000000  # 100万行
    df = pd.DataFrame({
        'x': np.random.randn(n),
        'y': np.random.randn(n),
        'category': np.random.choice(['A', 'B', 'C'], n)
    })
    return df
# 复杂计算函数
def complex_operation(df):
    """对DataFrame执行复杂操作"""
    df['z'] = df['x'] * df['y'] + np.sin(df['x']) * np.cos(df['y'])
    df['result'] = df.groupby('category')['z'].transform('mean')
    return df
# Dask并行处理
def dask_parallel_processing():
    # 创建Dask DataFrame
    ddf = dd.from_pandas(create_big_data(), npartitions=4)
    # 执行复杂操作
    result = ddf.map_partitions(complex_operation)
    # 触发计算
    final_result = result.compute()
    return final_result
# 执行
result = dask_parallel_processing()
print(result.head())

自定义并行循环

import pandas as pd
import numpy as np
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
class ParallelProcessor:
    def __init__(self, n_workers=None):
        self.n_workers = n_workers or cpu_count() - 1
    def parallel_for(self, df, func, axis=1):
        """通用并行循环"""
        if axis == 1:
            # 按行并行
            chunks = np.array_split(df, self.n_workers)
            with ProcessPoolExecutor(max_workers=self.n_workers) as executor:
                results = list(executor.map(self._process_chunk, chunks, [func]*len(chunks)))
            return pd.concat(results)
        else:
            # 按列并行
            cols = np.array_split(df.columns, self.n_workers)
            with ProcessPoolExecutor(max_workers=self.n_workers) as executor:
                results = list(executor.map(self._process_columns, [df]*len(cols), cols, [func]*len(cols)))
            return pd.concat(results, axis=1)
    def _process_chunk(self, chunk, func):
        """处理数据块"""
        return chunk.apply(func, axis=1)
    def _process_columns(self, df, columns, func):
        """处理列"""
        return df[columns].apply(func)
# 使用示例
df = pd.DataFrame(np.random.randn(1000, 10), columns=[f'col_{i}' for i in range(10)])
processor = ParallelProcessor(n_workers=4)
# 按行并行
result_rows = processor.parallel_for(df, lambda x: x.sum())
# 按列并行
result_cols = processor.parallel_for(df, lambda x: x.mean(), axis=0)

性能比较和最佳实践

def performance_benchmark():
    """性能基准测试"""
    import time
    # 创建中等规模数据
    df = pd.DataFrame({
        'A': np.random.randn(100000),
        'B': np.random.randn(100000),
        'C': np.random.randn(100000)
    })
    methods = {
        'serial': lambda: df.apply(lambda x: sum([np.sin(v) for v in x]), axis=1),
        'vectorized': lambda: np.sin(df['A']) + np.sin(df['B']) + np.sin(df['C']),
        'swifter': lambda: df.swifter.apply(lambda x: sum([np.sin(v) for v in x]), axis=1),
        'multiprocessing': lambda: parallel_processing(df)
    }
    for name, func in methods.items():
        start = time.time()
        result = func()
        elapsed = time.time() - start
        print(f"{name}: {elapsed:.3f}秒")
# 最佳实践总结
print("""
最佳实践建议：
1. 优先使用向量化操作（最快）
2. 数据量大时使用dask（可扩展到集群）
3. 普通并行使用swifter（最简单）
4. 复杂逻辑使用multiprocessing
5. IO密集型任务使用ThreadPoolExecutor
6. CPU密集型任务使用ProcessPoolExecutor
""")

注意事项

def parallel_cautions():
    """并行处理注意事项"""
    warnings = """
    1. 内存管理：
       - 注意内存占用，避免内存溢出
       - 使用array_split分割数据
    2. 数据依赖：
       - 确保数据分片之间没有依赖关系
       - 保持数据类型一致性
    3. 性能开销：
       - 小数据集不适合并行（通信开销大）
       - 数据量 < 10000行可能更慢
    4. 调试困难：
       - 并行代码难以调试
       - 建议先在串行模式下验证
    5. 资源管理：
       - 不要使用所有核心（保留1-2个）
       - 注意释放资源
    """
    print(warnings)
parallel_cautions()

选择哪种方法取决于：

数据规模（小/中/大）
计算复杂度（简单/复杂）
环境限制（单机/集群）
开发效率要求