refactor: replace yarn CLI shell-out with YARN REST API

yarn_client.py no longer invokes the 'yarn' binary via subprocess; it uses
httpx against /ws/v1/cluster/apps/* endpoints. This means the runtime image
no longer needs the Hadoop client installation — the only YARN-side
dependency left in the container is the config dir consumed by
spark-submit itself.

New model field:
  - Job.yarn_rm_url: str | None
  - PendingSubmission.yarn_rm_url: str | None (snapshotted at prepare)

prepare_submit_job snapshots Connection.yarn_rm_url into the pending
record (consistent with the existing master/deploy_mode/spark_conf
snapshot pattern); confirm_submit_job copies it onto the Job so
status/logs/kill can use it without re-looking-up the connection.

Resolution order for the RM URL at runtime:
  1. Job.yarn_rm_url (preferred — survives connection edits/deletes)
  2. Connection.yarn_rm_url fallback (if a future tool is added that
     doesn't go through a Job)
  3. YARN_RESOURCE_MANAGER_URL env var

Errors:
  - YarnConfigError (HTTP 4xx semantics) when URL is missing/malformed
  - YarnError for HTTP 4xx/5xx from the RM, network failures, missing
    state field, or unparseable log responses

10 new tests in test_yarn_client.py cover the REST surface:
success, 404, 5xx, missing state field, env-var fallback, malformed
URL, log 404 with log-aggregation hint, kill PUT body shape, and
httpx connection-error wrapping.

spark_executor/core/yarn_client.py

@@ -2,69 +2,109 @@
 """
 @Time   :2026/6/24
 @Author :tao.chen
+
+YARN ResourceManager REST API client. Replaces the previous `yarn` CLI shell-out
+so the runtime image does not need a Hadoop client installation — the
+`httpx` library already in pyproject.toml is enough.
+
+Endpoints used (YARN 2.6+):
+  GET  /ws/v1/cluster/apps/{appid}             -> app status + state
+  GET  /ws/v1/cluster/apps/{appid}/aggregated-logs  -> aggregated container logs
+  PUT  /ws/v1/cluster/apps/{appid}/state       -> kill an app (body: {"state":"KILLED"})
+
+The ResourceManager URL is passed in per call (snapshotted on the Job at
+confirm_submit_job time) and falls back to the YARN_RESOURCE_MANAGER_URL env
+var if unset. This matches the pattern the original Connection.yarn_rm_url
+field was designed for, but no longer requires the `yarn` CLI to interpret it.
 """
-import re
-import subprocess
+import json
+import os
+
+import httpx
 
 from common.logging import logger
 
 
 class YarnError(Exception):
-    """Raised when a yarn CLI invocation fails."""
+    """Raised when a YARN REST API call fails."""
 
 
-_STATE_RE = re.compile(r"State\s*:\s*(\S+)")
+class YarnConfigError(YarnError):
+    """Raised when the YARN ResourceManager URL is missing or malformed."""
 
 
-def _run(cmd: list[str]) -> "subprocess.CompletedProcess[str]":
-    logger.debug(f"yarn _run exec: {cmd}")
-    result = subprocess.run(cmd, capture_output=True, text=True, errors="replace")
+def _base_url(yarn_rm_url: str | None) -> str:
+    """Resolve and validate the RM URL. Raises YarnConfigError if unusable."""
+    url = yarn_rm_url or os.environ.get("YARN_RESOURCE_MANAGER_URL")
+    if not url:
+        raise YarnConfigError(
+            "YARN ResourceManager URL is not configured. "
+            "Set Connection.yarn_rm_url when saving the connection, "
+            "or set the YARN_RESOURCE_MANAGER_URL environment variable."
+        )
+    base = url.rstrip("/")
+    if not base.startswith(("http://", "https://")):
+        raise YarnConfigError(
+            f"YARN ResourceManager URL must start with http:// or https://: {url!r}"
+        )
+    return base
+
+
+def _request(method: str, url: str, *, json_body: dict | None = None,
+             timeout: float = 30.0) -> httpx.Response:
+    logger.debug(f"YARN {method} {url}" + (f" body={json_body}" if json_body else ""))
+    try:
+        resp = httpx.request(method, url, json=json_body, timeout=timeout)
+    except httpx.HTTPError as exc:
+        logger.error(f"YARN {method} {url} failed: {exc}")
+        raise YarnError(f"YARN connection failed: {exc}") from exc
     logger.debug(
-        f"yarn _run done rc={result.returncode} "
-        f"stdout_len={len(result.stdout)} stderr_len={len(result.stderr)}"
+        f"YARN {method} {url} -> {resp.status_code} "
+        f"({len(resp.content)} bytes)"
     )
-    return result
+    return resp
 
 
-def get_application_status(application_id: str) -> tuple[str, str]:
-    proc = _run(["yarn", "application", "-status", application_id])
-    if proc.returncode != 0:
-        logger.error(
-            f"yarn application -status failed (rc={proc.returncode}) for "
-            f"{application_id}: {proc.stderr[:500]}"
-        )
+def get_application_status(application_id: str, yarn_rm_url: str | None) -> tuple[str, str]:
+    """Return (state, raw_json_text) for an application, or raise YarnError."""
+    url = f"{_base_url(yarn_rm_url)}/ws/v1/cluster/apps/{application_id}"
+    resp = _request("GET", url)
+    if resp.status_code == 404:
+        raise YarnError(f"YARN application {application_id!r} not found")
+    if resp.status_code >= 400:
+        logger.error(f"YARN GET {url} -> {resp.status_code}: {resp.text[:500]}")
+        raise YarnError(f"YARN GET returned HTTP {resp.status_code}")
+    data = resp.json()
+    app = data.get("app", {})
+    state = app.get("state")
+    if not state:
+        raise YarnError(f"Could not parse YARN state from response: {data!r}")
+    logger.info(f"YARN status {application_id} -> {state}")
+    return state, json.dumps(data, indent=2)
+
+
+def get_application_logs(application_id: str, yarn_rm_url: str | None) -> str:
+    """Return aggregated container logs for an application as text."""
+    url = f"{_base_url(yarn_rm_url)}/ws/v1/cluster/apps/{application_id}/aggregated-logs"
+    resp = _request("GET", url, timeout=60.0)
+    if resp.status_code == 404:
         raise YarnError(
-            f"yarn application -status failed (rc={proc.returncode}): {proc.stderr}"
+            f"YARN aggregated logs not available for {application_id!r}. "
+            f"The application may not be in FINISHED state, or "
+            f"yarn.log-aggregation-enable is false on the cluster."
         )
-    match = _STATE_RE.search(proc.stdout)
-    if not match:
-        raise YarnError(f"Could not parse YARN state from output: {proc.stdout!r}")
-    state = match.group(1)
-    logger.info(f"yarn status {application_id} -> {state}")
-    return state, proc.stdout
+    if resp.status_code >= 400:
+        logger.error(f"YARN GET {url} -> {resp.status_code}: {resp.text[:500]}")
+        raise YarnError(f"YARN GET logs returned HTTP {resp.status_code}")
+    logger.info(f"YARN logs {application_id} -> {len(resp.text)} chars")
+    return resp.text
 
 
-def get_application_logs(application_id: str) -> str:
-    proc = _run(["yarn", "logs", "-applicationId", application_id])
-    if proc.returncode != 0:
-        logger.error(
-            f"yarn logs failed (rc={proc.returncode}) for {application_id}: {proc.stderr[:500]}"
-        )
-        raise YarnError(
-            f"yarn logs failed (rc={proc.returncode}): {proc.stderr}"
-        )
-    logger.info(f"yarn logs {application_id} -> {len(proc.stdout)} chars")
-    return proc.stdout
-
-
-def kill_application(application_id: str) -> None:
-    proc = _run(["yarn", "application", "-kill", application_id])
-    if proc.returncode != 0:
-        logger.error(
-            f"yarn application -kill failed (rc={proc.returncode}) for "
-            f"{application_id}: {proc.stderr[:500]}"
-        )
-        raise YarnError(
-            f"yarn application -kill failed (rc={proc.returncode}): {proc.stderr}"
-        )
-    logger.info(f"yarn kill {application_id} -> ok")
+def kill_application(application_id: str, yarn_rm_url: str | None) -> None:
+    """PUT state=KILLED to /ws/v1/cluster/apps/{appid}/state."""
+    url = f"{_base_url(yarn_rm_url)}/ws/v1/cluster/apps/{application_id}/state"
+    resp = _request("PUT", url, json_body={"state": "KILLED"})
+    if resp.status_code >= 400:
+        logger.error(f"YARN PUT {url} -> {resp.status_code}: {resp.text[:500]}")
+        raise YarnError(f"YARN kill returned HTTP {resp.status_code}: {resp.text}")
+    logger.info(f"YARN kill {application_id} -> ok")

spark_executor/models.py

@@ -15,6 +15,7 @@ class Job(BaseModel):
     queue: str
     submit_time: datetime
     connection: str
+    yarn_rm_url: str | None = None
 
 
 class JobStatus(BaseModel):
@@ -42,6 +43,7 @@ class PendingSubmission(BaseModel):
     connection: str
     master: str
     deploy_mode: str
+    yarn_rm_url: str | None = None
     script_path: str
     queue: str
     executor_memory: str

spark_executor/tools/kill.py

@@ -15,7 +15,7 @@ def kill_job(job_id: str) -> dict[str, str]:
     job = store.get(job_id)
     if job is None:
         raise KeyError(f"Unknown job_id: {job_id}")
-    kill_application(job.application_id)
+    kill_application(job.application_id, job.yarn_rm_url)
     logger.info(f"kill_job ok job_id={job_id} application_id={job.application_id}")
     return {
         "job_id": job_id,

spark_executor/tools/logs.py

@@ -15,7 +15,7 @@ def get_job_logs(job_id: str, tail_chars: int = 5000) -> str:
     job = store.get(job_id)
     if job is None:
         raise KeyError(f"Unknown job_id: {job_id}")
-    full = get_application_logs(job.application_id)
+    full = get_application_logs(job.application_id, job.yarn_rm_url)
     tailed = full[-tail_chars:] if len(full) > tail_chars else full
     logger.info(
         f"get_job_logs ok job_id={job_id} application_id={job.application_id} "

spark_executor/tools/status.py

@@ -16,6 +16,6 @@ def get_job_status(job_id: str) -> JobStatus:
     job = store.get(job_id)
     if job is None:
         raise KeyError(f"Unknown job_id: {job_id}")
-    state, raw = get_application_status(job.application_id)
+    state, raw = get_application_status(job.application_id, job.yarn_rm_url)
     logger.info(f"get_job_status ok job_id={job_id} application_id={job.application_id} state={state}")
     return JobStatus(application_id=job.application_id, state=state, raw=raw)

spark_executor/tools/submit.py

@@ -49,6 +49,7 @@ def prepare_submit_job(
         connection=connection,
         master=conn.master,
         deploy_mode=conn.deploy_mode,
+        yarn_rm_url=conn.yarn_rm_url,
         script_path=script_path,
         queue=queue,
         executor_memory=executor_memory,
@@ -119,6 +120,7 @@ def confirm_submit_job(*, pending_id: str) -> SubmitResult:
             queue=pending.queue,
             submit_time=datetime.utcnow(),
             connection=pending.connection,
+            yarn_rm_url=pending.yarn_rm_url,
         )
     )