SPARK-25003: Add helper methods to create new Extensions from Conf

Previously the only way to add extensions to the session was via the
getOrCreate method of the SparkSession Builder. To facilitate non-scala
Session creation we add a new constructor which takes in just the
context and Extensions. Then we also add a new Extensions constructor
which given a SparkConf generates an Extensions object with user config
already applied.

Author: RussellSpitzer

python/pyspark/sql/session.py

@@ -212,15 +212,17 @@ def __init__(self, sparkContext, jsparkSession=None):
         self._sc = sparkContext
         self._jsc = self._sc._jsc
         self._jvm = self._sc._jvm
+
         if jsparkSession is None:
             if self._jvm.SparkSession.getDefaultSession().isDefined() \
                     and not self._jvm.SparkSession.getDefaultSession().get() \
                         .sparkContext().isStopped():
                 jsparkSession = self._jvm.SparkSession.getDefaultSession().get()
             else:
-                jsparkSession = self._jvm.SparkSession.builder() \
-                    .sparkContext(self._jsc.sc()) \
-                    .getOrCreate()
+                extensions = self._sc._jvm.org.apache.spark.sql\
+                    .SparkSessionExtensions(self._jsc.getConf())
+                jsparkSession = self._jvm.SparkSession(self._jsc.sc(), extensions)
+
         self._jsparkSession = jsparkSession
         self._jwrapped = self._jsparkSession.sqlContext()
         self._wrapped = SQLContext(self._sc, self, self._jwrapped)

sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala

@@ -88,6 +88,10 @@ class SparkSession private(
     this(sc, None, None, new SparkSessionExtensions)
   }
 
+  private[sql] def this(sc: SparkContext, extensions: SparkSessionExtensions) {
+    this(sc, None, None, extensions)
+  }
+
   sparkContext.assertNotStopped()
 
   // If there is no active SparkSession, uses the default SQL conf. Otherwise, use the session's.
@@ -935,23 +939,7 @@ object SparkSession extends Logging {
           // Do not update `SparkConf` for existing `SparkContext`, as it's shared by all sessions.
         }
 
-        // Initialize extensions if the user has defined a configurator class.
-        val extensionConfOption = sparkContext.conf.get(StaticSQLConf.SPARK_SESSION_EXTENSIONS)
-        if (extensionConfOption.isDefined) {
-          val extensionConfClassName = extensionConfOption.get
-          try {
-            val extensionConfClass = Utils.classForName(extensionConfClassName)
-            val extensionConf = extensionConfClass.newInstance()
-              .asInstanceOf[SparkSessionExtensions => Unit]
-            extensionConf(extensions)
-          } catch {
-            // Ignore the error if we cannot find the class or when the class has the wrong type.
-            case e @ (_: ClassCastException |
-                      _: ClassNotFoundException |
-                      _: NoClassDefFoundError) =>
-              logWarning(s"Cannot use $extensionConfClassName to configure session extensions.", e)
-          }
-        }
+        SparkSessionExtensions.applyExtensionsFromConf(sparkContext.conf, extensions)
 
         session = new SparkSession(sparkContext, None, None, extensions)
         options.foreach { case (k, v) => session.initialSessionOptions.put(k, v) }

sql/core/src/main/scala/org/apache/spark/sql/SparkSessionExtensions.scala

@@ -19,10 +19,14 @@ package org.apache.spark.sql
 
 import scala.collection.mutable
 
+import org.apache.spark.SparkConf
 import org.apache.spark.annotation.{DeveloperApi, Experimental, InterfaceStability}
+import org.apache.spark.internal.Logging
 import org.apache.spark.sql.catalyst.parser.ParserInterface
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.StaticSQLConf
+import org.apache.spark.util.Utils
 
 /**
  * :: Experimental ::
@@ -66,6 +70,11 @@ class SparkSessionExtensions {
   type StrategyBuilder = SparkSession => Strategy
   type ParserBuilder = (SparkSession, ParserInterface) => ParserInterface
 
+  private[sql] def this(conf: SparkConf) {
+    this()
+    SparkSessionExtensions.applyExtensionsFromConf(conf, this)
+  }
+
   private[this] val resolutionRuleBuilders = mutable.Buffer.empty[RuleBuilder]
 
   /**
@@ -169,3 +178,29 @@ class SparkSessionExtensions {
     parserBuilders += builder
   }
 }
+
+object SparkSessionExtensions extends Logging {
+
+  /**
+   * Initialize extensions if the user has defined a configurator class in their SparkConf.
+   * This class will be applied to the extensions passed into this function.
+   */
+  private[sql] def applyExtensionsFromConf(conf: SparkConf, extensions: SparkSessionExtensions) {
+    val extensionConfOption = conf.get(StaticSQLConf.SPARK_SESSION_EXTENSIONS)
+    if (extensionConfOption.isDefined) {
+      val extensionConfClassName = extensionConfOption.get
+      try {
+        val extensionConfClass = Utils.classForName(extensionConfClassName)
+        val extensionConf = extensionConfClass.newInstance()
+          .asInstanceOf[SparkSessionExtensions => Unit]
+        extensionConf(extensions)
+      } catch {
+        // Ignore the error if we cannot find the class or when the class has the wrong type.
+        case e@(_: ClassCastException |
+                _: ClassNotFoundException |
+                _: NoClassDefFoundError) =>
+          logWarning(s"Cannot use $extensionConfClassName to configure session extensions.", e)
+      }
+    }
+  }
+}