PySpark读取MinIO海量小Parquet文件过慢的优化方案咨询

问题背景 #

• 源数据是MinIO（兼容S3）上的大量小Parquet文件（单文件约150KB），所有文件schema一致
• 使用spark.read.parquet(*keys)读取时耗时极长，40CPU集群仅读取操作就需约1小时
• 尝试通过压缩器预读取、重分区后持久化到PVC，但读取阶段耗时仍居高不下
• 改用Spark+Boto3将文件下载到PVC，发现Boto3下载速度更快，但存在异常：指定84000个S3密钥，仅能下载约3000个就标记任务完成（集群配置为20个执行器，每个1GB内存/1核）
• S3路径包含结构化元数据，示例格式：{bucket_name}/{metadata1=abc}/{metadata2=abc}/{metadata3=abc}/data.parquet

优化方案 #

一、直接优化Spark读取性能（优先推荐） #

无需额外下载文件，通过调整Spark配置和读取逻辑解决小文件问题：

1. 调整Spark S3相关核心配置：
   • spark.sql.parquet.mergeSchema=false：因schema统一，关闭schema合并减少不必要的开销
   • spark.hadoop.fs.s3a.connection.maximum=100：增大S3连接池上限，提升并发读取能力
   • spark.sql.files.maxPartitionBytes=128m：调整分区大小阈值，减少小文件带来的过多分区
   • spark.sql.files.openCostInBytes=1000000：降低小文件的打开成本权重，引导Spark合并小文件读取
   • spark.hadoop.fs.s3a.block.size=134217728（128MB）：匹配HDFS块大小，优化读取效率
2. 利用Spark自动分区发现：
   由于路径包含key=value格式的元数据，直接读取根路径让Spark自动识别分区，避免手动传入大量密钥：

   df = spark.read.parquet("s3a://my-bucket/")
   

   此方式会让Spark自动遍历路径、加载元数据，并优化文件读取的并行策略
3. 读取后合并重写：
   读取完成后，通过coalesce或repartition合并数据，写入大文件彻底解决小文件问题：

   # 根据集群资源调整合并后的分区数
   df.coalesce(100).write.mode("overwrite").parquet("s3a://my-bucket-consolidated/")
   

二、修复Spark+Boto3下载方案（若必须下载到PVC） #

现有代码存在3个关键问题导致下载不完整，修正后的代码如下：

import os
import time
import json
import boto3
from pyspark import TaskContext
from pyspark.sql import SparkSession
from concurrent.futures import ThreadPoolExecutor, as_completed

def process_files(broadcast_obj, iterator):     
    broadcast_vars = broadcast_obj.value
    local_dir = broadcast_vars["local_dir"]
    bucket_name = broadcast_vars["bucket_name"]
    endpoint_url = broadcast_vars["endpoint_url"]
    aws_access_key_id = broadcast_vars["aws_access_key_id"]
    aws_secret_access_key = broadcast_vars["aws_secret_access_key"]
    executor_id = TaskContext().get().taskAttemptId()
    error_keys = []
    
    # 初始化S3客户端
    s3 = boto3.client(
        "s3",
        endpoint_url=endpoint_url,
        aws_access_key_id=aws_access_key_id,
        aws_secret_access_key=aws_secret_access_key,
        config=boto3.session.Config(signature_version="s3v4"),
        verify=False,
    )
    
    start_time = time.time()
    key_count = 0
    error_count = 0
    
    # 线程池大小匹配执行器核数，避免过度并行导致的上下文切换
    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = []
        # 正确遍历迭代器中的密钥（原代码误用了外部变量）
        for key in iterator:
            local_path = os.path.join(local_dir, key)
            key_count += 1
            os.makedirs(os.path.dirname(local_path), exist_ok=True)
            # 直接传递参数给download_file，避免lambda变量捕获覆盖问题
            future = executor.submit(
                s3.download_file,
                bucket_name,
                key,
                local_path
            )
            futures.append((future, key))
        
        # 处理任务结果，记录失败的密钥而非异常对象
        for future, key in futures:
            try:
                future.result()
            except Exception as e:
                error_count += 1
                error_keys.append(key)
                print(f"下载失败 {key}: {str(e)}")
    
    # 写入执行日志
    with open(f"{local_dir}/timing_info_executor_{executor_id}_{int(time.time())}.txt", "w") as file:
        file.write(f"处理文件总数: {key_count}\n")
        file.write(f"失败数量: {error_count}\n")
        file.write(f"失败keys: {json.dumps(error_keys)}\n")
        file.write(f"总耗时: {time.time() - start_time:.2f} 秒\n")

if __name__ == '__main__':
    local_json_path = '/data/jsonfile.json'
    # 显式配置Spark执行器参数
    spark = SparkSession.builder \
        .appName("s3Downloader") \
        .config("spark.executor.instances", "20") \
        .config("spark.executor.memory", "1g") \
        .config("spark.executor.cores", "1") \
        .getOrCreate()
    
    broadcast_vars = spark.sparkContext.broadcast(
        {
            "endpoint_url": "objec_store_creds",  
            "aws_access_key_id": "objec_store_creds",
            "aws_secret_access_key": "objec_store_creds",
            "local_dir": '/data/temp',
            "bucket_name": 'my-bucket',
        }
    )
    
    # 加载S3密钥列表
    with open(local_json_path, "r") as read_content: 
        s3keys = json.load(read_content)
    
    # 计算分区数，确保为整数且至少为1
    num_partitions = max(1, len(s3keys) // 700)
    rdd = spark.sparkContext.parallelize(s3keys, num_partitions)
    
    # 执行下载任务
    rdd.foreachPartition(lambda p: process_files(broadcast_vars, p))
    
    spark.stop()

核心修正点：

• 修复lambda变量捕获覆盖问题，确保每个任务对应正确的密钥
• 正确遍历分区迭代器（原代码误用外部变量导致仅处理部分数据）
• 调整线程池大小匹配执行器核数，避免资源过载
• 修正分区数计算逻辑，确保为有效整数

方案选择建议 #

• 若无需本地PVC存储，优先选择优化Spark读取的方案：无需额外下载步骤，直接在Spark内完成数据合并，整体效率更高
• 若必须将文件下载到PVC，使用修正后的Boto3代码确保全量下载，后续可通过Spark或其他工具读取本地文件完成压缩合并

内容的提问来源于stack exchange，提问作者mr-price