diff --git a/airflow/providers/apache/spark/hooks/spark_submit.py b/airflow/providers/apache/spark/hooks/spark_submit.py
index a12636c86a16d..37f624e310873 100644
--- a/airflow/providers/apache/spark/hooks/spark_submit.py
+++ b/airflow/providers/apache/spark/hooks/spark_submit.py
@@ -17,11 +17,16 @@
 # under the License.
 from __future__ import annotations
 
+import base64
 import contextlib
 import os
 import re
+import shutil
 import subprocess
+import tempfile
 import time
+import uuid
+from pathlib import Path
 from typing import Any, Iterator
 
 from airflow.configuration import conf as airflow_conf
@@ -67,7 +72,9 @@ class SparkSubmitHook(BaseHook, LoggingMixin):
     :param executor_memory: Memory per executor (e.g. 1000M, 2G) (Default: 1G)
     :param driver_memory: Memory allocated to the driver (e.g. 1000M, 2G) (Default: 1G)
     :param keytab: Full path to the file that contains the keytab
+                        (will overwrite any keytab defined in the connection's extra JSON)
     :param principal: The name of the kerberos principal used for keytab
+                        (will overwrite any principal defined in the connection's extra JSON)
     :param proxy_user: User to impersonate when submitting the application
     :param name: Name of the job (default airflow-spark)
     :param num_executors: Number of executors to launch
@@ -101,14 +108,17 @@ def get_ui_field_behaviour(cls) -> dict[str, Any]:
         return {
             "hidden_fields": ["schema", "login", "password", "extra"],
             "relabeling": {},
+            "placeholders": {
+                "keytab": "<base64 encoded Keytab Content>",
+            },
         }
 
     @classmethod
     def get_connection_form_widgets(cls) -> dict[str, Any]:
         """Return connection widgets to add to Spark connection form."""
-        from flask_appbuilder.fieldwidgets import BS3TextFieldWidget
+        from flask_appbuilder.fieldwidgets import BS3PasswordFieldWidget, BS3TextFieldWidget
         from flask_babel import lazy_gettext
-        from wtforms import StringField
+        from wtforms import PasswordField, StringField
         from wtforms.validators import Optional, any_of
 
         return {
@@ -135,6 +145,17 @@ def get_connection_form_widgets(cls) -> dict[str, Any]:
             "namespace": StringField(
                 lazy_gettext("Kubernetes namespace"), widget=BS3TextFieldWidget(), validators=[Optional()]
             ),
+            "principal": StringField(
+                lazy_gettext("Principal"),
+                widget=BS3TextFieldWidget(),
+                validators=[Optional()],
+            ),
+            "keytab": PasswordField(
+                lazy_gettext("Keytab"),
+                widget=BS3PasswordFieldWidget(),
+                description="Run the command `base64 <your-keytab-path>` and use its output.",
+                validators=[Optional()],
+            ),
         }
 
     def __init__(
@@ -237,6 +258,8 @@ def _resolve_connection(self) -> dict[str, Any]:
             "deploy_mode": None,
             "spark_binary": self.spark_binary or DEFAULT_SPARK_BINARY,
             "namespace": None,
+            "principal": self._principal,
+            "keytab": self._keytab,
         }
 
         try:
@@ -269,6 +292,14 @@ def _resolve_connection(self) -> dict[str, Any]:
                 )
             conn_data["spark_binary"] = self.spark_binary
             conn_data["namespace"] = extra.get("namespace")
+            if conn_data["principal"] is None:
+                conn_data["principal"] = extra.get("principal")
+            if conn_data["keytab"] is None:
+                base64_keytab = extra.get("keytab")
+                if base64_keytab is not None:
+                    conn_data["keytab"] = self._create_keytab_path_from_base64_keytab(
+                        base64_keytab, conn_data["principal"]
+                    )
         except AirflowException:
             self.log.info(
                 "Could not load connection string %s, defaulting to %s", self._conn_id, conn_data["master"]
@@ -282,6 +313,45 @@ def _resolve_connection(self) -> dict[str, Any]:
     def get_conn(self) -> Any:
         pass
 
+    def _create_keytab_path_from_base64_keytab(self, base64_keytab: str, principal: str | None) -> str:
+        _uuid = uuid.uuid4()
+        temp_dir_path = Path(tempfile.gettempdir()).resolve()
+        temp_file_name = f"airflow_keytab-{principal or _uuid}"
+
+        keytab_path = temp_dir_path / temp_file_name
+        staging_path = temp_dir_path / f".{temp_file_name}.{_uuid}"
+
+        try:
+            keytab = base64.b64decode(base64_keytab)
+        except Exception as err:
+            self.log.error("Failed to decode base64 keytab: %s", err)
+            raise AirflowException("Failed to decode base64 keytab") from err
+
+        # If a keytab file with the same content exists, return its path for the --keytab argument.
+        if keytab_path.exists():
+            with open(keytab_path, "rb") as f:
+                existing_keytab = f.read()
+            if existing_keytab == keytab:
+                self.log.info("Keytab file already exists and is the same as the provided keytab")
+                return str(keytab_path)
+
+        # Generate a new keytab file from the base64 encoded keytab value to use as a --keytab argument.
+        try:
+            with open(staging_path, "wb") as f:
+                self.log.info("Saving keytab to %s", staging_path)
+                f.write(keytab)
+
+            self.log.info("Moving keytab from %s to %s", staging_path, keytab_path)
+            shutil.move(str(staging_path), str(keytab_path))
+            return str(keytab_path)
+        except Exception as err:
+            self.log.error("Failed to save keytab: %s", err)
+            raise AirflowException("Failed to save keytab") from err
+        finally:
+            if staging_path.exists():
+                self.log.info("Removing staging keytab file: %s", staging_path)
+                staging_path.unlink()
+
     def _get_spark_binary_path(self) -> list[str]:
         # Assume that spark-submit is present in the path to the executing user
         return [self._connection["spark_binary"]]
@@ -370,10 +440,10 @@ def _build_spark_submit_command(self, application: str) -> list[str]:
             connection_cmd += ["--executor-memory", self._executor_memory]
         if self._driver_memory:
             connection_cmd += ["--driver-memory", self._driver_memory]
-        if self._keytab:
-            connection_cmd += ["--keytab", self._keytab]
-        if self._principal:
-            connection_cmd += ["--principal", self._principal]
+        if self._connection["keytab"]:
+            connection_cmd += ["--keytab", self._connection["keytab"]]
+        if self._connection["principal"]:
+            connection_cmd += ["--principal", self._connection["principal"]]
         if self._use_krb5ccache:
             if not os.getenv("KRB5CCNAME"):
                 raise AirflowException(
@@ -708,11 +778,13 @@ def on_kill(self) -> None:
             if self._yarn_application_id:
                 kill_cmd = f"yarn application -kill {self._yarn_application_id}".split()
                 env = {**os.environ, **(self._env or {})}
-                if self._keytab is not None and self._principal is not None:
+                if self._connection["keytab"] is not None and self._connection["principal"] is not None:
                     # we are ignoring renewal failures from renew_from_kt
                     # here as the failure could just be due to a non-renewable ticket,
                     # we still attempt to kill the yarn application
-                    renew_from_kt(self._principal, self._keytab, exit_on_fail=False)
+                    renew_from_kt(
+                        self._connection["principal"], self._connection["keytab"], exit_on_fail=False
+                    )
                     env = os.environ.copy()
                     ccacche = airflow_conf.get_mandatory_value("kerberos", "ccache")
                     env["KRB5CCNAME"] = ccacche
diff --git a/airflow/providers/apache/spark/operators/spark_submit.py b/airflow/providers/apache/spark/operators/spark_submit.py
index 281919b2b14c8..16f232dd02946 100644
--- a/airflow/providers/apache/spark/operators/spark_submit.py
+++ b/airflow/providers/apache/spark/operators/spark_submit.py
@@ -58,7 +58,9 @@ class SparkSubmitOperator(BaseOperator):
     :param executor_memory: Memory per executor (e.g. 1000M, 2G) (Default: 1G)
     :param driver_memory: Memory allocated to the driver (e.g. 1000M, 2G) (Default: 1G)
     :param keytab: Full path to the file that contains the keytab (templated)
+                        (will overwrite any keytab defined in the connection's extra JSON)
     :param principal: The name of the kerberos principal used for keytab (templated)
+                        (will overwrite any principal defined in the connection's extra JSON)
     :param proxy_user: User to impersonate when submitting the application (templated)
     :param name: Name of the job (default airflow-spark). (templated)
     :param num_executors: Number of executors to launch
diff --git a/tests/providers/apache/spark/hooks/test_spark_submit.py b/tests/providers/apache/spark/hooks/test_spark_submit.py
index b526e25b6f952..8e3bdcbbf7ec9 100644
--- a/tests/providers/apache/spark/hooks/test_spark_submit.py
+++ b/tests/providers/apache/spark/hooks/test_spark_submit.py
@@ -17,9 +17,11 @@
 # under the License.
 from __future__ import annotations
 
+import base64
 import os
 from io import StringIO
-from unittest.mock import call, patch
+from pathlib import Path
+from unittest.mock import call, mock_open, patch
 
 import pytest
 
@@ -143,6 +145,22 @@ def setup_method(self):
                 extra='{"deploy-mode": "client"}',
             )
         )
+        db.merge_conn(
+            Connection(
+                conn_id="spark_principal_set",
+                conn_type="spark",
+                host="yarn",
+                extra='{"principal": "user/spark@airflow.org"}',
+            )
+        )
+        db.merge_conn(
+            Connection(
+                conn_id="spark_keytab_set",
+                conn_type="spark",
+                host="yarn",
+                extra='{"keytab": "privileged_user.keytab"}',
+            )
+        )
 
     @patch(
         "airflow.providers.apache.spark.hooks.spark_submit.os.getenv", return_value="/tmp/airflow_krb5_ccache"
@@ -335,6 +353,8 @@ def test_resolve_connection_yarn_default(self):
             "deploy_mode": None,
             "queue": None,
             "namespace": None,
+            "principal": None,
+            "keytab": None,
         }
         assert connection == expected_spark_connection
         assert dict_cmd["--master"] == "yarn"
@@ -355,6 +375,8 @@ def test_resolve_connection_yarn_default_connection(self):
             "deploy_mode": None,
             "queue": "root.default",
             "namespace": None,
+            "principal": None,
+            "keytab": None,
         }
         assert connection == expected_spark_connection
         assert dict_cmd["--master"] == "yarn"
@@ -376,6 +398,8 @@ def test_resolve_connection_mesos_default_connection(self):
             "deploy_mode": None,
             "queue": None,
             "namespace": None,
+            "principal": None,
+            "keytab": None,
         }
         assert connection == expected_spark_connection
         assert dict_cmd["--master"] == "mesos://host:5050"
@@ -396,6 +420,8 @@ def test_resolve_connection_spark_yarn_cluster_connection(self):
             "deploy_mode": "cluster",
             "queue": "root.etl",
             "namespace": None,
+            "principal": None,
+            "keytab": None,
         }
         assert connection == expected_spark_connection
         assert dict_cmd["--master"] == "yarn://yarn-master"
@@ -418,6 +444,8 @@ def test_resolve_connection_spark_k8s_cluster_connection(self):
             "master": "k8s://https://k8s-master",
             "deploy_mode": "cluster",
             "namespace": "mynamespace",
+            "principal": None,
+            "keytab": None,
         }
         assert connection == expected_spark_connection
         assert dict_cmd["--master"] == "k8s://https://k8s-master"
@@ -442,6 +470,8 @@ def test_resolve_connection_spark_k8s_cluster_ns_conf(self):
             "master": "k8s://https://k8s-master",
             "deploy_mode": "cluster",
             "namespace": "airflow",
+            "principal": None,
+            "keytab": None,
         }
         assert connection == expected_spark_connection
         assert dict_cmd["--master"] == "k8s://https://k8s-master"
@@ -463,6 +493,8 @@ def test_resolve_connection_spark_binary_set_connection(self):
             "deploy_mode": None,
             "queue": None,
             "namespace": None,
+            "principal": None,
+            "keytab": None,
         }
         assert connection == expected_spark_connection
         assert cmd[0] == "spark2-submit"
@@ -482,6 +514,8 @@ def test_resolve_connection_spark_binary_spark3_submit_set_connection(self):
             "deploy_mode": None,
             "queue": None,
             "namespace": None,
+            "principal": None,
+            "keytab": None,
         }
         assert connection == expected_spark_connection
         assert cmd[0] == "spark3-submit"
@@ -512,6 +546,8 @@ def test_resolve_connection_spark_binary_default_value_override(self):
             "deploy_mode": None,
             "queue": None,
             "namespace": None,
+            "principal": None,
+            "keytab": None,
         }
         assert connection == expected_spark_connection
         assert cmd[0] == "spark3-submit"
@@ -531,6 +567,8 @@ def test_resolve_connection_spark_binary_default_value(self):
             "deploy_mode": None,
             "queue": "root.default",
             "namespace": None,
+            "principal": None,
+            "keytab": None,
         }
         assert connection == expected_spark_connection
         assert cmd[0] == "spark-submit"
@@ -550,10 +588,110 @@ def test_resolve_connection_spark_standalone_cluster_connection(self):
             "deploy_mode": "cluster",
             "queue": None,
             "namespace": None,
+            "principal": None,
+            "keytab": None,
         }
         assert connection == expected_spark_connection
         assert cmd[0] == "spark-submit"
 
+    def test_resolve_connection_principal_set_connection(self):
+        # Given
+        hook = SparkSubmitHook(conn_id="spark_principal_set")
+
+        # When
+        connection = hook._resolve_connection()
+        cmd = hook._build_spark_submit_command(self._spark_job_file)
+
+        # Then
+        dict_cmd = self.cmd_args_to_dict(cmd)
+        expected_spark_connection = {
+            "master": "yarn",
+            "spark_binary": "spark-submit",
+            "deploy_mode": None,
+            "queue": None,
+            "namespace": None,
+            "principal": "user/spark@airflow.org",
+            "keytab": None,
+        }
+        assert connection == expected_spark_connection
+        assert dict_cmd["--principal"] == "user/spark@airflow.org"
+
+    def test_resolve_connection_principal_value_override(self):
+        # Given
+        hook = SparkSubmitHook(conn_id="spark_principal_set", principal="will-override")
+
+        # When
+        connection = hook._resolve_connection()
+        cmd = hook._build_spark_submit_command(self._spark_job_file)
+
+        # Then
+        dict_cmd = self.cmd_args_to_dict(cmd)
+        expected_spark_connection = {
+            "master": "yarn",
+            "spark_binary": "spark-submit",
+            "deploy_mode": None,
+            "queue": None,
+            "namespace": None,
+            "principal": "will-override",
+            "keytab": None,
+        }
+        assert connection == expected_spark_connection
+        assert dict_cmd["--principal"] == "will-override"
+
+    @patch(
+        "airflow.providers.apache.spark.hooks.spark_submit.SparkSubmitHook._create_keytab_path_from_base64_keytab",
+        return_value="privileged_user.keytab",
+    )
+    def test_resolve_connection_keytab_set_connection(self, mock_create_keytab_path_from_base64_keytab):
+        # Given
+        hook = SparkSubmitHook(conn_id="spark_keytab_set")
+
+        # When
+        connection = hook._resolve_connection()
+        cmd = hook._build_spark_submit_command(self._spark_job_file)
+
+        # Then
+        dict_cmd = self.cmd_args_to_dict(cmd)
+        expected_spark_connection = {
+            "master": "yarn",
+            "spark_binary": "spark-submit",
+            "deploy_mode": None,
+            "queue": None,
+            "namespace": None,
+            "principal": None,
+            "keytab": "privileged_user.keytab",
+        }
+        assert connection == expected_spark_connection
+        assert dict_cmd["--keytab"] == "privileged_user.keytab"
+
+    @patch(
+        "airflow.providers.apache.spark.hooks.spark_submit.SparkSubmitHook._create_keytab_path_from_base64_keytab"
+    )
+    def test_resolve_connection_keytab_value_override(self, mock_create_keytab_path_from_base64_keytab):
+        # Given
+        hook = SparkSubmitHook(conn_id="spark_keytab_set", keytab="will-override")
+
+        # When
+        connection = hook._resolve_connection()
+        cmd = hook._build_spark_submit_command(self._spark_job_file)
+
+        # Then
+        dict_cmd = self.cmd_args_to_dict(cmd)
+        expected_spark_connection = {
+            "master": "yarn",
+            "spark_binary": "spark-submit",
+            "deploy_mode": None,
+            "queue": None,
+            "namespace": None,
+            "principal": None,
+            "keytab": "will-override",
+        }
+        assert connection == expected_spark_connection
+        assert dict_cmd["--keytab"] == "will-override"
+        assert (
+            not mock_create_keytab_path_from_base64_keytab.called
+        ), "Should not call _create_keytab_path_from_base64_keytab"
+
     def test_resolve_spark_submit_env_vars_standalone_client_mode(self):
         # Given
         hook = SparkSubmitHook(conn_id="spark_standalone_cluster_client_mode", env_vars={"bar": "foo"})
@@ -894,3 +1032,146 @@ def test_masks_passwords(self, command: str, expected: str) -> None:
 
         # Then
         assert command_masked == expected
+
+    def test_create_keytab_path_from_base64_keytab_with_decode_exception(self):
+        hook = SparkSubmitHook()
+        invalid_base64 = "invalid_base64"
+
+        with pytest.raises(AirflowException, match="Failed to decode base64 keytab"):
+            hook._create_keytab_path_from_base64_keytab(invalid_base64, None)
+
+    @patch("pathlib.Path.exists")
+    @patch("builtins.open", new_callable=mock_open)
+    def test_create_keytab_path_from_base64_keytab_with_write_exception(
+        self,
+        mock_open,
+        mock_exists,
+    ):
+        # Given
+        hook = SparkSubmitHook()
+
+        keytab_value = b"abcd"
+        base64_keytab = base64.b64encode(keytab_value).decode("UTF-8")
+        _mock_open = mock_open()
+        _mock_open.write.side_effect = Exception("Write failed")
+        mock_exists.return_value = False
+
+        # When
+        with pytest.raises(AirflowException, match="Failed to save keytab"):
+            hook._create_keytab_path_from_base64_keytab(base64_keytab, None)
+
+        # Then
+        assert mock_exists.call_count == 2  # called twice (before write, after write)
+
+    @patch("airflow.providers.apache.spark.hooks.spark_submit.shutil.move")
+    @patch("pathlib.Path.exists")
+    @patch("builtins.open", new_callable=mock_open)
+    def test_create_keytab_path_from_base64_keytab_with_move_exception(
+        self,
+        mock_open,
+        mock_exists,
+        mock_move,
+    ):
+        # Given
+        hook = SparkSubmitHook()
+
+        keytab_value = b"abcd"
+        base64_keytab = base64.b64encode(keytab_value).decode("UTF-8")
+        mock_exists.return_value = False
+        mock_move.side_effect = Exception("Move failed")
+
+        # When
+        with pytest.raises(AirflowException, match="Failed to save keytab"):
+            hook._create_keytab_path_from_base64_keytab(base64_keytab, None)
+
+        # Then
+        mock_open().write.assert_called_once_with(keytab_value)
+        mock_move.assert_called_once()
+        assert mock_exists.call_count == 2  # called twice (before write, after write)
+
+    @patch("airflow.providers.apache.spark.hooks.spark_submit.uuid.uuid4")
+    @patch("pathlib.Path.resolve")
+    @patch("airflow.providers.apache.spark.hooks.spark_submit.shutil.move")
+    @patch("pathlib.Path.exists")
+    @patch("builtins.open", new_callable=mock_open)
+    def test_create_keytab_path_from_base64_keytab_with_new_keytab(
+        self,
+        mock_open,
+        mock_exists,
+        mock_move,
+        mock_resolve,
+        mock_uuid4,
+    ):
+        # Given
+        hook = SparkSubmitHook()
+
+        keytab_value = b"abcd"
+        base64_keytab = base64.b64encode(keytab_value).decode("UTF-8")
+        mock_uuid4.return_value = "uuid"
+        mock_resolve.return_value = Path("resolved_path")
+        mock_exists.return_value = False
+
+        # When
+        keytab = hook._create_keytab_path_from_base64_keytab(base64_keytab, None)
+
+        # Then
+        assert keytab == "resolved_path/airflow_keytab-uuid"
+        mock_open().write.assert_called_once_with(keytab_value)
+        mock_move.assert_called_once()
+
+    @patch("pathlib.Path.resolve")
+    @patch("airflow.providers.apache.spark.hooks.spark_submit.shutil.move")
+    @patch("pathlib.Path.exists")
+    @patch("builtins.open", new_callable=mock_open)
+    def test_create_keytab_path_from_base64_keytab_with_new_keytab_with_principal(
+        self,
+        mock_open,
+        mock_exists,
+        mock_move,
+        mock_resolve,
+    ):
+        # Given
+        hook = SparkSubmitHook()
+
+        principal = "user/spark@airflow.org"
+        keytab_value = b"abcd"
+        base64_keytab = base64.b64encode(keytab_value).decode("UTF-8")
+        mock_resolve.return_value = Path("resolved_path")
+        mock_exists.return_value = False
+
+        # When
+        keytab = hook._create_keytab_path_from_base64_keytab(base64_keytab, principal)
+
+        # Then
+        assert keytab == f"resolved_path/airflow_keytab-{principal}"
+        mock_open().write.assert_called_once_with(keytab_value)
+        mock_move.assert_called_once()
+
+    @patch("pathlib.Path.resolve")
+    @patch("pathlib.Path.exists")
+    @patch("builtins.open", new_callable=mock_open)
+    def test_create_keytab_path_from_base64_keytab_with_existing_keytab(
+        self,
+        mock_open,
+        mock_exists,
+        mock_resolve,
+    ):
+        # Given
+        hook = SparkSubmitHook()
+
+        principal = "user/spark@airflow.org"
+        keytab_value = b"abcd"
+        base64_keytab = base64.b64encode(keytab_value)
+        mock_resolve.return_value = Path("resolved_path")
+        mock_exists.return_value = True
+        _mock_open = mock_open()
+        _mock_open.read.return_value = keytab_value
+
+        # When
+        keytab = hook._create_keytab_path_from_base64_keytab(base64_keytab.decode("UTF-8"), principal)
+
+        # Then
+        assert keytab == f"resolved_path/airflow_keytab-{principal}"
+        mock_open.assert_called_with(Path(f"resolved_path/airflow_keytab-{principal}"), "rb")
+        _mock_open.read.assert_called_once()
+        assert not _mock_open.write.called, "Keytab file should not be written"
