Tenho um programa que lê arquivos binários muito grandes (~100 GB-TB) em blocos usando numpy memmap. O programa faz uma única passagem pelos dados, então não há necessidade de armazenar nada em cache, já que nunca há necessidade de voltar e reler, mas, np.memmappor padrão, armazena os dados em cache. Como resultado, o uso de RAM satura rapidamente, mesmo sem haver necessidade real disso.
Existe alguma maneira de desativar o cache de dados ou, caso contrário, limpar o cache manualmente? Encontrei alguns outros tópicos sobre o assunto que sugerem que a única maneira é excluir todas as referências a flushele memmap, executar o coletor de lixo e recriar o arquivo memmapdo zero, o que funciona, mas obviamente não é o ideal. Posso fazer melhor?
Aqui está um MWE que demonstra o problema (observe que ele criará 2 GB de lixo aleatório no seu HDD se você executá-lo). Como você pode ver, o uso de RAM reflete a quantidade cumulativa de dados carregados, mesmo que seja chunk_sizepequena. Idealmente, o uso de RAM seria limitado à quantidade de dados contidos em um único bloco.
import numpy as np
import os
import psutil
import gc
import time
# Parameters
filename = 'test_memmap.bin'
file_size_gb = 2 # Change this if needed
dtype = np.float32
element_size = np.dtype(dtype).itemsize
num_elements = (file_size_gb * 1024**3) // element_size
chunk_size = 1_000_000 # Number of elements to read at once
# Step 1: Create a large binary file
if not os.path.exists(filename):
print("Creating file...")
with open(filename, 'wb') as f:
f.write(np.random.rand(num_elements).astype(dtype).tobytes())
# Step 2: Process file using memmap in chunks
print("Processing with memmap...")
mm = np.memmap(filename, dtype=dtype, mode='r')
process = psutil.Process(os.getpid())
for i in range(0, len(mm), chunk_size):
chunk = mm[i:i+chunk_size]
# Simulate processing
chunk.sum()
# Monitor RAM usage
mem = process.memory_info().rss / (1024 ** 2) # in MB
print(f"Step {i // chunk_size + 1}, RAM usage: {mem:.2f} MB")
del mm
gc.collect()
time.sleep(5) #os takes a second to catch up
mem = process.memory_info().rss / (1024 ** 2) # in MB
print(f"Final RAM usage after deleting memmap: {mem:.2f} MB")