我需要按行处理一个巨大的
尽管每个子进程原则上只应该消耗少量数据,但它需要(几乎)与包含原始
我写了一个最小示例来复制这种行为。它唯一做的就是创建一个具有随机数字的大型
在并行执行的主函数:
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()