使用线程池合并AWS S3小文件时出现内存溢出问题求助

你的实现存在明确的内存占用问题，直接导致了内存溢出，具体问题点如下：

核心内存问题 #

• 一次性加载所有文件内容到内存：read_file会把每个小文件的完整内容读入内存，read_write又会把当前组所有小文件的内容一次性拼接后写入大文件。这意味着每组文件的全部内容都会常驻内存，再加上线程池同时运行10个线程，相当于同时有10组文件的所有数据占满内存，处理10-15组后内存耗尽完全符合逻辑。
• S3FileSystem实例重复创建：每个read_write调用都新建一个s3fs.S3FileSystem()实例，会产生多余的连接资源和对象开销，间接加剧内存压力。

优化方案 #

1. 流式读写（关键优化） #

不要一次性加载所有文件内容，改为逐个读取小文件并即时写入大文件，让数据流式通过内存，大幅降低内存占用：

from multiprocessing.pool import ThreadPool

# 复用S3FileSystem实例，避免重复创建
s3 = s3fs.S3FileSystem()

def read_write(file_group):
    target_path, file_keys = file_group  # 假设file_group是(目标路径, 文件键列表)的结构
    with s3.open(target_path, 'wb') as target_f:
        for key in file_keys:
            # 逐个读取小文件并写入，不用把所有内容存在内存
            with s3.open(f"{bucket}/{key}", 'rb') as source_f:
                # 按块读写，进一步控制内存占用，比如每次读64KB
                while chunk := source_f.read(65536):
                    target_f.write(chunk)

# 假设file_groups是分组后的任务列表，每个元素是(目标路径, 该组的文件键列表)
ThreadPool(10).map(read_write, file_groups)

2. 合理控制分组大小 #

不要让单组文件的总大小过大，比如限制每组总大小在1GB以内，避免单线程处理时内存临时占用过高。

3. 复用S3连接 #

全局或线程内复用S3FileSystem实例，减少连接对象的重复创建开销。

4. 可选：监控内存占用 #

可以在代码中加入内存监控逻辑，方便排查是否还有其他内存泄漏点：

import psutil

def log_memory():
    process = psutil.Process()
    print(f"当前内存占用: {process.memory_info().rss / 1024 / 1024:.2f} MB")

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