### 如何用Python压缩超大文件的最佳实践 #### 使用`shutil`模块进行简单压缩 对于较小的文件，可以使用内置的`shutil`模块来执行简单的压缩操作。然而，当面对超大文件时，这种方法可能不是最优的选择，因为整个文件会被加载到内存中。 ```python import shutil def simple_compress_file(input_path, output_path): shutil.make_archive(output_path, 'zip', input_path) ``` 此方法适用于相对较小的数据集[^1]。 #### 流式读取并分块写入ZIP档案 为了有效地处理超大文件而不占用过多内存资源，推荐采用流式读取的方式逐块处理文件内容，并将其逐步添加至ZIP存档中： ```python from zipfile import ZipFile import os CHUNK_SIZE = 8 * 1024 * 1024 # 设置每块大小为8MB def compress_large_file(file_name, zip_file_name): with ZipFile(zip_file_name, mode='w') as zf: file_size = os.path.getsize(file_name) with open(file_name, "rb") as f_in: while True: chunk = f_in.read(CHUNK_SIZE) if not chunk: break temp_filename = f"{file_name}.part" with open(temp_filename, "wb") as tmpf: tmpf.write(chunk) zf.write(temp_filename, arcname=os.path.basename(file_name)) os.remove(temp_filename) compress_large_file('large_data.txt', 'compressed.zip') ``` 上述代码展示了如何通过设置合理的缓冲区大小（如8 MB），从而实现对单个大型文本文件的有效压缩。这不仅减少了内存消耗，还提高了程序运行效率。 #### 利用多线程加速压缩过程 如果目标平台支持并发执行，则可以通过引入多线程机制进一步提升性能表现。需要注意的是，在实际部署前应充分测试以确保稳定性和安全性。 ```python from concurrent.futures import ThreadPoolExecutor from queue import Queue from threading import Lock class Compressor: def __init__(self, num_workers=4): self.num_workers = num_workers self.queue = Queue() self.lock = Lock() def worker(self): while True: task = self.queue.get() if task is None: break part_num, data_chunk = task filename_part = f'chunk_{part_num}' with self.lock: with open(filename_part, 'wb') as fp: fp.write(data_chunk) with ZipFile(f'result-{part_num}.zip', 'a') as zp: zp.write(filename_part) os.unlink(filename_part) def start_compression(self, source_filepath): total_parts = (os.stat(source_filepath).st_size + CHUNK_SIZE - 1) // CHUNK_SIZE workers = [] with ThreadPoolExecutor(max_workers=self.num_workers) as executor: for i in range(self.num_workers): t = executor.submit(self.worker) workers.append(t) with open(source_filepath, 'rb') as src_f: for idx in range(total_parts): content = src_f.read(CHUNK_SIZE) self.queue.put((idx, content)) for _ in range(self.num_workers): self.queue.put(None) for w in workers: w.result() if __name__ == '__main__': compressor = Compressor(num_workers=4) compressor.start_compression('very_large_file.bin') ``` 这段脚本实现了基于队列的任务分配模式以及多个工作线程之间的同步控制逻辑，能够显著加快大规模数据集的压缩速度。