Databricks上PySpark序列化解压大7z文件时的内存问题

这个问题我之前处理超大压缩包的时候也踩过坑，核心矛盾就是你现在把所有解压后的内容都塞到一个列表里返回给DataFrame——不仅把Executor内存撑得满满当当，还直接触发了PySpark序列化的2G上限限制。给你几个实用的解决思路，按推荐优先级排序：

1. 直接在UDF内写入解压内容（最优方案） #

不要把所有解压后的内容都存在内存里返回，而是处理一个符合条件的文件就直接写到DBFS/ADLS存储，DataFrame里只记录这些文件的路径。这样完全避开了大对象序列化的问题，内存占用也极低，因为处理一个写一个，不用缓存所有内容。

修改后的代码示例：

import uuid
import io
import py7zr
from pyspark.sql import functions as F
from pyspark.sql.types import ArrayType, StringType

ENCODING = "utf-8"
# 定义解压后文件的存储根路径
UNZIPPED_ROOT = "/mnt/unzipped_content/"

@F.udf(returnType=ArrayType(StringType()))
def unzip_and_write_udf(content, source_file_path):
    output_paths = []
    # 用原文件的路径片段生成前缀，方便后续溯源
    source_prefix = source_file_path.split("/")[-1].replace("*", "_")
    
    with py7zr.SevenZipFile(io.BytesIO(content), mode='r') as z:
        for file_name, bytes_stream in z.readall().items():
            if file_name.startswith("v1") or file_name.startswith("v2"):
                # 生成唯一文件名，避免同文件名冲突
                unique_suffix = uuid.uuid4().hex
                target_file_name = f"{source_prefix}_{unique_suffix}_{file_name}"
                target_dbfs_path = f"{UNZIPPED_ROOT}{target_file_name}"
                # DBFS在Python环境中可以通过/dbfs/前缀直接访问本地文件系统
                local_target_path = f"/dbfs{target_dbfs_path}"
                
                # 直接写入内容到存储
                with open(local_target_path, 'w', encoding=ENCODING) as f:
                    f.write(bytes_stream.read().decode(ENCODING))
                
                output_paths.append(target_dbfs_path)
    return output_paths

# 读取原始压缩文件
df = spark.read.format("binaryFile").load("/mnt/file_pattern*")
# 调用UDF，传入原文件路径用于生成唯一文件名
df = df.withColumn("unzipped_file_paths", unzip_and_write_udf(F.col("content"), F.col("path")))
# 保存路径元数据（后续可以根据这些路径处理解压后的文件）
df.write.mode("overwrite").parquet("/mnt/test_dump_unzipped_metadata")

2. 分块读取解压内容（适合必须保留内容在DataFrame的场景） #

如果业务需求必须把解压内容留在DataFrame里，可以把大文件的内容分块读取，每个块控制在远小于2G的大小（比如4MB），这样列表里的每个元素都不会触发序列化限制。后续可以通过explode把每个块拆成单独的行，方便处理。

修改后的代码示例：

import io
import py7zr
from pyspark.sql import functions as F
from pyspark.sql.types import ArrayType, StringType

ENCODING = "utf-8"
CHUNK_SIZE = 4 * 1024 * 1024  # 4MB分块，可根据需要调整

@F.udf(returnType=ArrayType(StringType()))
def unzip_chunked_udf(content):
    extracted_chunks = []
    
    with py7zr.SevenZipFile(io.BytesIO(content), mode='r') as z:
        for file_name, bytes_stream in z.readall().items():
            if file_name.startswith("v1") or file_name.startswith("v2"):
                # 分块读取内容，避免一次性加载大文件到内存
                while True:
                    chunk = bytes_stream.read(CHUNK_SIZE)
                    if not chunk:
                        break
                    # 把每个块作为独立元素加入列表
                    extracted_chunks.append(chunk.decode(ENCODING))
    return extracted_chunks

# 读取原始压缩文件
df = spark.read.format("binaryFile").load("/mnt/file_pattern*")
# 调用分块解压UDF
df = df.withColumn("unzipped_chunks", unzip_chunked_udf(F.col("content")))
# 把分块拆成单独的行（可选，根据业务需求）
df = df.withColumn("unzipped_content", F.explode(F.col("unzipped_chunks")))
# 保存结果
df.write.mode("overwrite").parquet("/mnt/test_dump_unzipped_chunks")

3. 调整序列化相关参数（应急方案，不推荐） #

“can not serialize object larger than 2G”这个错误本质是Python pickle模块的旧限制（Python 3.4+已支持大对象，但PySpark的跨JVM序列化仍有约束）。你可以尝试调整以下Spark配置，但这只是临时应急手段，大对象仍会占用大量Executor内存，极易引发OOM：

• 在集群配置或Notebook开头添加：

  spark.conf.set("spark.driver.maxResultSize", "8g")
  spark.conf.set("spark.executor.memory", "16g")
  spark.conf.set("spark.python.worker.reuse", "false")
  

不过这个方案治标不治本，优先推荐前两种方法。

备注：内容来源于stack exchange，提问作者WilliamEllisWebb