如何确保所有Spark Worker停止后执行文件合并操作？

首先直接给你结论：把汇总操作放在try-with-resources块之后是正确的做法，但核心前提是要搞清楚Spark的Driver/Executor执行模型，避免把汇总逻辑误放到Worker（Executor）端执行。

接下来详细拆解你的问题和解决方案：

为什么你的担心是合理的？ #

如果错误地将FileUtil.copyMerge放在Spark的转换/行动算子（比如map、foreach）里，这段代码会被分发到所有Worker的Executor上并行执行，必然会导致多个进程同时写入同一个文件，造成覆盖或者文件损坏。而且如果任务还没完成就执行合并，还会读到未写完的临时文件，导致数据不完整。

正确的执行时机与位置 #

Spark的作业执行是由Driver进程主导的：

1. Driver负责提交作业，等待所有Executor（Worker上的进程）完成任务。
2. 你在try-with-resources块里初始化的SparkSession/SparkContext，会在块结束时自动关闭，而关闭前会确保所有提交的Spark任务都已经执行完成（行动算子是阻塞式的，会等到所有任务结束才返回）。

所以正确的代码结构应该是这样的（以Java为例）：

import org.apache.hadoop.fs.FileUtil;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.SaveMode;

public class DataMergeExample {
    public static void main(String[] args) {
        // try-with-resources管理SparkSession，自动关闭
        try (SparkSession spark = SparkSession.builder()
                .appName("DataMergeJob")
                .master("yarn") // 或者你的集群模式
                .getOrCreate()) {
            
            // 1. 在这里执行你的核心数据处理逻辑
            // 处理完成后，将数据输出到一个临时目录（不要直接输出到最终文件）
            spark.read()
                .format("parquet")
                .load("hdfs://input-path")
                .write()
                .mode(SaveMode.Overwrite)
                .parquet("hdfs://temp-output-dir");
                
        } // 到这里，SparkSession已关闭，所有Worker的任务100%完成，临时目录的文件都是完整的
        
        // 2. 在Driver端执行合并操作，仅执行一次
        try {
            org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
            // 合并临时目录的所有文件到最终文件
            FileUtil.copyMerge(
                org.apache.hadoop.fs.FileSystem.get(conf),
                new org.apache.hadoop.fs.Path("hdfs://temp-output-dir"),
                org.apache.hadoop.fs.FileSystem.get(conf),
                new org.apache.hadoop.fs.Path("hdfs://final-output-file"),
                false,
                conf,
                null
            );
            // 可选：合并完成后删除临时目录
            org.apache.hadoop.fs.FileSystem.get(conf).delete(new org.apache.hadoop.fs.Path("hdfs://temp-output-dir"), true);
        } catch (Exception e) {
            e.printStackTrace();
            // 处理合并失败的异常
        }
    }
}

关键注意点 #

• 必须在Driver端执行合并：copyMerge的代码要写在Spark算子之外，确保只有Driver进程执行一次，避免多Worker并发写入。
• 临时目录的必要性：不要直接让Spark输出单个文件（比如coalesce(1)），大数据量下会把所有数据压到一个Executor，容易OOM；用临时目录输出多文件，再合并更高效。
• 异常处理：如果Spark作业执行失败，try-with-resources块会提前退出，这时候不要执行合并，避免处理不完整的数据。
• 文件完整性：Spark的输出机制会确保任务完成后才会生成最终的输出文件（临时文件会被重命名），所以try块结束时，临时目录里的文件都是完整可用的，不会有未完成的文件。

替代方案（可选） #

如果你不想手动调用FileUtil.copyMerge，也可以使用Spark的saveAsTextFile（针对RDD）或者write().text()（针对DataFrame）配合coalesce(1)，但再次强调：这种方式会将所有数据 shuffle 到一个Executor，仅适合小数据量场景；大数据量下，copyMerge是更优的选择。

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