Apache Spark 2.0 jobs take a long time to finish when they are finished

Phenomenon

When using Apache Spark 2.x, you may encounter this phenomenon: although our Spark Jobs have all been completed, our program is still executing. For example, we use Spark SQL to execute some SQL, which generates a large number of files in the end. Then we can see that all Spark Jobs of this SQL have actually been run to completion, but this query statement is still running. From the log, we can see that the driver node is moving the files generated by tasks to the directory of the final table one by one. This phenomenon is easy to occur when our job generates a lot of files. This article will introduce a method to solve this problem.

Why does this phenomenon occur?

Spark 2.x uses Hadoop 2.x. When it saves the generated file to HDFS, it finally calls saveAsHadoopFile, which uses FileOutputCommitter, as follows:

The problem lies in the FileOutputCommitter implementation of Hadoop 2.x. There are two noteworthy methods in FileOutputCommitter: commitTask and commitJob. In the FileOutputCommitter implementation of Hadoop 2.x, the mapreduce.fileoutputcommitter.algorithm.version parameter controls how commitTask and commitJob work. The specific code is as follows (for the sake of convenience, I removed irrelevant statements. The complete code can be found in FileOutputCommitter.java):

As you can see, in the commitTask method, there is a conditional judgment algorithmVersion == 1, which is the value of the mapreduce.fileoutputcommitter.algorithm.version parameter, which defaults to 1. If this parameter is 1, then when the Task is completed, the data temporarily generated by the Task will be moved to the corresponding directory of the task, and then moved to the final job output directory when commitJob is called. The default value of this parameter in Hadoop 2.x is 1! This is why we see that the job is completed, but the program is still moving data, resulting in the entire job not being completed. In the end, the commitJob function is executed by Spark's Driver, so there is a reason for the slow execution.

And we can see that if we set the value of the mapreduce.fileoutputcommitter.algorithm.version parameter to 2, then when commitTask is executed, the mergePaths method will be called to directly move the data generated by the Task from the Task temporary directory to the directory finally generated by the program. When executing commitJob, there is no need to move data directly, so it will naturally be much faster than the default value.

Note that in versions prior to Hadoop 2.7.0, we can achieve this by setting the mapreduce.fileoutputcommitter.algorithm.version parameter to a value other than 1, because the program does not limit this value to 2. However, as of Hadoop 2.7.0, the value of the mapreduce.fileoutputcommitter.algorithm.version parameter must be 1 or 2. For details, see MAPREDUCE-4815.

How to set this parameter in Spark

The problem has been found, and we can solve it in the program. There are several ways:

• Set spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version 2 directly in conf/spark-defaults.conf . This has a global impact.
• Set it directly in the Spark program, spark.conf.set("mapreduce.fileoutputcommitter.algorithm.version", "2"), this is at the job level.
• If you are using the Dataset API to write data to HDFS, you can set dataset.write.option("mapreduce.fileoutputcommitter.algorithm.version", "2") .

However, if your Hadoop version is 3.x, the default value of the mapreduce.fileoutputcommitter.algorithm.version parameter is already set to 2. For details, see MAPREDUCE-6336 and MAPREDUCE-6406.

Because this parameter has some impact on performance, in Spark 2.2.0, this parameter has been recorded in the Spark configuration document configuration.html , see SPARK-20107 for details.

Summarize

The above is what I introduced to you about Apache Spark 2.0. I hope it will be helpful to you!