Python提取超大tar.xc归档：解决内存错误与提速需求

问题描述 #

需要提取一个超大的tar.xz归档文件，其中包含一个大小接近15GB的.dat文件并保存至指定文件夹。使用tarfile.open(path/to/archive).getmembers().extractall()时出现MemoryError。

目前通过ExFileObject的迭代器逐行读取归档内文件，收集固定大小批次后保存为文件，解决了内存问题但处理速度极慢。希望找到按固定字节大小（如100MB）读取归档的方法来提升处理速度。

原实现代码 #

class RowsAccumulator:
    def __init__(self, border_list_size=300000000):
        self.border_list_size = border_list_size
        self.rows: bytes = b''
        self.generated_files = 1

    def add_row(self,row):
        # self.rows += str(row)
        self.rows += self.rows.join([row])
        return self.check_border()

    def check_border(self) -> bool:
        # if len(self.rows.encode('utf-8')) > self.border_lsit_size:
        print(len(self.rows), end='\r')
        if len(self.rows) > self.border_list_size:
            return True
        return False

    def delete_row(self):
        # self.rows = ''
        self.rows: bytes = b''
        self.generated_files += 1


def unpack_archive(path: str) -> object:
    try:
        file = tarfile.open(path, 'r:*')
    except:
        print('unexpected file type in {}'.format(path))
        return None
    print('unpacked {}'.format(path))
    return file


def deep_to_file(folder_with_archive, target_dir):
    unpacked = unpack_archive(folder_with_archive)
    if unpacked:
        for member in unpacked.getmembers():
            accum = RowsAccumulator()
            for log_row in unpacked.extractfile(member).__iter__():
                if accum.add_row(log_row):
                    with open(file_path,'w') as f:
                        f.write(accum.rows.decode(encoding='utf-8'))
                    f.close()
                    accum.delete_row()

优化方案 #

1. 修复核心逻辑错误

原代码中RowsAccumulator.add_row方法的self.rows += self.rows.join([row])是严重错误，该语句会将现有self.rows内容与新行重复拼接，导致字节串指数级增长，这是速度慢的核心原因之一。正确的拼接方式应为self.rows += row。

2. 改用固定字节块读取

放弃逐行迭代的方式，直接使用extractfile()返回文件对象的read(size)方法按固定大字节块读取，每次读取100MB（或更大的合理值，比如128MB），能大幅提升IO效率。

3. 简化写入逻辑，减少内存占用

不需要先在内存中累积固定大小的内容再写入，读取一块就直接写入目标文件，这样内存占用始终维持在单个块的大小，同时避免了内存中字节串拼接的开销。

优化后的代码示例 #

import os
import tarfile

def unpack_archive(path: str) -> tarfile.TarFile | None:
    try:
        return tarfile.open(path, 'r:*')
    except Exception as e:
        print(f'unexpected file type in {path}: {str(e)}')
        return None

def extract_large_member(archive: tarfile.TarFile, member: tarfile.TarInfo, target_dir: str, block_size: int = 1024 * 1024 * 100):
    # 确保目标目录存在
    os.makedirs(target_dir, exist_ok=True)
    target_path = os.path.join(target_dir, member.name)
    
    with archive.extractfile(member) as src, open(target_path, 'wb') as dst:
        while True:
            block = src.read(block_size)
            if not block:
                break
            dst.write(block)
            # 可选：打印进度
            print(f'已写入 {os.path.getsize(target_path)/1024/1024:.2f} MB', end='\r')
    print(f'\n文件 {member.name} 提取完成，保存至 {target_path}')

def deep_to_file(folder_with_archive, target_dir):
    unpacked = unpack_archive(folder_with_archive)
    if unpacked:
        for member in unpacked.getmembers():
            # 只处理文件，跳过目录等
            if member.isfile():
                extract_large_member(unpacked, member, target_dir)
        unpacked.close()

关键优化点说明 #

• 使用二进制模式直接读写，避免编码转换的额外开销（如果.dat文件是文本格式，可修改为open(target_path, 'w', encoding='utf-8')并写入block.decode('utf-8')）。
• 按大字节块读写，减少IO操作次数，充分利用磁盘IO带宽。
• 移除不必要的内存累积逻辑，内存占用仅为单个块的大小，彻底避免MemoryError。

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