Python读取gzip文件报zlib.error: Error -3解压错误解决方案

报错不是单条数据损坏导致的，核心问题是写入逻辑不符合gzip格式规范：

• 每次调用gzip.open(..., 'a')追加写入时，Python的gzip模块会在当前文件末尾写入一个全新的独立gzip压缩流（包含独立流头、压缩数据、校验尾），最终生成的文件是多个独立gzip块的直接拼接产物
• 常规gzip读取逻辑默认只识别第一个完整压缩流，读到后续拼接流的位置时，压缩上下文不匹配，就会抛出zlib.error: Error -3 while decompressing data: invalid distance too far back错误。

用zlib增量解压逻辑，遇到解压错误时自动重置解压上下文，跳过错误字节寻找下一个有效gzip流，同时自动跳过JSON解析失败的行，代码如下：

import zlib
import json

def read_multi_stream_gzip(file_path, chunk_size=1024*1024):
    valid_lines = []
    # 初始化适配gzip格式的增量解压对象
    decompressor = zlib.decompressobj(16 + zlib.MAX_WBITS)
    buffer = b""
    
    with open(file_path, "rb") as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            try:
                decompressed = decompressor.decompress(chunk)
                buffer += decompressed
                # 拆分已解压的完整行
                while b"\n" in buffer:
                    line_raw, buffer = buffer.split(b"\n", 1)
                    line_raw = line_raw.strip()
                    if not line_raw:
                        continue
                    try:
                        valid_lines.append(json.loads(line_raw.decode("utf-8")))
                    except json.JSONDecodeError:
                        continue
            except zlib.error:
                # 遇到流错误，重置解压对象，回退1字节继续扫描下一个有效流
                decompressor = zlib.decompressobj(16 + zlib.MAX_WBITS)
                f.seek(-(len(chunk) - 1), 1)
                buffer = b""
    # 处理缓冲区最后剩余的内容
    if buffer.strip():
        try:
            valid_lines.append(json.loads(buffer.decode("utf-8")))
        except:
            pass
    return valid_lines

# 调用示例
data = read_multi_stream_gzip("tweets.gz")

修复现有文件 #

系统自带的gzip命令原生支持解压多个拼接的gzip流，直接用命令行转储一次就能生成规范的单流gzip文件，成功率远高于自定义脚本：

gzip -dc tweets.gz | gzip -c > tweets_fixed.gz

修复后的tweets_fixed.gz可以直接用原有读取代码正常打开遍历。

修正后续写入逻辑 #

不要再每次写入都重新以追加模式打开gzip文件，两种正确写法二选一：

• 流式持续写入场景：初始化时打开一次文件句柄，所有数据写完再关闭

  # 初始化阶段打开文件，指定文本模式自动处理编码
  fout = gzip.open(output_path + out_fn, 'wt', encoding='utf-8')
  
  # 每次新数据到达时直接写入，无需重复open
  json_line = json.dumps(json.loads(data)) + "\n"
  fout.write(json_line)
  fout.flush()  # 按需调用，避免内容滞留缓冲区
  
  # 所有数据写入完成后关闭句柄
  fout.close()
  

• 必须跨批次/跨进程追加场景：放弃gzip追加模式，每次追加前先解压现有内容，合并新数据后重新压缩（大文件场景效率较低）。

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