我必须逐行处理一个庞大的pandas.DataFrame(几十GB),其中每行操作都很长(几十毫秒).所以我有了将框架拆分成块并使用多处理并行处理每个块的想法.这确实加快了任务,但内存消耗是一场噩梦.

虽然每个子进程原则上只占用一小部分数据,但它需要(几乎)与包含原始DataFrame的原始父进程一样多的内存.即使删除父进程中使用过的部分也无济于事.

我写了一个复制这种行为的最小例子.它唯一能做的就是创建一个带有随机数的大型DataFrame,将其分成最多100行的小块,并在多处理期间简单地打印一些有关DataFrame的信息(这里通过大小为4的mp.Pool).

并行执行的主要功能：

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的内存.但是,池的子进程需要相同的内存量,尽管它们只包含原始DataFame的1％(块大小为100,原始长度为10000).为什么？

我能做些什么呢？尽管我的分块,Python(3)是否将整个DataFrame发送到每个子进程？这是大熊猫内存管理的问题还是多处理和数据酸洗的错误？谢谢！

用于简单复制和粘贴的整个脚本,以防您想自己尝试：

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,pid))
def df_chunking(df,inplace=True)
def main():
    # Job parameters
    n_jobs = 4  # Poolsize
    size = (10000,chunksize))
    pool.close()
    pool.join()
    print('DONE')
if __name__ == '__main__':
    main()