diff --git a/docs/modules/airflow/examples/example_spark_kubernetes_sensor.py b/docs/modules/airflow/examples/example_spark_kubernetes_sensor.py index 4ddf3afd..3c551d0b 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 b7ba3b90..19a24da5 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`. +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]