From 4f115d02a3c0da6649a24de9a23bb376577f6ae8 Mon Sep 17 00:00:00 2001 From: ghostp13409 Date: Thu, 12 Oct 2023 18:54:05 +0530 Subject: [PATCH 1/3] property-files command added to SparkSubmitOperator --- airflow/providers/apache/spark/operators/spark_submit.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/airflow/providers/apache/spark/operators/spark_submit.py b/airflow/providers/apache/spark/operators/spark_submit.py index 0620a44c0ea46..6de2a862881c2 100644 --- a/airflow/providers/apache/spark/operators/spark_submit.py +++ b/airflow/providers/apache/spark/operators/spark_submit.py @@ -160,6 +160,11 @@ def on_kill(self) -> None: self._hook = self._get_hook() self._hook.on_kill() + def property_files(self) -> None: + if self.hook is None: + self._hook = self._get_hook() + self.hook.property_files() + def _get_hook(self) -> SparkSubmitHook: return SparkSubmitHook( conf=self._conf, From 99f31cce5efb34bbcf26f4b8613be6a23e38d6c7 Mon Sep 17 00:00:00 2001 From: ghostp13409 Date: Thu, 12 Oct 2023 20:34:04 +0530 Subject: [PATCH 2/3] 34838- property-files command added to SparkSubmitOperator --- airflow/providers/apache/spark/operators/spark_submit.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/airflow/providers/apache/spark/operators/spark_submit.py b/airflow/providers/apache/spark/operators/spark_submit.py index 6de2a862881c2..9cb9cd017b08e 100644 --- a/airflow/providers/apache/spark/operators/spark_submit.py +++ b/airflow/providers/apache/spark/operators/spark_submit.py @@ -163,7 +163,7 @@ def on_kill(self) -> None: def property_files(self) -> None: if self.hook is None: self._hook = self._get_hook() - self.hook.property_files() + self._hook.property_files() def _get_hook(self) -> SparkSubmitHook: return SparkSubmitHook( From 1ddae961d9bc2db772c7a92fa74011fa462e1041 Mon Sep 17 00:00:00 2001 From: ghostp13409 Date: Sat, 14 Oct 2023 00:03:40 +0530 Subject: [PATCH 3/3] reverted previous changes and added property files option in the Spark-Submit command --- airflow/providers/apache/spark/hooks/spark_submit.py | 4 ++++ airflow/providers/apache/spark/operators/spark_submit.py | 9 ++++----- 2 files changed, 8 insertions(+), 5 deletions(-) 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 9cb9cd017b08e..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 @@ -160,14 +163,10 @@ def on_kill(self) -> None: self._hook = self._get_hook() self._hook.on_kill() - def property_files(self) -> None: - if self.hook is None: - self._hook = self._get_hook() - self._hook.property_files() - 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,