create SparkHadoopWriterConfig directly in PairRDDFunctions.

Author: jiangxb1987

core/src/main/scala/org/apache/spark/internal/io/SparkHadoopWriterConfig.scala

@@ -82,27 +82,3 @@ abstract class SparkHadoopWriterConfig[K, V: ClassTag] extends Serializable {
   def checkOutputSpecs(jobContext: JobContext): Unit
 
 }
-
-object SparkHadoopWriterConfig {
-
-  /**
-   * Instantiates a SparkHadoopWriterConfig using the given configuration.
-   */
-  def instantiate[K, V](className: String, conf: Configuration)(
-      implicit ctorArgTag: ClassTag[(K, V)]): SparkHadoopWriterConfig[K, V] = {
-    val clazz = Utils.classForName(className).asInstanceOf[Class[SparkHadoopWriterConfig[K, V]]]
-
-    // First try the one with argument (conf: SerializableConfiguration).
-    // If that doesn't exist, try the one with (conf: SerializableJobConf).
-    try {
-      val ctor = clazz.getDeclaredConstructor(
-        classOf[SerializableConfiguration], classOf[ClassTag[(K, V)]])
-      ctor.newInstance(new SerializableConfiguration(conf), ctorArgTag)
-    } catch {
-      case _: NoSuchMethodException =>
-        val ctor = clazz.getDeclaredConstructor(
-          classOf[SerializableJobConf], classOf[ClassTag[(K, V)]])
-        ctor.newInstance(new SerializableJobConf(conf.asInstanceOf[JobConf]), ctorArgTag)
-    }
-  }
-}

core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala

@@ -39,7 +39,7 @@ import org.apache.spark.internal.io._
 import org.apache.spark.internal.Logging
 import org.apache.spark.partial.{BoundedDouble, PartialResult}
 import org.apache.spark.serializer.Serializer
-import org.apache.spark.util.Utils
+import org.apache.spark.util.{SerializableConfiguration, SerializableJobConf, Utils}
 import org.apache.spark.util.collection.CompactBuffer
 import org.apache.spark.util.random.StratifiedSamplingUtils
 
@@ -1051,9 +1051,7 @@ class PairRDDFunctions[K, V](self: RDD[(K, V)])
    * configured for a Hadoop MapReduce job.
    */
   def saveAsNewAPIHadoopDataset(conf: Configuration): Unit = self.withScope {
-    val config = SparkHadoopWriterConfig.instantiate[K, V](
-      className = classOf[SparkHadoopMapReduceWriterConfig[K, V]].getName,
-      conf = conf)
+    val config = new SparkHadoopMapReduceWriterConfig[K, V](new SerializableConfiguration(conf))
     SparkHadoopWriter.write(
       rdd = self,
       config = config)
@@ -1066,9 +1064,7 @@ class PairRDDFunctions[K, V](self: RDD[(K, V)])
    * MapReduce job.
    */
   def saveAsHadoopDataset(conf: JobConf): Unit = self.withScope {
-    val config = SparkHadoopWriterConfig.instantiate[K, V](
-      className = classOf[SparkHadoopMapRedWriterConfig[K, V]].getName,
-      conf = conf)
+    val config = new SparkHadoopMapRedWriterConfig[K, V](new SerializableJobConf(conf))
     SparkHadoopWriter.write(
       rdd = self,
       config = config)