Spark+Scala：高效合并多DataFrame的性能优化问询

当前代码的核心问题：

• 使用Scala本地并行集合（par.map/par.reduce）处理文件列表，所有文件解析逻辑集中在Driver节点，无法利用Spark集群的Executor资源，导致CPU利用率极低；同时Driver内存压力过大，容易引发集群调度异常。
• 频繁调用union合并小DataFrame，生成了极长的执行 lineage（报错中的stage 7718就是明证），Spark调度器需要处理数千次stage切换，最终导致Executor崩溃。

优化方案 #

1. 改用Spark分布式文件处理逻辑 #

将文件列表转为Spark分布式数据集，让Executor并行处理文件，彻底卸载Driver节点的压力：

def fileListToDataframe(flist: String): DataFrame = {
    // 读取文件列表为Spark分布式DataFrame
    val filePaths = spark.read.textFile(flist)
    
    // 提前获取目标Schema（从第一个文件解析，避免重复推导）
    val sampleFilePath = filePaths.take(1).head
    val sampleDF = fileToDF(sampleFilePath)
    val targetSchema = sampleDF.schema
    
    // 分布式处理每个文件，直接返回Row迭代器
    val resultDF = filePaths.flatMap { filePath =>
        // 在Executor端解析文件，返回Row集合的迭代器
        val df = fileToDF(filePath)
        df.collect().iterator
    }.toDF(targetSchema)
    
    resultDF
}

注意：确保NetCDF-java的依赖包已上传到所有Executor节点（通过--jars参数或者集群依赖管理）。

2. 避免频繁Union，使用批量写入合并 #

如果处理10K文件时仍有调度压力，可以分批次处理并写入临时存储（如Parquet），最后合并结果：

def fileListToDataframe(flist: String): DataFrame = {
    val ipFiles = scala.io.Source.fromFile(flist).mkString.split("\n").toList
    val batchSize = 1000 // 每批处理1000个文件
    val sampleDF = fileToDF(ipFiles.head)
    val targetSchema = sampleDF.schema
    
    // 分批次处理并写入临时Parquet文件
    ipFiles.grouped(batchSize).zipWithIndex.foreach { case (batch, idx) =>
        val batchDF = spark.createDataset(batch).flatMap { filePath =>
            fileToDF(filePath).collect().iterator
        }.toDF(targetSchema)
        
        // 写入临时目录，使用append模式
        batchDF.write.mode("append").parquet("/tmp/netcdf_temp_data")
    }
    
    // 读取所有临时文件合并为最终DataFrame
    spark.read.parquet("/tmp/netcdf_temp_data")
}

Parquet格式的读写效率远高于原生DataFrame合并，且能避免 lineage过长的问题。

3. 优化Spark集群配置 #

针对大文件数量场景，调整以下关键配置：

• Driver内存：--driver-memory 32g（根据实际情况调整，避免Driver内存溢出）
• Executor资源：--executor-memory 16g --executor-cores 4（让每个Executor能高效处理多个文件，提升CPU利用率）
• Shuffle分区数：spark.sql.shuffle.partitions=200（设置为Executor总核心数的2-3倍，避免小分区过多）
• 自适应执行：spark.sql.adaptive.enabled=true（让Spark自动优化执行计划，合并小任务）
• Executor日志配置：spark.executor.extraJavaOptions="-XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/tmp/heapdump"（用于排查OOM问题）

4. 优化fileToDF函数 #

• 严格只读取NetCDF文件中必要的变量和维度，避免加载冗余数据到内存；
• 确保NetCDF文件资源被正确关闭，避免Executor端文件句柄泄漏：

def fileToDF(filePath: String): DataFrame = {
    var ncFile: NetcdfFile = null
    try {
        ncFile = NetcdfFiles.open(filePath)
        // 仅读取需要的变量，解析为Row
        // ... 你的解析逻辑 ...
    } finally {
        if (ncFile != null) ncFile.close() // 强制关闭文件
    }
}

5. 排查Executor退出原因 #

报错中Executor退出代码为0，可能是资源泄漏或静默OOM：

• 查看Executor的stderr日志，定位具体异常；
• 检查fileToDF中是否有未释放的资源（如文件句柄、内存缓冲区）；
• 启用堆转储分析，确认是否存在内存泄漏。

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