我需要按行处理一个巨大的
尽管每个子进程原则上只应该消耗少量数据，但它需要(几乎)与包含原始
我写了一个最小示例来复制这种行为。它唯一做的就是创建一个具有随机数字的大型
在并行执行的主函数：
pandas.DataFrame(几十GB)，每一行操作都非常耗时(数十毫秒)。所以我想把框架分成块，使用multiprocessing并行处理每个块。这确实加快了任务的速度，但内存消耗是一场噩梦。尽管每个子进程原则上只应该消耗少量数据，但它需要(几乎)与包含原始
DataFrame的父进程一样多的内存。即使在父进程中删除已使用的部分也没有帮助。我写了一个最小示例来复制这种行为。它唯一做的就是创建一个具有随机数字的大型
DataFrame，将其分成最多100行的小片段，并在mp.Pool（这里是大小为4）中通过一些信息简单地打印有关DataFrame的内容。在并行执行的主函数：
def just_wait_and_print_len_and_idx(df):
"""Waits for 5 seconds and prints df length and first and last index"""
# Extract some info
idx_values = df.index.values
first_idx, last_idx = idx_values[0], idx_values[-1]
length = len(df)
pid = os.getpid()
# Waste some CPU cycles
time.sleep(1)
# Print the info
print('First idx {}, last idx {} and len {} '
'from process {}'.format(first_idx, last_idx, length, pid))
生成器助手可以将 DataFrame 分割成小块：
def df_chunking(df, chunksize):
"""Splits df into chunks, drops data of original df inplace"""
count = 0 # Counter for chunks
while len(df):
count += 1
print('Preparing chunk {}'.format(count))
# Return df chunk
yield df.iloc[:chunksize].copy()
# Delete data in place because it is no longer needed
df.drop(df.index[:chunksize], inplace=True)
主程序如下：
def main():
# Job parameters
n_jobs = 4 # Poolsize
size = (10000, 1000) # Size of DataFrame
chunksize = 100 # Maximum size of Frame Chunk
# Preparation
df = pd.DataFrame(np.random.rand(*size))
pool = mp.Pool(n_jobs)
print('Starting MP')
# Execute the wait and print function in parallel
pool.imap(just_wait_and_print_len_and_idx, df_chunking(df, chunksize))
pool.close()
pool.join()
print('DONE')
标准输出如下：
Starting MP
Preparing chunk 1
Preparing chunk 2
First idx 0, last idx 99 and len 100 from process 9913
First idx 100, last idx 199 and len 100 from process 9914
Preparing chunk 3
First idx 200, last idx 299 and len 100 from process 9915
Preparing chunk 4
...
DONE
问题：
主进程需要大约120MB的内存。然而，池子的子进程需要相同数量的内存，尽管它们仅包含原始DataFrame的1％（大小为100的块与原始长度为10000的数据）。为什么？
我该怎么办？Python（3）是否会将整个DataFrame发送到每个子进程中，而不管我的块处理方式？这是pandas内存管理的问题，还是multiprocessing和数据pickling的问题？谢谢！
完整的脚本，供您简单复制和粘贴，以便自行尝试：
import multiprocessing as mp
import pandas as pd
import numpy as np
import time
import os
def just_wait_and_print_len_and_idx(df):
"""Waits for 5 seconds and prints df length and first and last index"""
# Extract some info
idx_values = df.index.values
first_idx, last_idx = idx_values[0], idx_values[-1]
length = len(df)
pid = os.getpid()
# Waste some CPU cycles
time.sleep(1)
# Print the info
print('First idx {}, last idx {} and len {} '
'from process {}'.format(first_idx, last_idx, length, pid))
def df_chunking(df, chunksize):
"""Splits df into chunks, drops data of original df inplace"""
count = 0 # Counter for chunks
while len(df):
count += 1
print('Preparing chunk {}'.format(count))
# Return df chunk
yield df.iloc[:chunksize].copy()
# Delete data in place because it is no longer needed
df.drop(df.index[:chunksize], inplace=True)
def main():
# Job parameters
n_jobs = 4 # Poolsize
size = (10000, 1000) # Size of DataFrame
chunksize = 100 # Maximum size of Frame Chunk
# Preparation
df = pd.DataFrame(np.random.rand(*size))
pool = mp.Pool(n_jobs)
print('Starting MP')
# Execute the wait and print function in parallel
pool.imap(just_wait_and_print_len_and_idx, df_chunking(df, chunksize))
pool.close()
pool.join()
print('DONE')
if __name__ == '__main__':
main()