feat(spark)!: delegate SparkApplication driver service account to the Spark Operator

docs/source/spark/index.rst

@@ -52,13 +52,11 @@ using ``num_executors`` and ``resources_per_executor``.
 
 .. note::
 
-   Batch job submission requires the ``spark-operator-spark`` ServiceAccount to
-   exist in the target namespace, with the required SparkApplication RBAC
-   permissions bound to it. Otherwise, ``submit_job()`` requests will fail.
-
-   This is a current Spark Operator requirement and is expected to be simplified
-   once `kubeflow/spark-operator#3049 <https://github.com/kubeflow/spark-operator/issues/3049>`_
-   is resolved.
+   Batch job submission relies on the Spark Operator's fallback ServiceAccount
+   (``spark-operator-spark`` with the standard Helm install) having the
+   required SparkApplication RBAC permissions in the target namespace.
+   ``SparkClient`` does not set a ServiceAccount on the driver spec, so this
+   operator-configured fallback is always used for batch jobs.
 
 Two Ways to Run Spark
 -----------------------

examples/spark/README.md

@@ -35,7 +35,7 @@ Install spark dependencies:
 uv pip install kubeflow[spark]
 ```
 
-The Spark examples run against a Kubernetes cluster with the Spark Operator installed. Batch job submission requires a `spark-operator-spark` ServiceAccount in the target namespace with the required SparkApplication RBAC permissions. See the [Spark SDK docs](https://sdk.kubeflow.org/en/latest/spark/index.html) for prerequisites.
+The Spark examples run against a Kubernetes cluster with the Spark Operator installed. Batch job submission relies on the Spark Operator's fallback ServiceAccount (`spark-operator-spark` by default with the standard Helm install) having the required SparkApplication RBAC permissions in the target namespace; the SDK does not set a ServiceAccount on the driver spec, so this operator-configured fallback is always used. See the [Spark SDK docs](https://sdk.kubeflow.org/en/latest/spark/index.html) for prerequisites.
 
 ## Running Examples
 

kubeflow/spark/backends/kubernetes/constants.py

@@ -51,7 +51,6 @@
 DEFAULT_DRIVER_MEMORY = "512Mi"
 DEFAULT_EXECUTOR_CPU = 1
 DEFAULT_EXECUTOR_MEMORY = "512Mi"
-DEFAULT_SERVICE_ACCOUNT = "spark-operator-spark"
 
 # Function-based Spark job script
 FUNC_JOB_VOLUME_NAME = "spark-app-source"

kubeflow/spark/backends/kubernetes/utils.py

@@ -670,6 +670,10 @@ def get_spark_job_driver_spec(
 ) -> models.SparkV1beta2DriverSpec:
     """Build DriverSpec for SparkApplication.
 
+    The service account is intentionally left unset so that the Spark
+    Operator's fallback service account (configured cluster-wide by the
+    operator installation) is used unless overridden via `driver`.
+
     Returns:
         SparkApplication DriverSpec model.
 
@@ -682,7 +686,7 @@ def get_spark_job_driver_spec(
     return models.SparkV1beta2DriverSpec(
         cores=cores,
         memory=memory,
-        service_account=constants.DEFAULT_SERVICE_ACCOUNT,
+        service_account=driver.service_account if driver else None,
     )
 
 

kubeflow/spark/backends/kubernetes/utils_test.py

@@ -1245,9 +1245,14 @@ def test_read_pod_logs(test_case: TestCase) -> None:
     "test_case",
     [
         TestCase(
-            name="default spark job driver spec",
+            name="default spark job driver spec leaves service account unset",
             expected_status=SUCCESS,
-            config={},
+            config={"driver": None},
+        ),
+        TestCase(
+            name="spark job driver spec honors service account override",
+            expected_status=SUCCESS,
+            config={"driver": Driver(service_account="custom-sa")},
         ),
     ],
 )
@@ -1256,15 +1261,16 @@ def test_get_spark_job_driver_spec(test_case: TestCase) -> None:
 
     print("Executing test:", test_case.name)
 
-    spec = get_spark_job_driver_spec()
+    driver = test_case.config["driver"]
+    spec = get_spark_job_driver_spec(driver=driver)
 
     assert test_case.expected_status == SUCCESS
 
     assert spec.cores == constants.DEFAULT_DRIVER_CPU
     assert spec.memory == _memory_kubernetes_to_spark(
         constants.DEFAULT_DRIVER_MEMORY,
     )
-    assert spec.service_account == constants.DEFAULT_SERVICE_ACCOUNT
+    assert spec.service_account == (driver.service_account if driver else None)
 
     print("test execution complete")