Skip to content

Commit 2cb4257

Browse files
swebrazvan
andauthored
fix: remove driver log pulling from SparkKubernetesSensor example (#863)
* fix: remove driver log pulling from sensor example * Update docs/modules/airflow/pages/usage-guide/applying-custom-resources.adoc Co-authored-by: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> --------- Co-authored-by: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com>
1 parent 700d6a2 commit 2cb4257

2 files changed

Lines changed: 8 additions & 39 deletions

File tree

‎docs/modules/airflow/examples/example_spark_kubernetes_sensor.py‎

Lines changed: 0 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@
1717
# under the License.
1818

1919
from typing import Optional, Dict
20-
from kubernetes import client
2120
from airflow.exceptions import AirflowException
2221
from airflow.sensors.base import BaseSensorOperator
2322
from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
@@ -34,7 +33,6 @@ def __init__(
3433
self,
3534
*,
3635
application_name: str,
37-
attach_log: bool = False,
3836
namespace: Optional[str] = None,
3937
kubernetes_conn_id: str = "kubernetes_in_cluster", # <2>
4038
api_group: str = "spark.stackable.tech",
@@ -44,44 +42,13 @@ def __init__(
4442
) -> None:
4543
super().__init__(**kwargs)
4644
self.application_name = application_name
47-
self.attach_log = attach_log
4845
self.namespace = namespace
4946
self.kubernetes_conn_id = kubernetes_conn_id
5047
self.hook = KubernetesHook(conn_id=self.kubernetes_conn_id)
5148
self.api_group = api_group
5249
self.api_version = api_version
5350
self.poke_interval = poke_interval
5451

55-
def _log_driver(self, application_state: str, response: dict) -> None:
56-
if not self.attach_log:
57-
return
58-
status_info = response["status"]
59-
if "driverInfo" not in status_info:
60-
return
61-
driver_info = status_info["driverInfo"]
62-
if "podName" not in driver_info:
63-
return
64-
driver_pod_name = driver_info["podName"]
65-
namespace = response["metadata"]["namespace"]
66-
log_method = (
67-
self.log.error
68-
if application_state in self.FAILURE_STATES
69-
else self.log.info
70-
)
71-
try:
72-
log = ""
73-
for line in self.hook.get_pod_logs(driver_pod_name, namespace=namespace):
74-
log += line.decode()
75-
log_method(log)
76-
except client.rest.ApiException as e:
77-
self.log.warning(
78-
"Could not read logs for pod %s. It may have been disposed.\n"
79-
"Make sure timeToLiveSeconds is set on your SparkApplication spec.\n"
80-
"underlying exception: %s",
81-
driver_pod_name,
82-
e,
83-
)
84-
8552
def poke(self, context: Dict) -> bool:
8653
self.log.info("Poking: %s", self.application_name)
8754
response = self.hook.get_custom_object(
@@ -98,11 +65,6 @@ def poke(self, context: Dict) -> bool:
9865
f"SparkApplication status could not be established: {response}"
9966
)
10067
return False
101-
if (
102-
self.attach_log
103-
and application_state in self.FAILURE_STATES + self.SUCCESS_STATES
104-
):
105-
self._log_driver(application_state, response)
10668
if application_state in self.FAILURE_STATES:
10769
raise AirflowException(
10870
f"SparkApplication failed with state: {application_state}"

‎docs/modules/airflow/pages/usage-guide/applying-custom-resources.adoc‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,13 @@ include::example$example-spark-dag.py[]
9595
<6> the subsequent task to monitor the job
9696
<7> the jobs are chained together in the correct order
9797

98+
[NOTE]
99+
====
100+
The sensor only tracks the `status.phase` of the SparkApplication.
101+
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`.
102+
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.
103+
====
104+
98105
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:
99106

100107
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
109116

110117
== Logging
111118

112-
As mentioned above, the logs are available from the webserver UI if the jobs run with the `celeryExecutor`.
119+
As mentioned above, the Airflow task logs are available from the webserver UI if the jobs run with the `celeryExecutor`.
113120
If the SDP logging mechanism has been deployed, log information can also be retrieved from the vector backend (e.g. Opensearch):
114121

115122
image::airflow_dag_log_opensearch.png[Opensearch]

0 commit comments

Comments
 (0)