[SPARK-6352] [SQL] Add DirectParquetOutputCommitter

[ypcat]

Add a DirectParquetOutputCommitter class that skips _temporary directory when saving to s3. To enable it, set hadoop configuration "spark.sql.parquet.output.committer.class" to "org.apache.spark.sql.parquet.DirectParquetOutputCommitter".

[ash211]

Jenkins this is ok to test

[SparkQA]

Test build #28650 has finished for PR 5042 at commit 769bd67.

• This patch fails RAT tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • class DirectParquetOutputCommitter(outputPath: Path, context: TaskAttemptContext)

[shaneknapp]

jenkins, test this please

[SparkQA]

Test build #28653 has finished for PR 5042 at commit 769bd67.

• This patch fails RAT tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • class DirectParquetOutputCommitter(outputPath: Path, context: TaskAttemptContext)

[SparkQA]

Test build #28709 has finished for PR 5042 at commit c42468c.

• This patch fails RAT tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #28712 has finished for PR 5042 at commit 0d540b9.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[ypcat]

I change the design to allow more general usage. User can set "spark.sql.parquet.output.committer.class" to a class extending ParquetOutputFormat.
I still include the implementation of DirectParquetOutputCommitter, if it is unwanted, we can move it to be only used in test case.

[SparkQA]

Test build #28993 has finished for PR 5042 at commit e17bf47.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #28992 has finished for PR 5042 at commit 9ae7545.

• This patch passes all tests.
• This patch does not merge cleanly.
• This patch adds no public classes.

[liancheng]

Hey @ypcat, sorry for the late review. If I understand this correctly, with DirectParquetOutputCommitter, all tasks writes to the final output directory directly. Then what if speculative tasks are scheduled? This may end up with two tasks trying to write to the same file simultaneously, right?

[liancheng]

At least we should state explicitly that speculative tasks should be disabled when using the direct output committer.

[JoshRosen]

@liancheng, if this is intended only for writing directly to S3, then I think we purposely want to avoid writing to _temporary because the move / rename of the temporary file to the final file is expensive and because S3 PUT operations are atomic. If you have two tasks racing to write to the same file, then the last writer should win and we shouldn't have to worry about partially-written files. As with DirectOutputCommitter, this is not safe for use on filesystems where file creation / overwrite isn't atomic, such as HDFS, but that's not a use-case that we have to worry about here (I think this is only for use by advanced users who know that they're writing directly to S3).

[ypcat]

@liancheng this DirectParquetOutputCommitter is based on DirectOutputCommitter in this thread and was intended to use on s3. I was not aware of the problem of speculation. I think we can either:

1. exclude DirectParquetOutputCommitter and let user implement their own like DirectOutputCommitter.
2. write down the consequences in documentation as @JoshRosen described.

[liancheng]

Should we check for mapreduce.output.fileoutputformat.outputdir instead?

[liancheng]

Oh I see, this is for Hadoop 1 compatibility.

[liancheng]

@JoshRosen @ypcat Thanks for the detailed explanation, I didn't realize S3 PUT operations are atomic at first. This LGTM. Merging to master.

@ypcat Thanks again for working on this!

[JoshRosen]

We can't change this now, but note that the commit message / PR description here is slightly out of date w.r.t. the actual changes being committed, since we now use the spark.sql.parquet.output.committer.class configuration instead of spark.sql.parquet.useDirectParquetOutputCommitter.

We may want to document this configuration at some point; if we do, I think the right place is the Parquet Configuration section in the SQL programming guide: https://spark.apache.org/docs/latest/sql-programming-guide.html#configuration

[JoshRosen]

Actually, on giving this a closer look I'm not sure whether this faithfully respects all of the Parquet configurations for controlling OutputCommitter behavior. For instance, it looks like there's a ParquetOutputFormat.ENABLE_JOB_SUMMARY configuration that can be used to suppress the writing of the summary file (https://github.com/Parquet/parquet-mr/blame/fa8957d7939b59e8d391fa17000b34e865de015d/parquet-hadoop/src/main/java/parquet/hadoop/ParquetOutputCommitter.java#L48). I don't really know much about Parquet, so I can't really speak to whether this is a configuration that would be commonly used, but we should probably try to support it.

[ypcat]

@JoshRosen I updated PR description. It looks like the parquet code base I used was too old that it did not include the ParquetOutputFormat.ENABLE_JOB_SUMMARY thing. I will add support to this.

[JoshRosen]

It looks like this patch broke the Hadoop 1.x build because the test code uses Configuration.unset, which isn't present in Hadoop 1.x:

[info] Compiling 1 Scala source to /home/jenkins/workspace/Spark-Master-SBT/AMPLAB_JENKINS_BUILD_PROFILE/hadoop1.0/label/centos/examples/target/scala-2.10/classes...
[error] /home/jenkins/workspace/Spark-Master-SBT/AMPLAB_JENKINS_BUILD_PROFILE/hadoop1.0/label/centos/sql/core/src/test/scala/org/apache/spark/sql/parquet/ParquetIOSuite.scala:402: value unset is not a member of org.apache.hadoop.conf.Configuration
[error]       configuration.unset("spark.sql.parquet.output.committer.class")
[error]                     ^
[error] one error found

To unbreak the build, I'm going to revert this patch and re-open the JIRA. Let's open a new PR to fix this. At the same time, we can also address the ENABLE_JOB_SUMMARY flag.

[ypcat]

I cannot find a way to unset a config value in hadoop 1.x API. The closest thing is to set it to a default value, which I think should be fine in test code.
And I found I cannot add more commits to this PR since it is closed. Should we reopen it or use a new PR?

[ash211]

I think it's best to open a new PR

[ypcat]

New PR is #5525