我的个人技术笔记

21.6 Copy-on-Write与并行处理

4 min read

核心内容

Copy-on-Write（CoW） 减少 DataFrame 不必要的内存拷贝；并行处理工具 利用多核 CPU 加速大规模计算。

一、Copy-on-Write 减少拷贝

什么是 CoW

Copy-on-Write 是 pandas 2.0 引入的写时复制机制：只有在对数据执行修改时，才会触发真正的拷贝；读操作共享底层数据，显著减少内存占用与拷贝开销。

import pandas as pd
 
df = pd.DataFrame({'a': [1, 2, 3], 'b': [4, 5, 6]})
 
# 默认情况下，切片会复制数据
sub = df[['a']]
sub['a'] = 100          # 修改 sub 不影响 df
 
# 启用 CoW 后：
# 子 DataFrame 共享底层数据，修改时仅在写入点拷贝

启用方式

# 方式 1：全局选项
pd.set_option('mode.copy_on_write', True)
 
# 方式 2：future 选项（推荐，提前适配 3.x）
pd.set_option('future.copy_on_write', True)
 
# 方式 3：配置文件 / 环境变量
# pandas 2.3 可通过环境变量 PANDAS_COPY_ON_WRITE=1

CoW 带来的行为变化

操作默认行为CoW 行为
df2 = df[['a']]可能复制共享数据，零拷贝
df2['a'] = 1修改 df2（可能触发 SettingWithCopyWarning）仅修改 df2，df 不变
链式赋值警告或不一致明确安全
df['new'] = ...可能原地修改安全写时复制

链式赋值问题

# 反模式（启用 CoW 后可避免）
df[df['a'] > 1]['b'] = 99
 
# 推荐写法
df.loc[df['a'] > 1, 'b'] = 99

CoW 的性能收益

  • 大量切片 / 过滤操作时，共享内存避免重复拷贝
  • 内存峰值显著降低
  • 注意事项：必须使用 .loc/.iloc 进行赋值，df['col'] = ... 在 CoW 下语义等同拷贝

二、并行处理工具

1. multiprocessing — 多进程并行

适合 CPU 密集型任务（pandas 本身单线程）：

import pandas as pd
from multiprocessing import Pool
 
df = pd.DataFrame({
    'group': np.random.choice(list('ABCD'), 100_000),
    'value': np.random.rand(100_000)
})
 
def process_group(g):
    sub = df[df['group'] == g]
    return g, sub['value'].mean(), sub['value'].std()
 
# 多进程并行
with Pool(4) as pool:
    results = pool.map(process_group, list('ABCD'))
 
# 也可以按块并行
chunks = np.array_split(df, 8)
 
def process_chunk(chunk):
    return chunk.groupby('group')['value'].sum()
 
with Pool(8) as pool:
    partials = pool.map(process_chunk, chunks)
 
final = pd.concat(partials).groupby(level=0).sum()

2. concurrent.futures

from concurrent.futures import ProcessPoolExecutor
 
with ProcessPoolExecutor(max_workers=4) as executor:
    futures = [executor.submit(process_group, g) for g in 'ABCD']
    results = [f.result() for f in futures]

3. swifter / pandarallel（第三方）

# swifter：自动选择向量化 / apply / dask
import swifter
df['new'] = df.swifter.apply(lambda r: custom_func(r), axis=1)
 
# pandarallel：并行 apply
from pandarallel import pandarallel
pandarallel.initialize(progress_bar=True)
df['new'] = df.parallel_apply(lambda r: custom_func(r), axis=1)

4. numba 并行

from numba import njit, prange
 
@njit(parallel=True)
def parallel_sum(arr):
    total = 0.0
    for i in prange(len(arr)):
        total += arr[i]
    return total
 
parallel_sum(np.random.rand(10_000_000))

并行注意事项

  • 多进程需在 if __name__ == '__main__': 保护下运行（Windows）
  • 数据需要在进程间序列化，小数据并行反而更慢
  • pandas 内部许多操作（groupby、merge）本身已有 C 级优化，优先考虑向量化

适用场景决策

场景方案
读多写少的切片操作启用 CoW
CPU 密集型自定义函数multiprocessing / swifter
大规模分组计算分块 + 并行聚合
循环型数值算法numba @njit(parallel=True)
I/O 密集型使用 pyarrow / 分块读取

相关笔记

Graph View

Backlinks