更改spark _temporary目录路径，避免删除parquets

Question

当两个或多个Spark作业具有相同的输出目录时，相互删除文件将不可避免。

我正在使用spark 2.4.4在append模式下写入一个数据帧，我想在spark的tmp目录中添加一个时间戳，以避免这些删除。

示例：

我用hdfs:/outputFile/0/tmp/file1.parquet编写的JobSpark

使用其他数据调用相同的spark作业，并在hdfs:/outputFil/0/tm/file2.parquet中写入

我希望jobSpark1在hdfs:/outputFile/0/tmp+(timeStamp)/file1.parquet中写入，另一个作业在hdfs:/outputFile/0/tmp+(timeStamp)/file2.parquet中写入，然后将包移动到hdfs:/outputFile/

Answer 1

df
  .write
        .option("mapreduce.fileoutputcommitter.algorithm.version", "2")
        .partitionBy("XXXXXXXX")
        .mode(SaveMode.Append)
        .format(fileFormat)
        .save(path)

当Spark向现有数据集追加数据时，Spark使用FileOutputCommitter来管理暂存输出文件和最终输出文件。FileOutputCommitter的行为直接影响写入数据的作业的性能。

FileOutputCommitter有两个方法: commitTask和commitJob。Apache Spark 2.0及更高版本使用Apache Hadoop2，它使用mapreduce.fileoutputcommitter.algorithm.version的值来控制commitTask和commitJob的工作方式。在Hadoop2中，mapreduce.fileoutputcommitter.algorithm.version的默认值为1，本版本中，commitTask会将任务产生的数据从任务临时目录移动到作业临时目录，当所有任务完成后，commitJob会将数据从作业临时目录移动到最终目的地。

由于驱动程序正在执行commitJob的工作，因此对于云存储，此操作可能需要很长时间。你可能经常认为你的手机是“挂着”的。但是，当mapreduce.fileoutputcommitter.algorithm.version的值为2时，commitTask会将任务生成的数据直接移动到最终目的地，而commitJob基本上是无操作的。