基于Databricks PySpark批量增量解压大Zip文件方案咨询

针对你描述的场景——每日20个5GB Zip文件（解压后50GB CSV）、跨存储账户容器挂载、Databricks Runtime 12.1 + 8台Standard_DS3_v2节点，以下是落地实现方案：

一、增量处理：精准识别未处理文件 #

核心思路是维护处理元数据日志，用Delta Lake存储已处理文件的路径和时间戳，避免重复处理。

1. 初始化处理日志表 #

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, TimestampType
from pyspark.sql.functions import current_timestamp

spark = SparkSession.builder.appName("ZipIncrementalProc").getOrCreate()

# 定义日志表结构
log_schema = StructType([
    StructField("zip_file_path", StringType(), nullable=False),
    StructField("processing_time", TimestampType(), nullable=False)
])

# 创建Delta日志表（首次运行）
log_table_path = "/dbfs/mnt/cnt-output/processed_zip_logs"
if not spark.catalog.tableExists("processed_zip_logs"):
    spark.createDataFrame([], log_schema).write.format("delta").save(log_table_path)
    spark.sql(f"CREATE TABLE processed_zip_logs USING DELTA LOCATION '{log_table_path}'")

2. 筛选未处理文件 #

通过左连接对比输入文件列表与已处理日志，得到待处理文件集合：

# 扫描输入目录下所有Zip文件
input_zips = spark.read.format("binaryFile").load("/mnt/cnt-input/*.zip").select("path")

# 读取已处理日志
processed_zips = spark.table("processed_zip_logs").select("zip_file_path")

# 过滤未处理文件（左反连接）
unprocessed_zips = input_zips.join(processed_zips, input_zips.path == processed_zips.zip_file_path, "left_anti")

二、批量解压Zip并保存CSV #

利用Spark分布式并行能力处理Zip文件，避免单节点瓶颈。这里采用Python标准库zipfile实现解压，结合UDF完成分布式计算。

1. 解压UDF实现（返回完整CSV内容） #

import zipfile
from io import BytesIO
from pyspark.sql.functions import udf, col, explode
from pyspark.sql.types import ArrayType, StructType, StructField, StringType

# 定义解压后CSV数据结构
csv_data_schema = StructType([
    StructField("source_zip", StringType(), nullable=False),
    StructField("csv_filename", StringType(), nullable=False),
    StructField("csv_content", StringType(), nullable=False)
])

@udf(returnType=ArrayType(csv_data_schema))
def extract_zip_content(zip_binary, zip_path):
    csv_items = []
    with BytesIO(zip_binary) as zip_buf:
        with zipfile.ZipFile(zip_buf, "r") as zf:
            for file_name in zf.namelist():
                if file_name.endswith(".csv"):
                    with zf.open(file_name) as csv_file:
                        content = csv_file.read().decode("utf-8")
                        csv_items.append({
                            "source_zip": zip_path,
                            "csv_filename": file_name,
                            "csv_content": content
                        })
    return csv_items

# 读取未处理Zip的二进制数据并解压
zip_with_binary = input_zips.join(unprocessed_zips, input_zips.path == unprocessed_zips.path).select("path", "content")
extracted_csvs = zip_with_binary.withColumn("csv_items", explode(extract_zip_content(col("content"), col("path")))).select("csv_items.*")

2. 写入输出容器 #

将解压后的CSV直接写入cnt-output，可按源Zip文件分区避免重名：

# 自定义输出路径规则（添加源Zip哈希前缀避免文件名冲突）
from pyspark.sql.functions import udf
import hashlib

@udf(StringType())
def get_output_path(source_zip, csv_filename):
    zip_hash = hashlib.md5(source_zip.encode()).hexdigest()[:8]
    return f"/mnt/cnt-output/unzipped/{zip_hash}_{csv_filename}"

# 生成输出路径并保存CSV
csv_with_path = extracted_csvs.withColumn("output_path", get_output_path(col("source_zip"), col("csv_filename")))

# 用foreachPartition写入文件（确保每个CSV独立存储）
def write_csv_partition(partition):
    for row in partition:
        with open(row["output_path"], "w", encoding="utf-8") as f:
            f.write(row["csv_content"])

csv_with_path.foreachPartition(write_csv_partition)

3. 更新处理日志 #

处理完成后，将已处理的Zip文件路径写入日志表：

unprocessed_zips.withColumn("processing_time", current_timestamp()) \
    .withColumnRenamed("path", "zip_file_path") \
    .write.format("delta").mode("append").save(log_table_path)

三、基于指定集群的性能优化配置 #

针对8台Standard_DS3_v2（4vCPU/14GB内存）、Runtime 12.1，调整以下Spark配置最大化性能：

# 并行度与分区优化
spark.conf.set("spark.sql.shuffle.partitions", "32")  # 8节点×4核，匹配集群核心数
spark.conf.set("spark.default.parallelism", "32")
spark.conf.set("spark.sql.adaptive.enabled", "true")  # 开启自适应执行，自动调整分区

# 内存配置（预留4GB给系统进程）
spark.conf.set("spark.executor.memory", "10g")
spark.conf.set("spark.driver.memory", "10g")
spark.conf.set("spark.executor.memoryOverhead", "2g")

# IO与文件优化
spark.conf.set("spark.sql.files.maxRecordsPerFile", "100000")  # 限制输出文件大小，避免小文件
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.split.maxsize", "134217728")  # 128MB拆分输入文件
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.split.minsize", "67108864")  # 64MB最小拆分

# 压缩配置（若需压缩输出CSV）
spark.conf.set("spark.sql.csv.compression.codec", "gzip")

额外优化点 #

• 存储挂载优化：确保ADLS Gen2容器用OAuth认证（而非SAS），减少IO认证开销
• 节点预热：运行前先执行小批量任务预热节点缓存
• 监控与调优：通过Databricks UI监控任务进度，调整分区数或内存配置

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