There is no problem with spark-sql writing to the object storage path (the abnormal path disappears automatically)

1. Problem analysis

1, environment
spark3.2 hadoop3.2.2

2, Problem phenomenon
When inserting overwrite table to the hive table, an error is reported that the path does not exist, causing the task to fail.

When the path to the table is on hdfs, there is no problem. There will be problems when the path of the table is on the object store.

Insert overwrite table with partition path is no problem.

Specific error:

org.apache.spark.sql.AnalysisException: Path does not exist: s3a://xxxx/hive/warehouse/tablename
                 at org.apache.spark.sql.errors.QueryCompilationErrors$.dataPathNotExistError(QueryCompilationErrors.scala:978)
                 at org.apache.spark.sql.execution.datasources.DataSource$.$anonfun$checkAndGlobPathIfNecessary$4(DataSource.scala:780)
                 at org.apache.spark.sql.execution.datasources.DataSource$.$anonfun$checkAndGlobPathIfNecessary$4$adapted(DataSource.scala:777)

3. Analysis of the cause of the problem
After analyzing the full task logs, it was found that spark am failed twice.

The first failure was caused by an abnormal data conversion in the task, which is a business exception. However, when the fault-tolerant mechanism of yarn starts spark am for the second time, an exception of Path does not exist is reported.

In the execution logic of spark inseroverwrite, the path of the table is deleted first, and then written. And the action of deleting the path is done on the driver side. The re-creation of the table path is done when the task is executed. So the problem is why the task does not create the object storage table path again. First sort out the writing logic of the task (the algorithm of fileoutcommitter version 1.0 is used by default). Before the task writes data, a temporary directory will be created and the data will be written into the temporary directory.

When a task is successfully executed, the data is transferred from the temporary directory to the intermediate directory through the commit task logic of the commiter. When all tasks are successfully completed, the driver will call the commit job logic of the commiter to transfer the data in the intermediate directory to the final path, which is the table path. It should be noted that if the directory or path involved in the above process does not exist, it will be created automatically.

For the table whose table path is on hdfs, the temporary directory and the intermediate directory are all under the table path. So even if the task has a business exception, the temporary directory has been created before the exception occurs, and the path of the table has also been created. When am is executed for the second time, the Path does not exist error will not be reported.

For tables whose table path is in object storage, the temporary directory is a local directory, similar to /xx/xxx. Therefore, the task has a business exception, and the table path is not created. Why is the temporary directory a local directory, because the task uses the s3a protocol to commit task and commit job.

So the problem is on the s3a protocol.

4. Problem solution
Do not use the s3a protocol to execute insert overwrite table.

set spark.sql.sources.commitProtocolClass=org.apache.spark.sql.execution.datasources.SQLHadoopMapReduceCommitProtocol;
set spark.sql.sources.outputCommitterClass=;

Note: If the s3a protocol is not used, the commit performance will be seriously affected, because the mv or rename of the object storage is copied first and then deleted.

3.parquet file write optimization

set spark.sql.parquet.fs.optimized.committer.optimization-enabled=true;

The actual commitTask details of FileOutputCommitter are related to the parameter mapreduce.fileoutputcommitter.algorithm.version (the default value is 1).

When mapreduce.fileoutputcommitter.algorithm.version=1:

The operation of commit is to rename ${output.dir}/_temporary/${appAttemptId}/_temporary/${taskAttemptId} to ${output.dir}/_temporary/${appAttemptId}/${taskId}

When mapreduce.fileoutputcommitter.algorithm.version=2:

The operation of commit is to move the files under ${output.dir}/_temporary/${appAttemptId}/_temporary/${taskAttemptId} to the ${output.dir} directory (that is, the final output directory)

The spark task can enable the commit logic of version 2 by setting the spark configuration spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version=2
Before hadoop 2.7.0, the implementation of FileOutputCommitter did not distinguish between versions, and the commit logic of version=1 was uniformly used. Therefore, if the hadoop dependency package version of spark is lower than 2.7.0, it is useless to set mapreduce.fileoutputcommitter.algorithm.version=2

2. Detailed introduction to commit principle

When Spark outputs data to HDFS, the following problems need to be solved:

• Since multiple tasks write data to HDFS at the same time, how to ensure that all files written by all tasks are either visible to the outside world at the same time, or not visible to the outside world at the same time, that is, to ensure data consistency
• The same Task may have two identical Task instances to write the same data to HDFS because of Speculation, how to ensure that only one commit is successful
• For large jobs (such as tens of thousands or even hundreds of thousands of tasks), how to efficiently manage all files

commit principle

This article explains the principle of commit in detail by executing the following Spark program in Local mode

sparkContext.textFile("/json/input.zstd")
  .map(_.split(","))
  .saveAsTextFile("/jason/test/tmp")

Before describing the principle of commit in detail, several predicates need to be explained

• Task, that is, a Task of a Stage in a Job of an Application
• TaskAttempt, each execution of Task is regarded as a TaskAttempt. For the same Task, multiple TaskAttemps may exist at the same time
• Application Attempt, that is, an execution of Application

In this article, the following abbreviations will be used

• ${output.dir.root} is the output directory root path
• ${appAttempt} is the Application Attempt ID, which is an integer and starts from 0
• ${taskAttemp} is the Task Attetmp ID, which is an integer and starts from 0

Check Job Output Directory

Before starting the Job, the Driver first checks whether the output directory already exists through the checkOutputSpecs method of FileOutputFormat. If it already exists, directly throw FileAlreadyExistsException

Driver executes setupJob

Before the job starts, the Driver (this example uses local mode, so it is executed by the main thread) calls FileOuputCommitter.setupJob to create the Application Attempt directory, namely output.dir.root/temporary/xxxx/{appAttempt}

Task executes setupTask

The FileOutputCommitter.setupTask method is executed by each Task (this example uses local mode, so it is executed by the task thread). This method does nothing because the Task temporary directory is created by the Task on demand.

Create Task directory on demand

In this example, the Task needs to create a LineRecordWriter through the getRecordWriter method of TextOutputFormat to write data. Before creating, you need to set the Task output path through FileOutputFormat.getTaskOutputPath, that is, ${output.dir.root}/_temporary/${appAttempt}/_temporary/${taskAttempt}/${fileName }. All data of this Task Attempt are written in the files under this directory

Check if you need commit

After the Task execution data is written, check whether it needs to commit through the FileOutputCommitter.needsTaskCommit method. It is written under ${output.dir.root}/_temporary/${appAttempt}/_temporary/${taskAttempt} data.

The check is based on whether the ${output.dir.root}/_temporary/${appAttempt}/_temporary/${taskAttempt} directory exists

If commit is required and Output commit coordination is enabled, the OutputCommitCoordinator on the Driver side needs to use RPC to determine whether the Task Attempt can be committed

The reason why the CommitCoordinator on the Driver side needs to judge whether to commit is because there may be cases where multiple Attemps of the same Task write data at the same time and apply for commit due to speculation or other reasons (such as the previous TaskAttemp not being Killed successfully).

CommitCoordinator

When the TaskAttempt applying for commitTask is a failed Attempt, it will be rejected directly

If the TaskAttempt succeeds, and the CommitCoordinator has not allowed the commit request of other Attempts of the Task, the commit request of the TaskAttempt is allowed

If the CommitCoordinator has allowed the TaskAttempt’s commit request before, it will continue to agree to the TaskAttempt’s commit request, that is, the CommitCoordinator’s processing of the application is idempotent.

If the TaskAttempt succeeds, and the CommitCoordinator has previously allowed the commit requests of other Attempts of the Task, the commit request of the current TaskAttempt is directly rejected

In order to achieve the above functions, OutputCommitCoordinator maintains a StageState for each ActiveStage as follows

private case class StageState(numPartitions: Int) {
  val authorizedCommitters = Array.fill[TaskAttemptNumber](numPartitions)(NO_AUTHORIZED_COMMITTER)
  val failures = mutable.Map[PartitionId, mutable.Set[TaskAttemptNumber]]()
  }

In this data structure, the TaskAttempt that each Task is allowed to commit is saved. The default is NO_AUTHORIZED_COMMITTER

At the same time, all failed Attempts for each Task are saved

commitTask

When the TaskAttempt is allowed to commit, the Task (in this case, it is executed by the task thread because it uses the local model) will commitTask in the following way.

When the value of mapreduce.fileoutputcommitter.algorithm.version is 1 (the default value), the Task will assign taskAttemptPath to ${output.dir.root}/_temporary/${appAttempt}/_temporary/ ${taskAttempt} The command is committedTaskPath, which is ${output.dir.root}/_temporary/${appAttempt}/${taskAttempt}

If the value of mapreduce.fileoutputcommitter.algorithm.version is 2, directly set taskAttemptPath as ${output.dir.root}/_temporary/${appAttempt}/_temporary/${taskAttempt}< All files in /code> are moved to outputPath, which is ${output.dir.root}/