9

I need to upload a dataframe to S3 bucket but I do not have delete permissions on the bucket. Is there any way I can avoid creating this _temporary directory on S3? Maybe any way in spark to use local FS for _temporary directory and then uploading final resulting file to S3 bucket or totally avoid _temporary directory.

Thanks in advance.

2 Answers 2

12

No.

Data is written into _temporary/jobAttemptID/taskAttemptID/ and then renamed into the dest dir during task/job commit.

What you can do is write to hdfs for your jobs and then copy up using distcp. There are lots of advantages for this, not least being "with a consistent filesystem you don't run the risk of data loss you have from the s3n or s3a connectors"

2019-07-11 Update. The Apache Hadoop S3A committers let you commit work without the temp folder or rename, delivering performance and correct results even against an inconsistent S3 Store. This is how you can safely commit work. Amazon EMR have their own reimplementation of this own work, albeit (currently without the complete failure semantics which Spark expects

7
  • I see similar behavior during partitioned writes to hdfs, ie df.write.partitionBy(keys).parquet('/location') Are there alternatives here? Mar 2, 2018 at 23:28
  • the temp dirs are used so that tasks can run in parallel and failure is fixed by retrying. As rename() is fast and atomic on HDFS, its nothing to worry about
    – stevel
    Mar 5, 2018 at 13:06
  • Hello Steve, where can I learn more about these s3a committers? Do they work with spark 2.2 and spark 2.4? Sep 30, 2019 at 9:56
  • It says it here docs.cloudera.com/HDPDocuments/HDP3/HDP-3.1.4/… that it is only used for spark sql? Or is that link incorrect? Are there any limitation to s3a committers? Sep 30, 2019 at 9:58
  • they aren't in the shipping ASF spark releases; though they will be in spark 3.0. They are in all versions of hadoop and spark in HDP-3.x;
    – stevel
    Oct 1, 2019 at 12:17
11

Yes, you can avoid creating _temporary directory when uploading dataframe to s3.

When Spark appends data to an existing dataset, Spark uses FileOutputCommitter to manage staging output files and final output files.

By default, output committer algorithm uses version 1. In this version, FileOutputCommitter has two methods, commitTask and commitJob. commitTask moves data generated by a task from the task temporary directory to job temporary directory and when all tasks complete, commitJob moves data to from job temporary directory to the final destination.

However, when output committer algorithm uses version 2, commitTask moves data generated by a task directly to the final destination and commitJob is basically a no-op.

How do I set spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version to 2? You can set this config by using any of the following methods:

  • When you launch your cluster, you can put spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version 2 in the Spark config.
  • spark.conf.set("mapreduce.fileoutputcommitter.algorithm.version", "2")
  • When you write data using Dataset API, you can set it in the option, i.e. dataset.write.option("mapreduce.fileoutputcommitter.algorithm.version", "2").

Read more about the output committer algorithm versions databricks-blog and mapred-default

2
  • Thanks for the links but this did not work for me (Apache 2.3.3/Hadoop 2.7.3). It still creates the _temporary directory. When saving under "foo": foo/_temporary/0/_temporary/attempt_...000000_0/part-00000-....snappy.parquet However, the commitTask writes to task temporary and then moves that output directly to final dir (done by executor, parallelism, faster) The commitJob (done by driver?) then essentially is a no-op as explained in the Databricks blog.
    – medale
    Aug 26, 2019 at 0:51
  • 5
    v2 commit algorithm still needs a temp dir for each task attempt; this is how it can recover from a failed task attempt and support speculative execution. But: while job commit is now repeatable, a failure during a task commit can leave the output directory in an unknown state. Spark assume task commits are atomic and just retries here -really it should fail. The MR V2 algorithm is not safe; nor is the EMR "optimized committer" github.com/steveloughran/zero-rename-committer/releases
    – stevel
    Aug 26, 2019 at 10:48

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.