diff --git a/airflow/providers/apache/spark/hooks/spark_submit.py b/airflow/providers/apache/spark/hooks/spark_submit.py index b6b5b9396899a..ec9b69909c0cc 100644 --- a/airflow/providers/apache/spark/hooks/spark_submit.py +++ b/airflow/providers/apache/spark/hooks/spark_submit.py @@ -96,6 +96,7 @@ def get_ui_field_behaviour() -> dict[str, Any]: def __init__( self, conf: dict[str, Any] | None = None, + properties_file: str | None = None, conn_id: str = "spark_default", files: str | None = None, py_files: str | None = None, @@ -123,6 +124,7 @@ def __init__( ) -> None: super().__init__() self._conf = conf or {} + self._properties_file = properties_file self._conn_id = conn_id self._files = files self._py_files = py_files @@ -287,6 +289,8 @@ def _build_spark_submit_command(self, application: str) -> list[str]: "--conf", f"spark.kubernetes.namespace={self._connection['namespace']}", ] + if self._properties_file: + connection_cmd += ["--properties-file", self._properties_file] if self._files: connection_cmd += ["--files", self._files] if self._py_files: diff --git a/airflow/providers/apache/spark/operators/spark_submit.py b/airflow/providers/apache/spark/operators/spark_submit.py index 0620a44c0ea46..44b874b6ca0f5 100644 --- a/airflow/providers/apache/spark/operators/spark_submit.py +++ b/airflow/providers/apache/spark/operators/spark_submit.py @@ -74,6 +74,7 @@ class SparkSubmitOperator(BaseOperator): template_fields: Sequence[str] = ( "_application", "_conf", + "_properties-file" "_files", "_py_files", "_jars", @@ -94,6 +95,7 @@ def __init__( *, application: str = "", conf: dict[str, Any] | None = None, + properties_file: str | None = None, conn_id: str = "spark_default", files: str | None = None, py_files: str | None = None, @@ -123,6 +125,7 @@ def __init__( super().__init__(**kwargs) self._application = application self._conf = conf + self._properties_file = properties_file self._files = files self._py_files = py_files self._archives = archives @@ -163,6 +166,7 @@ def on_kill(self) -> None: def _get_hook(self) -> SparkSubmitHook: return SparkSubmitHook( conf=self._conf, + properties_file=self._properties_file, conn_id=self._conn_id, files=self._files, py_files=self._py_files,