如何流式读取3GB压缩、解压后超40GB的Gzip格式JSON大文件

• 待处理文件为3GB大小的Gzip压缩包，内部存储JSON格式数据，完整解压后体积超过60GB，本地存储空间无法容纳全量解压后的文件
• 常规处理方案需要先下载完整压缩包、全量解压后再读取数据，受存储限制无法落地
• 自行尝试边下载边解析时程序报错，提示需要完整文件才能处理，原有实现代码存在逻辑问题

原有参考代码如下：

default_path_download = '.'
def saveanddelete(download_url, name):
  with requests.get(download_url, headers=headersfordowlnoading, stream=True) as r:
      content_lenght = int(r.headers['Content-Length'])
      print(f"Download File Size : {round(content_lenght/1000000, 5)} MB")
      r.raise_for_status()
      with open(f"{default_path_download}/{name}", 'wb') as f:
          total_chunk_downloaded = 0
          chunk_size_to_get = 4096
          for chunk in r.iter_content(chunk_size=chunk_size_to_get):
              total_chunk_downloaded += len(chunk)
              if chunk:
                  f.write(chunk)
                  done = int(50 * total_chunk_downloaded / content_lenght)
                  print(f"Downloaded :  ", end='',flush=True)
                  print(f"{round(total_chunk_downloaded/1000000, 5)} MB [{'=' * done}{' ' * (50-done)}]\r", end='', flush=True)
                  break
          print("\n")
          print("Download Complete")

Gzip格式原生支持流式解压，不需要获取完整压缩包即可逐块解压已下载的字节流，配合流式JSON解析能力，可以实现全程不落地完整压缩包、不落地全量解压文件的边下载边处理，内存和磁盘占用都极低。

原有代码问题 #

你之前的实现报错核心原因有两个：

• 下载循环里写了break语句，仅读取第一个4KB大小的块就终止了下载，拿到的压缩流不完整，自然无法正常解压
• 代码逻辑是先把压缩块写入本地文件，再尝试读取解压，本质还是走全量下载再解压的逻辑，没有用到流式处理能力

可直接运行的实现 #

首先安装依赖：
pip install requests ijson

其中ijson是流式JSON解析库，不需要加载完整JSON内容即可逐块解析出结构化数据，避免内置json模块必须读取全量内容才能解析的问题。

实现代码：

import gzip
import requests
import ijson

def stream_process_gzip_json(download_url, request_headers, item_handler):
    """
    边下载Gzip压缩的JSON文件边解压、解析处理，无需本地存储完整压缩包/解压文件
    :param item_handler: 自定义单条数据处理函数，处理完单条数据立即释放内存
    """
    with requests.get(download_url, headers=request_headers, stream=True) as r:
        r.raise_for_status()
        total_compressed_size = int(r.headers.get('Content-Length', 0))
        # 直接将网络响应流接入Gzip解压流，不写入本地压缩文件
        gzip_decompress_stream = gzip.GzipFile(fileobj=r.raw)
        
        # 如果压缩包内是JSON数组格式，用ijson逐次迭代数组内的单个元素
        # 如果是JSON Lines逐行JSON格式，可直接逐行读取gzip_decompress_stream解析
        for item in ijson.items(gzip_decompress_stream, 'item'):
            # 执行业务逻辑，比如提取目标字段、写入小体积结果文件、存入数据库等
            item_handler(item)
            
            # 可选：输出处理进度
            downloaded_bytes = r.raw.tell()
            if total_compressed_size:
                progress = round(downloaded_bytes / total_compressed_size * 100, 2)
                print(f"处理进度: {progress}%\r", end='', flush=True)
    print("\n全量数据处理完成")

# 自定义单条数据处理逻辑示例
def custom_item_process(item):
    # 替换为你自己的业务逻辑即可
    print(f"当前处理数据ID: {item.get('id')}")

# 调用时传入你的下载地址、请求头、自定义处理函数即可
# stream_process_gzip_json(your_download_link, headersfordowlnoading, custom_item_process)

注意事项 #

• 如果压缩包内是JSON Lines格式（每行一个独立JSON对象），可以不用ijson，直接对解压流逐行读取，每读到完整一行就用json.loads解析即可，资源占用更低
• 整个处理流程的磁盘占用仅和你最终输出的结果文件大小有关，不需要预留3GB压缩包+60GB解压文件的存储空间，内存占用通常稳定在几十MB级别
• 不要尝试用内置json.load/json.loads加载解压流，这类方法必须拿到完整JSON内容才能解析，会直接触发内存溢出。

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