Optionally expose the driver UI port as NodePort

docs/running-on-kubernetes.md

@@ -194,6 +194,14 @@ from the other deployment modes. See the [configuration page](configuration.html
     Time to wait for the driver pod to start running before aborting its execution.
   </td>
 </tr>
+<tr>
+  <td><code>spark.kubernetes.driver.service.exposeUiPort</code></td>
+  <td><code>false</code></td>
+  <td>
+    Whether to expose the driver Web UI port as a service NodePort. Turned off by default because NodePort is a limited
+    resource. Use alternatives such as Ingress if possible.
+  </td>
+</tr>
 </table>
 
 ## Current Limitations

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala

@@ -226,13 +226,15 @@ private[spark] class Client(
     logInfo("Successfully submitted local resources and driver configuration to" +
       " driver pod.")
     // After submitting, adjust the service to only expose the Spark UI
+    val uiServiceType = if (sparkConf.get(EXPOSE_KUBERNETES_DRIVER_SERVICE_UI_PORT)) "NodePort"
+      else "ClusterIP"
     val uiServicePort = new ServicePortBuilder()
       .withName(UI_PORT_NAME)
       .withPort(uiPort)
       .withNewTargetPort(uiPort)
       .build()
     kubernetesClient.services().withName(kubernetesAppId).edit().editSpec()
-      .withType("ClusterIP")
+      .withType(uiServiceType)
       .withPorts(uiServicePort)
       .endSpec()
       .done()

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/config.scala

@@ -156,6 +156,15 @@ package object config {
         .stringConf
         .createOptional
 
+  private[spark] val EXPOSE_KUBERNETES_DRIVER_SERVICE_UI_PORT =
+    ConfigBuilder("spark.kubernetes.driver.service.exposeUiPort")
+      .doc("""
+          | Whether to expose the driver Web UI port as a service NodePort. Turned off by default
+          | because NodePort is a limited resource. Use alternatives such as Ingress if possible.
+        """.stripMargin)
+      .booleanConf
+      .createWithDefault(false)
+
   private[spark] val KUBERNETES_DRIVER_POD_NAME =
     ConfigBuilder("spark.kubernetes.driver.pod.name")
       .doc("""