Nataneljpwd commented on code in PR #71939:
URL: https://github.com/apache/airflow/pull/71939#discussion_r3906723113
##########
providers/apache/spark/tests/unit/apache/spark/hooks/test_spark_submit.py:
##########
@@ -164,6 +164,14 @@ def setup_connections(self, create_connection_without_db):
extra='{"deploy-mode": "client"}',
)
)
+ create_connection_without_db(
+ Connection(
+ conn_id="spark_standalone_cluster_rpc_port",
+ conn_type="spark",
+ host="spark://spark-standalone-master:7077",
Review Comment:
What about a test with ha? And a test with ha where the first host fails?
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -668,6 +668,21 @@ def _build_spark_submit_command(self, application: str) ->
list[str]:
return connection_cmd
+ def _get_standalone_rest_base_url(self) -> str:
+ """
+ Return the Spark standalone REST API base URL derived from the master
URL.
+
+ The master URL points at the binary RPC port (default 7077), but the
REST API
+ used for driver status/kill requests listens on
``spark.master.rest.port``
+ (default 6066). ``spark-submit --status/--kill`` derives its REST URL
from the
+ master URL itself, so it can never connect through the binary port.
Mirrors
+ ``_StandaloneSparkSubmitBackend.get_job_status``.
+ """
+ # ponytail: first host only for HA masters; per-host failover lives in
the operator backend
+ first_master = self._connection["master"].replace("spark://",
"").split(",")[0].strip()
Review Comment:
What if the first master is not available? Will it just fail? Maybe we
should try or exhaust all options first?
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -676,22 +691,21 @@ def _build_track_driver_status_command(self) -> list[str]:
"""
curl_max_wait_time = 30
spark_host = self._connection["master"]
- if spark_host.endswith(":6066"):
- spark_host = spark_host.replace("spark://", "http://";)
- connection_cmd = [
- "/usr/bin/curl",
- "--max-time",
- str(curl_max_wait_time),
- f"{spark_host}/v1/submissions/status/{self._driver_id}",
- ]
- self.log.info(connection_cmd)
-
+ if "spark://" in spark_host:
# The driver id so we can poll for its status
if not self._driver_id:
raise AirflowException(
"Invalid status: attempted to poll driver status but no
driver id is known. Giving up."
)
+ url =
f"{self._get_standalone_rest_base_url()}/v1/submissions/status/{self._driver_id}"
Review Comment:
In here as well we should handle HA better
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -676,22 +691,21 @@ def _build_track_driver_status_command(self) -> list[str]:
"""
curl_max_wait_time = 30
spark_host = self._connection["master"]
- if spark_host.endswith(":6066"):
- spark_host = spark_host.replace("spark://", "http://";)
- connection_cmd = [
- "/usr/bin/curl",
- "--max-time",
- str(curl_max_wait_time),
- f"{spark_host}/v1/submissions/status/{self._driver_id}",
- ]
- self.log.info(connection_cmd)
-
+ if "spark://" in spark_host:
Review Comment:
I would maybe add a comment stating that this means we are in standalone
mode, should be obvious but rather be safe
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
