优化Python文件/目录删除脚本运行时长的方案咨询

我有一个Python脚本，用来删除修改时间早于X天的文件及空目录，但在包含数百万文件和目录的超大目录中运行时，预估需要6周才能完成。分析后发现核心瓶颈是文件与目录的遍历环节。脚本以K8s Job形式运行，资源充足，当前代码如下：

import os
import logging
import re

EXAMINED_FILES = 0
DELETED_FILES = 0
EXAMINED_DIRS = 0
DELETED_DIRS = 0
SKIPPED_DELETE_DIRS = 0

def delete_files(root_directory, delete_time_threshold):
    global EXAMINED_FILES
    global DELETED_FILES
    try:
        for dirpath, dirnames, filenames in os.walk(root_directory):
            for file in filenames:
                file_path = os.path.join(dirpath, file)
                try:
                    file_mtime = os.path.getmtime(file_path)
                    EXAMINED_FILES += 1
                    if file_mtime < delete_time_threshold:
                        os.remove(file_path)
                        logging.debug(f"File {file} deleted because mtime {file_mtime} is older than threshold {delete_time_threshold}")
                        DELETED_FILES += 1
                except Exception as e:
                    logging.error(f"Error deleting file {file_path}")
    except Exception as e:
        logging.error(f"Error walking root directory {root_directory}: {e}")

def delete_empty_directories(root_directory, allowed_empty_dirs):
    global EXAMINED_DIRS
    global DELETED_DIRS
    global SKIPPED_DELETE_DIRS
    try:
        for dirpath, dirnames, filenames in os.walk(root_directory):
            if dirpath != root_directory: #don't look at the root directory
                EXAMINED_DIRS += 1
                try:
                    if not dirnames and not filenames:
                        relative_dirpath = re.sub(f'^{root_directory}/', '', dirpath)
                        if relative_dirpath and relative_dirpath in allowed_empty_dirs:
                            logging.debug(f"Skipping deletion of allowed empty directory: {dirpath}")
                            SKIPPED_DELETE_DIRS += 1
                        else:
                            os.rmdir(dirpath)
                            logging.debug(f"Deleted empty directory: {dirpath}")
                            DELETED_DIRS += 1
                except Exception as e:
                    logging.error(f"Error deleting directory {dirpath}")
    except Exception as e:
        logging.error(f"Error walking root directory {root_directory}: {e}")

核心优化方案 #

1. 替换os.walk为scandir，减少系统调用 #

os.walk每次获取文件修改时间都要单独发起系统调用，而os.scandir会在遍历目录时缓存文件的stat信息（包括mtime），直接读取缓存即可，比os.walk快2-3倍，这是最立竿见影的优化。

2. 合并两次遍历为一次，避免重复IO #

原脚本分两次遍历整个目录树（一次删文件、一次删空目录），直接翻倍了磁盘IO开销。改成先递归处理文件删除，再反向检查空目录（从最深层到顶层），一次遍历完成所有操作。

3. 用多进程并行处理，榨干资源 #

既然K8s资源充足，用多进程拆分顶层子目录的处理任务，并行遍历不同子树。注意不要用多线程（Python的GIL会限制CPU利用），用ProcessPoolExecutor更合适。

4. 减少不必要的计算与IO #

• 把正则替换相对路径改成字符串切片：relative_dirpath = dirpath[len(root_directory)+1:]，比正则快得多。
• 降低日志输出频率，比如每处理1000个文件/目录才输出一次，减少日志IO的开销。
• 去掉全局变量，改用字典传递统计信息，避免多进程下的同步问题。

优化后的完整代码 #

import os
import logging
from concurrent.futures import ProcessPoolExecutor
from functools import partial

def init_stats():
    return {
        'examined_files': 0,
        'deleted_files': 0,
        'examined_dirs': 0,
        'deleted_dirs': 0,
        'skipped_dirs': 0
    }

def process_dir(dirpath, delete_threshold, allowed_empty_dirs, stats):
    try:
        with os.scandir(dirpath) as entries:
            for entry in entries:
                if entry.is_file(follow_symlinks=False):
                    stats['examined_files'] += 1
                    # 直接从scandir缓存读取mtime，无需额外系统调用
                    if entry.stat().st_mtime < delete_threshold:
                        try:
                            os.remove(entry.path)
                            stats['deleted_files'] += 1
                            # 每删1000个文件打一次日志，减少IO
                            if stats['deleted_files'] % 1000 == 0:
                                logging.debug(f"已删除 {stats['deleted_files']} 个文件")
                        except Exception as e:
                            logging.error(f"删除文件失败 {entry.path}: {e}")
                elif entry.is_dir(follow_symlinks=False):
                    # 先递归处理子目录
                    process_dir(entry.path, delete_threshold, allowed_empty_dirs, stats)
                    # 子目录处理完后，检查当前目录是否为空
                    stats['examined_dirs'] += 1
                    try:
                        with os.scandir(entry.path) as sub_entries:
                            # any()会在第一个条目出现时停止扫描，效率更高
                            if not any(sub_entries):
                                # 用切片代替正则获取相对路径
                                relative_path = entry.path[len(dirpath)+1:] if len(entry.path) > len(dirpath)+1 else ''
                                if relative_path in allowed_empty_dirs:
                                    stats['skipped_dirs'] += 1
                                    logging.debug(f"跳过允许保留的空目录: {entry.path}")
                                else:
                                    os.rmdir(entry.path)
                                    stats['deleted_dirs'] += 1
                    except Exception as e:
                        logging.error(f"检查/删除目录失败 {entry.path}: {e}")
    except Exception as e:
        logging.error(f"处理目录失败 {dirpath}: {e}")

def clean_old_files_and_dirs(root_dir, delete_threshold, allowed_empty_dirs, max_workers=4):
    stats = init_stats()
    # 多进程并行处理顶层子目录
    with ProcessPoolExecutor(max_workers=max_workers) as executor:
        with os.scandir(root_dir) as root_entries:
            top_dirs = [entry.path for entry in root_entries if entry.is_dir(follow_symlinks=False)]
        # 绑定固定参数，传递给每个进程
        task_func = partial(process_dir, delete_threshold=delete_threshold,
                           allowed_empty_dirs=allowed_empty_dirs, stats=stats)
        executor.map(task_func, top_dirs)
    # 处理根目录下的文件
    process_dir(root_dir, delete_threshold, allowed_empty_dirs, stats)
    # 输出最终统计
    logging.info(f"清理完成：检查文件 {stats['examined_files']} 个，删除 {stats['deleted_files']} 个；"
                f"检查目录 {stats['examined_dirs']} 个，删除 {stats['deleted_dirs']} 个，跳过 {stats['skipped_dirs']} 个")
    return stats

额外提速建议 #

如果允许调用系统命令，直接用find工具会比Python脚本快一个数量级：

# 删除早于X天的文件（X替换为具体天数）
find /your/root/dir -type f -mtime +X -delete
# 删除空目录（排除允许保留的目录）
find /your/root/dir -type d -empty ! -path "/allowed/dir1" ! -path "/allowed/dir2" -delete

系统级工具是C语言实现的，遍历效率远超Python，适合超大规模目录场景。

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