FileOutputCommitter中提交mapreduce.fileoutputcommitter.algorithm.version有v1和v2两个版本。

v1版本Spark写入文件的流程：

1.当task完成的时候，会将task的结果文件先写入到临时目录下面。

2.所有的task完成后，将所有的结果文件写入到结果目录。

3.删除临时目录，生成标记文件。

v1的弊端：第二步写入操作是在driver端进行的，而且是单线程进行。当结果文件数量很多的时候，耗时就会线性增加。

v2版本Spark写入文件的流程：

1.当task完成的时候，将task的结果文件直接写入结果目录。

2.所有的task完成后，删除临时目录，生成标记文件。

可以看的v2比v1少了一个rename的过程。写入结果目录是发生在Executor上，task是同时进行的，相当于多线程，速度更快。但是存在一致性问题，在所有task写入结果目录过程，用户可以看到部分数据。

FileOutputCommitter源码解析

对应的类是org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter

commit task

Executor端完成task会执行commitTask，可以看到当algorithmVersion == 1的时候会执行rename操作，将task结果文件到committedTaskPath（临时文件）。

algorithmVersion不是1的时候，使用mergePaths将结果文件直接写入到outputPath

commit job

在driver执行，当job完成（所有的task完成），执行commitJob

algorithmVersion == 1，执行mergePaths将结果文件直接写入到outputPath，这里就是性能瓶颈所在的地方，这里是单线程进行的，要是单个文件耗时20ms，1000个文件就是20000ms（20s），它是线性增加的。

algorithmVersion == 2的话，在commitTask中已经写入到结果目录了。

最后删除临时目录，生成标记文件。

和hive结合使用

上面提到使用v2的时候，很有可能出现结果目录中出现部分文件的情况，这要是hive的表目录就能在这个表读取到部分数据，这是不对的。应该保证原子性，要么读取的数据是旧的数据，要么是新的数据，不能读取部分数据。

入口类：org.apache.spark.sql.hive.execution.InsertIntoHiveTable

hive的临时目录生成

类似这种：{hive表数据目录}/.hive-staging_hive_2025-06-26_17-33-07_088_701862708150638596-1/-ext-10000/

processInsert

这里就一个简单的场景，就是写入到非分区表。

首先是先将数据写入到hive的临时目录中，在load到hive表中。

写入hive临时目录

可以看到commiter变量，这个就是上面的FileOutputCommitter，有两种提交方式。

runJob就是spark提交了任务，开始执行。这里执行task，task完成回调commitTask方法，runJob是阻塞的，会等到所有task完成。然后回调commitJob方法。根据上面FileOutputCommitter，此时数据已经全部写入了hive的临时目录下面了。

loadTable

loadTable是使用反射的方式执行_loadTableMethod_方法，对应就是Hive的loadTable方法。

hive的loadTable中如果是替换就是replaceFiles，否则是copyFiles。

replaceFiles

对应sparksql的 insert overwrite

首先将原有的目录下面的文件都放到回收站，即删除，再将结果文件从临时目录移动到正式目录。

moveFiles 移动结果文件。可以看到使用线程池（默认是25个线程）进行结果文件的重命名来移动文件的。不同于FileOutputCommitter的v1中单线程进行rename，Hive是多线程来rename的，效率更高。

copyFiles

对应sparksql的 insert into

可以看到它也是使用线程池进行并行操作的。

整体流程

v2流程，建议使用v2，效率更高。