From 2250e7d17908509136887ef38079653759c268f1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Florian=20M=C3=BCller?= Date: Tue, 15 Sep 2026 16:40:07 +0200 Subject: [PATCH 1/2] fix: remove driver log pulling from sensor example --- .../example_spark_kubernetes_sensor.py | 38 ------------------- .../applying-custom-resources.adoc | 9 ++++- 2 files changed, 8 insertions(+), 39 deletions(-) diff --git a/docs/modules/airflow/examples/example_spark_kubernetes_sensor.py b/docs/modules/airflow/examples/example_spark_kubernetes_sensor.py index 4ddf3afd7..3c551d0bb 100644 --- a/docs/modules/airflow/examples/example_spark_kubernetes_sensor.py +++ b/docs/modules/airflow/examples/example_spark_kubernetes_sensor.py @@ -17,7 +17,6 @@ # under the License. from typing import Optional, Dict -from kubernetes import client from airflow.exceptions import AirflowException from airflow.sensors.base import BaseSensorOperator from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook @@ -34,7 +33,6 @@ def __init__( self, *, application_name: str, - attach_log: bool = False, namespace: Optional[str] = None, kubernetes_conn_id: str = "kubernetes_in_cluster", # <2> api_group: str = "spark.stackable.tech", @@ -44,7 +42,6 @@ def __init__( ) -> None: super().__init__(**kwargs) self.application_name = application_name - self.attach_log = attach_log self.namespace = namespace self.kubernetes_conn_id = kubernetes_conn_id self.hook = KubernetesHook(conn_id=self.kubernetes_conn_id) @@ -52,36 +49,6 @@ def __init__( self.api_version = api_version self.poke_interval = poke_interval - def _log_driver(self, application_state: str, response: dict) -> None: - if not self.attach_log: - return - status_info = response["status"] - if "driverInfo" not in status_info: - return - driver_info = status_info["driverInfo"] - if "podName" not in driver_info: - return - driver_pod_name = driver_info["podName"] - namespace = response["metadata"]["namespace"] - log_method = ( - self.log.error - if application_state in self.FAILURE_STATES - else self.log.info - ) - try: - log = "" - for line in self.hook.get_pod_logs(driver_pod_name, namespace=namespace): - log += line.decode() - log_method(log) - except client.rest.ApiException as e: - self.log.warning( - "Could not read logs for pod %s. It may have been disposed.\n" - "Make sure timeToLiveSeconds is set on your SparkApplication spec.\n" - "underlying exception: %s", - driver_pod_name, - e, - ) - def poke(self, context: Dict) -> bool: self.log.info("Poking: %s", self.application_name) response = self.hook.get_custom_object( @@ -98,11 +65,6 @@ def poke(self, context: Dict) -> bool: f"SparkApplication status could not be established: {response}" ) return False - if ( - self.attach_log - and application_state in self.FAILURE_STATES + self.SUCCESS_STATES - ): - self._log_driver(application_state, response) if application_state in self.FAILURE_STATES: raise AirflowException( f"SparkApplication failed with state: {application_state}" diff --git a/docs/modules/airflow/pages/usage-guide/applying-custom-resources.adoc b/docs/modules/airflow/pages/usage-guide/applying-custom-resources.adoc index b7ba3b90b..144404ac3 100644 --- a/docs/modules/airflow/pages/usage-guide/applying-custom-resources.adoc +++ b/docs/modules/airflow/pages/usage-guide/applying-custom-resources.adoc @@ -95,6 +95,13 @@ include::example$example-spark-dag.py[] <6> the subsequent task to monitor the job <7> the jobs are chained together in the correct order +[NOTE] +==== +The sensor only tracks the `status.phase` of the SparkApplication. +It cannot copy the Spark driver log into the Airflow task log: the Spark operator deletes the driver Pod as soon as the application reaches `Succeeded` or `Failed`, and the SparkApplication status does not reference the driver Pod. +To keep driver and executor logs, enable log aggregation for the SparkApplication as described in xref:spark-k8s:usage-guide/logging.adoc[], or use the xref:spark-k8s:usage-guide/history-server.adoc[Spark history server] for event logs. +==== + Once this DAG is xref:usage-guide/mounting-dags.adoc[mounted] in the DAG folder it can be called and its progress viewed from within the Webserver UI: image::airflow_dag_graph.png[Airflow Connections] @@ -109,7 +116,7 @@ TIP: A full example of the above is used as an integration test https://github.c == Logging -As mentioned above, the logs are available from the webserver UI if the jobs run with the `celeryExecutor`. +As mentioned above, the Airflow task logs are available from the webserver UI if the jobs run with the `celeryExecutor`. If the SDP logging mechanism has been deployed, log information can also be retrieved from the vector backend (e.g. Opensearch): image::airflow_dag_log_opensearch.png[Opensearch] From ac60da51a0c562bc9d8dfb8d8c1892754d31073f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Florian=20M=C3=BCller?= Date: Tue, 15 Sep 2026 17:20:44 +0200 Subject: [PATCH 2/2] Update docs/modules/airflow/pages/usage-guide/applying-custom-resources.adoc Co-authored-by: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> --- .../airflow/pages/usage-guide/applying-custom-resources.adoc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/modules/airflow/pages/usage-guide/applying-custom-resources.adoc b/docs/modules/airflow/pages/usage-guide/applying-custom-resources.adoc index 144404ac3..19a24da5c 100644 --- a/docs/modules/airflow/pages/usage-guide/applying-custom-resources.adoc +++ b/docs/modules/airflow/pages/usage-guide/applying-custom-resources.adoc @@ -98,7 +98,7 @@ include::example$example-spark-dag.py[] [NOTE] ==== The sensor only tracks the `status.phase` of the SparkApplication. -It cannot copy the Spark driver log into the Airflow task log: the Spark operator deletes the driver Pod as soon as the application reaches `Succeeded` or `Failed`, and the SparkApplication status does not reference the driver Pod. +It cannot copy the Spark driver log into the Airflow task log: the Spark operator deletes the driver Pod as soon as the application reaches `Succeeded` or `Failed`. To keep driver and executor logs, enable log aggregation for the SparkApplication as described in xref:spark-k8s:usage-guide/logging.adoc[], or use the xref:spark-k8s:usage-guide/history-server.adoc[Spark history server] for event logs. ====