Make FileOutputCommiter as default (#86)

salesforce · Aug 25, 2018 · e515e32 · e515e32
1 parent cf812f7
commit e515e32
Show file tree

Hide file tree

Showing 7 changed files with 3 additions and 149 deletions.
diff --git a/build.gradle b/build.gradle
@@ -197,10 +197,8 @@ configure(allProjs) {
         ignoreFailures = true
         include '**/*.java', '**/*.scala'
         exclude '**/org/apache/spark/ml/SparkDefaultParamsReadWrite.scala',
-                '**/com/salesforce/op/utils/io/DirectMapreduceOutputCommitter.scala',
                 '**/com/salesforce/op/test/TestSparkContext.scala',
                 '**/com/salesforce/op/test/TempDirectoryTest.scala',
-                '**/com/salesforce/op/utils/io/DirectOutputCommitter.scala',
                 '**/com/salesforce/op/stages/impl/tuning/OpCrossValidation.scala',
                 '**/com/salesforce/op/stages/impl/tuning/OpTrainValidationSplit.scala',
                 '**/com/salesforce/op/test/*.java',

diff --git a/gradle/spark.gradle b/gradle/spark.gradle
@@ -70,8 +70,7 @@ task sparkSubmit(type: Exec, dependsOn: copyLog4jToSpark) {
             "spark.hadoop.avro.output.codec=deflate",
             "spark.hadoop.avro.mapred.deflate.level=6",
             "spark.hadoop.validateOutputSpecs=false",
-            "spark.hadoop.mapred.output.committer.class=com.salesforce.op.utils.io.DirectOutputCommitter",
-            "spark.hadoop.spark.sql.sources.outputCommitterClass=com.salesforce.op.utils.io.DirectMapreduceOutputCommitter"
+            "spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version=2"
     ].collect { ["--conf", it] }.flatten()
 
     environment SPARK_HOME: sparkHome

diff --git a/helloworld/gradle/spark.gradle b/helloworld/gradle/spark.gradle
@@ -70,8 +70,7 @@ task sparkSubmit(type: Exec, dependsOn: copyLog4jToSpark) {
             "spark.hadoop.avro.output.codec=deflate",
             "spark.hadoop.avro.mapred.deflate.level=6",
             "spark.hadoop.validateOutputSpecs=false",
-            "spark.hadoop.mapred.output.committer.class=com.salesforce.op.utils.io.DirectOutputCommitter",
-            "spark.hadoop.spark.sql.sources.outputCommitterClass=com.salesforce.op.utils.io.DirectMapreduceOutputCommitter"
+            "spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version=2"
     ].collect { ["--conf", it] }.flatten()
 
     environment SPARK_HOME: sparkHome

diff --git a/templates/simple/spark.gradle b/templates/simple/spark.gradle
@@ -197,8 +197,7 @@ task sparkSubmit(dependsOn: copyLog4jToSparkNoInstall) {
                 "spark.hadoop.avro.output.codec=deflate",
                 "spark.hadoop.avro.mapred.deflate.level=6",
                 "spark.hadoop.validateOutputSpecs=false",
-                "spark.hadoop.mapred.output.committer.class=com.salesforce.op.utils.io.DirectOutputCommitter",
-                "spark.hadoop.spark.sql.sources.outputCommitterClass=com.salesforce.op.utils.io.DirectMapreduceOutputCommitter"
+                "spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version=2"
         ].collect { ["--conf", it] }.flatten()
 
         def hadoopConfDir = System.env.HOME + "/.fake_hadoop_conf"

diff --git a/utils/src/main/scala/com/salesforce/op/utils/io/DirectMapreduceOutputCommitter.scala b/utils/src/main/scala/com/salesforce/op/utils/io/DirectMapreduceOutputCommitter.scala
diff --git a/utils/src/main/scala/com/salesforce/op/utils/io/DirectOutputCommitter.scala b/utils/src/main/scala/com/salesforce/op/utils/io/DirectOutputCommitter.scala
diff --git a/utils/src/main/scala/com/salesforce/op/utils/io/avro/AvroInOut.scala b/utils/src/main/scala/com/salesforce/op/utils/io/avro/AvroInOut.scala
@@ -32,7 +32,6 @@ package com.salesforce.op.utils.io.avro
 
 import java.net.URI
 
-import com.salesforce.op.utils.io.DirectOutputCommitter
 import com.salesforce.op.utils.spark.RichRDD._
 import org.apache.avro.Schema
 import org.apache.avro.generic.GenericRecord
@@ -156,7 +155,6 @@ object AvroInOut {
 
     private def createJobConfFromContext(schema: String)(implicit sc: SparkSession) = {
       val jobConf = new JobConf(sc.sparkContext.hadoopConfiguration)
-      jobConf.setOutputCommitter(classOf[DirectOutputCommitter])
       AvroJob.setOutputSchema(jobConf, new Schema.Parser().parse(schema))
       jobConf
     }