Python检测内存占用过高时停止缓存DataFrame的最优实现方法

最优实现思路是内存阈值检测 + LRU缓存淘汰机制结合，避免只检测内存不淘汰导致缓存命中率大幅下降的问题。

第一步：依赖选择 #

内存检测用Python通用的psutil库，可以精准获取当前进程内存占用、系统剩余内存等指标，无需调用系统底层命令，兼容性覆盖Windows/Linux/macOS。

第二步：核心逻辑改造 #

核心逻辑包含3部分：

• 缓存使用LRU结构，内存不足时优先淘汰最久未使用的缓存条目
• 写入缓存前先检测当前进程内存占用是否超过预设阈值，同时预估新写入的DataFrame大小
• 若当前剩余可用内存不足以容纳新的DataFrame，先淘汰旧缓存直到有足够空间，若淘汰完所有旧缓存仍不够则跳过缓存

代码示例 #

import psutil
from collections import OrderedDict
import pandas as pd

class Foo:
    # 缓存改用有序字典实现LRU
    _cache = OrderedDict()
    # 配置项：进程内存占用最大阈值，示例设为系统总内存的70%，也可固定为字节数值
    MAX_PROCESS_MEM_RATIO = 0.7

    @classmethod
    def _get_available_memory(cls) -> int:
        """获取当前进程还能使用的内存空间（单位：字节）"""
        process = psutil.Process()
        current_used = process.memory_info().rss
        total_system = psutil.virtual_memory().total
        max_allowed = total_system * cls.MAX_PROCESS_MEM_RATIO
        return max(max_allowed - current_used, 0)

    @classmethod
    def get_df(cls, bar):
        # 命中缓存则把条目移到队尾，标记为最近使用
        if bar in cls._cache:
            cls._cache.move_to_end(bar)
            return cls._cache[bar]
        
        # 先计算目标df
        df = cls.calculate_df(bar)
        # 计算df真实内存占用，deep=True才能正确统计object类型列的大小
        df_size = df.memory_usage(deep=True).sum()

        # 检测内存，预留20%冗余避免其他操作触发内存溢出
        available = cls._get_available_memory()
        while available < df_size * 1.2 and len(cls._cache) > 0:
            # 淘汰最久未使用的队首条目
            cls._cache.popitem(last=False)
            available = cls._get_available_memory()
        
        # 清理后内存足够才写入缓存
        if available >= df_size * 1.2:
            cls._cache[bar] = df
        
        return df

    @classmethod
    def calculate_df(cls, bar):
        # 原有计算逻辑保持不变
        ......
        return df

可选优化点 #

• 如果bar的取值有明显冷热区分，可以加TTL过期机制，定期清理长时间未使用的缓存，进一步降低内存占用
• 若单份DataFrame体积差异极大，可以额外设置单条缓存的最大大小阈值，超过阈值的大df直接跳过缓存，避免挤占小df的缓存空间
• 多进程场景下需要调低内存阈值，避免多个进程同时占用内存超过系统限制

内容的提问来源于stack exchange，提问作者abisko