opencode-build/skills/spark-executor-mcp-operate/SKILL.md

---

name: spark-executor-mcp-operate description: Operation guide for the spark-executor-mcp service — 16 MCP tools that submit, monitor, fetch logs from, and kill PySpark jobs on YARN, plus connection-profile and script-file management. USE THIS SKILL whenever the user wants to do anything with the spark-executor-mcp MCP service: submit a Spark/PySpark job to YARN, check job status or result, fetch container logs, kill a running job, save/list/get/delete a YARN connection, or write/read/update a PySpark script file. Also use it when the user asks generally about Spark, YARN, pyspark, cluster jobs, or distributed compute via this service. Do NOT guess tool names or invent a submit flow — read this skill first to learn the two-step prepare/confirm flow, the dual-ID contract (job_id vs application_id), and the SQL guard / path restrictions. The MCP service does NOT generate PySpark code — the calling LLM writes the code, then uses write_job_file to persist it on disk before submitting.

spark-executor-mcp 操作指南

Spark-executor-mcp is an MCP (Model Context Protocol) service that exposes 16 tools for managing PySpark jobs on YARN. It runs as a FastAPI app mounted at /spark-executor-mcp; clients (typically LLM agents) call the tools via the standard MCP tools/call flow.

This skill is the canonical reference for an LLM agent using the service. It does not describe implementation details — for that, see the source in spark_executor/.

---

0. Mindset before you start

Three things you MUST internalize before calling any tool:

1. The submit flow is two-step: prepare_submit_job then confirm_submit_job. Prepare is cheap and reversible (just snapshots params + writes JSON to disk). Confirm is irreversible — it actually invokes spark-submit and a YARN application starts running. Never skip the prepare step, even when the user is in a hurry.

2. Every Job has two IDs that are NOT interchangeable everywhere.

   • job_id — local, 12-char hex (a1b2c3d4e5f6), generated by the service. Used for: looking up the Job record, all MCP tool calls.
   • application_id — YARN's own ID (application_17400000001_0001). Used for: YARN UI, tracking_url, the YARN REST API. The service's 4 job-lifecycle tools accept either ID (after the 2026-06-29 fix), but other systems (YARN UI, kubectl-style scripts) only know application_id. Always store both from every confirm_submit_job response.

3. The LLM writes the PySpark code. The MCP service writes the file. write_job_file is a file writer with SQL-safety checks, not a code generator. You compose the code in your context, then call write_job_file(code=...) to drop it on the container's filesystem where spark-submit can see it.

4. Three decisions MUST be approved by the human user before you act on them — never pick defaults silently, never auto-confirm. See §0.1 below. This is non-negotiable: confirm_submit_job starts a real YARN job that costs cluster time and may write to production data paths, and the user is the one who has to live with the result. Get their explicit "yes, confirm" before pulling the trigger.

0.1. Mandatory user confirmations (the three checkpoints)

For every Spark job the user asks you to run, you must stop and ask the human user to approve these three decisions in order. Do not pick defaults on their behalf. Do not assume "they'd obviously want prod / default queue / 2G / 2 cores / 2 executors" — those are YOUR defaults, not theirs.

Checkpoint 1 — when the user first asks to run a job: ASK the user for:

• Which connection to submit to. Show the existing connections (from list_connections) as options; if the right one doesn't exist, ask for the new connection's full profile (name, master, deploy_mode, yarn_rm_url, spark_conf, auth). See §5 for the field reference.
• The app_name (human-readable label for the YARN app). The service has no default for this; without it prepare_submit_job returns 400.

Do not skip past this even if the user is in a hurry or the request seems obvious. "Run a daily job on prod" still needs you to confirm which "daily job", which "prod", and what app_name to label it with.

Checkpoint 2 — before prepare_submit_job: ASK the user for the values of these submit parameters, even if they have server-side defaults:

• app_name (if you didn't get it at Checkpoint 1)
• queue
• executor_memory
• executor_cores
• num_executors
• any extra_args (jars, py-files, …) needed for this script

Tell the user the server-side defaults you WOULD use if they don't care (say "I'll use default queue, 2G memory, 2 cores, 2 executors unless you say otherwise") so they can either approve or override. The service itself rejects implicit defaults (returns 400 listing the assumed values) — this is by design, and you should respect it by being the one to surface the assumed values, not by silently letting the service 400 and then re-asking.

Checkpoint 3 — before confirm_submit_job (the irreversible one): Present a one-screen summary of exactly what will happen, and wait for the user to say "confirm" / "yes" / "do it" (or any unambiguous affirmative). The summary must include:

• The connection name and its master / deploy_mode / yarn_rm_url
• The app_name and the script path
• queue, executor_memory, executor_cores, num_executors
• The YARN tracking URL that will be returned (you don't know it yet, but you can say "you'll get a tracking URL once it starts")

The user's exact words don't need to be "confirm" — any clear affirmative works ("go", "ship it", "y", "🚀", "do it"). A non-answer or a "wait" / "hold on" / "let me think" is NOT consent; wait.

After they say yes, you may call confirm_submit_job. Never call confirm_submit_job without that explicit go-ahead.

---

1. Tool index

Operation ID	HTTP	Group	Purpose
save_connection	POST	Connection	Upsert a named YARN connection (master, deploy_mode, spark_conf, …)
list_connections	POST	Connection	List all saved connections
get_connection	POST	Connection	Fetch one connection by name
delete_connection	POST	Connection	Delete a connection by name
write_job_file	POST	Job file	Write LLM-written PySpark code to disk
read_job_file	POST	Job file	Read a script file back (capped at 1 MB)
update_job_file	POST	Job file	Overwrite a script file (capped at 1 MB)
prepare_submit_job	POST	Submit	Snapshot a job to PendingSubmission; does NOT submit
confirm_submit_job	POST	Submit	Actually invoke spark-submit for a pending_id
update_pending_job	POST	Submit	Edit a PENDING submission (script_path, queue, …)
cancel_pending_job	POST	Submit	Cancel a PENDING submission
list_pending_jobs	POST	Submit	List every PendingSubmission regardless of status
get_pending_job	POST	Submit	Fetch one PendingSubmission by id
get_job_status	POST	Job lifecycle	Query YARN for a job's live state (RUNNING / SUCCEEDED / …)
get_job_result	POST	Job lifecycle	Terminal-oriented view: final_status, diagnostics, timing
get_job_logs	POST	Job lifecycle	Aggregated container logs (default last 5000 chars)
kill_job	POST	Job lifecycle	PUT state=KILLED to YARN REST for the app

(Total: 16 MCP tools + /health GET, which is not exposed as a tool.)

---

2. The canonical happy path: submit a PySpark job

-- CHECKPOINT 1: ask user for connection + app_name --
1. list_connections / save_connection   # pick or create the right profile
2. write_job_file                        # write the LLM's PySpark to disk
3. read_job_file                         # (optional) sanity-check what was written
-- CHECKPOINT 2: ask user for queue / memory / cores / num_executors --
4. prepare_submit_job                    # snapshot + persist; no YARN call yet
-- CHECKPOINT 3: present summary, wait for explicit "yes, confirm" --
5. confirm_submit_job                    # NOW spark-submit runs (irreversible)
6. get_job_status / logs                 # poll while running
7. get_job_result                        # terminal view when done

The three CHECKPOINT lines above are mandatory pauses. Do NOT collapse them or skip past them. See §0.1 for the full rules; the short version is: ask the user before each tool call that affects cluster state, and never call confirm_submit_job without their explicit go-ahead.

2.1 Save a connection (one-time per cluster)

save_connection(
    name="prod",
    master="yarn",
    deploy_mode="cluster",
    yarn_rm_url="http://rm.example.com:8088",
    spark_conf={"spark.sql.shuffle.partitions": "200"},
    # optional: auth_type="kerberos", ssl_verify=False, ...
)

name is the lookup key — keep it short and stable. Re-calling save_connection with the same name overwrites; the original yarn_rm_url and spark_conf are snapshot into the PendingSubmission at prepare time and survive later edits to the connection (this is intentional — the user can re-point the connection at a different cluster without retargeting in-flight submissions).

2.2 Write the script to disk

The LLM writes the PySpark code. Then:

script_path = write_job_file(code=PYSPARK_SOURCE)["script_path"]
# Returns {"script_path": "/absolute/path/in/container.py", "bytes": N}

The path lives inside the container's SPARK_EXECUTOR_JOBS_DIR (default ./data/jobs/). For pre-existing scripts on the host, mount the host directory into the container and skip this step — pass the in-container path directly to prepare_submit_job.

2.3 Prepare (snapshot, no YARN call)

prepare_submit_job(
    connection="prod",
    script_path=script_path,
    app_name="daily-aggregation",     # REQUIRED, no default
    queue="default",                  # these 4 must be explicit
    executor_memory="2G",
    executor_cores=2,
    num_executors=2,
    # extra_args={"jars": "hdfs://...jar"},  # optional
)
# Returns {"pending_id": "p_a1b2c3", "status": "PENDING", "parameters": {...}}

You MUST pass queue, executor_memory, executor_cores, num_executors explicitly every time — even though defaults exist server-side, the service rejects implicit defaults with a 400 listing the assumed values. This forces the agent (or the human reviewer) to see and approve what will actually be submitted.

The response is the only chance to capture the pending_id you need for confirm_submit_job. Store it.

2.4 (Optional) Edit before submit

If something needs to change between prepare and confirm:

update_pending_job(
    pending_id="p_a1b2c3",
    queue="research",
    # script_path can also be changed — but the new file MUST exist
    # and pass the SQL guard
)

Refuses to edit anything that is no longer PENDING.

2.5 Confirm (the irreversible step)

result = confirm_submit_job(pending_id="p_a1b2c3")
# Returns {"job_id": "a1b2c3d4e5f6",
#          "application_id": "application_17400000001_0001",
#          "tracking_url": "http://rm:8088/proxy/application_17400000001_0001/"}

STORE BOTH IDS NOW. Future calls to get_job_status / get_job_logs / kill_job accept either, but application_id is the one YARN itself recognizes.

confirm_submit_job is idempotent on SUBMITTED (re-calling returns the same record) and retries on transient SparkSubmitError (3 attempts with 5 s delay by default). It transitions to FAILED only after all retries are exhausted.

2.6 Monitor / fetch / kill

# While running
state = get_job_status(job_id="a1b2c3d4e5f6").state
logs  = get_job_logs(job_id="a1b2c3d4e5f6", tail_chars=10000)

# When done
result = get_job_result(job_id="a1b2c3d4e5f6")
# result.state         -> "SUCCEEDED" / "FAILED" / "KILLED" / ...
# result.final_status  -> "SUCCEEDED" / "FAILED" / "KILLED" / "UNDEFINED"
# result.diagnostics   -> YARN's "why" string (often empty on success)
# result.tracking_url  -> the same proxy URL confirm_submit_job returned

# To stop it
kill_job(job_id="a1b2c3d4e5f6")

All four accept either job_id or application_id. Pass whichever you have on hand; the service tries job_id first, falls back to application_id.

---

3. The PendingSubmission state machine

   prepare_submit_job
         |
         v
     +-------+  cancel_pending_job
     |PENDING+----------> CANCELLED (terminal)
     +---+---+
         | confirm_submit_job
         v
     +----------+
     | SUBMITTED|  (terminal — kill via kill_job, not cancel)
     +----------+
         (or)
     +--------+
     | FAILED |  <-- confirm_submit_job exhausted retries
     +--------+
         | confirm_submit_job (resets to PENDING, retries)
         v
      PENDING

• cancel_pending_job refuses to cancel SUBMITTED or FAILED records — those are terminal; use kill_job instead.
• update_pending_job refuses to edit anything that is not PENDING.
• confirm_submit_job on a SUBMITTED record is a no-op (returns the same job_id / application_id). On a FAILED record, it resets to PENDING and retries. On a CANCELLED record, it raises 400.

---

4. The dual-ID contract

ID	Where it comes from	What it's good for
job_id	service-generated, 12-char hex (a1b2c3d4e5f6)	Looking up the Job in this service's store; all 4 lifecycle tools
application_id	parsed from spark-submit stderr (application_<ts>_<n>)	YARN UI, YARN REST, tracking_url, talking to ops scripts

Both IDs are returned by confirm_submit_job. Both are accepted by get_job_status / get_job_result / get_job_logs / kill_job. You don't need to remember which to pass — but DO remember to store both from the confirm response because:

• If the Job store has been wiped (rare; only on data dir loss), only application_id is still meaningful to YARN — you can't poll a YARN app you can't identify.
• If you're going to paste a link to a teammate, tracking_url (derived from application_id) is the human-friendly form.

If a lifecycle tool returns KeyError("No Job found for id='X' …"), check that X came from a successful confirm_submit_job response. The error message intentionally mentions both ID forms and shows an example of each so you can self-diagnose.

---

5. Connection profiles

Connection is the cluster profile — it's the "where to submit" record. Fields:

Field	Required	Default	Notes
name	yes	—	The lookup key
master	yes	"yarn"	"yarn" / "spark://…" / "k8s://…" / "local[N]"
deploy_mode	yes	"cluster"	"cluster" or "client"
yarn_rm_url	no	env YARN_RESOURCE_MANAGER_URL	Required if you'll call get_job_status / get_job_logs / kill_job (the lifecycle tools need a YARN endpoint)
spark_conf	no	{}	Merged into spark-submit --conf at submit time
ssl_verify	no	env SPARK_EXECUTOR_SSL_VERIFY_DEFAULT	bool
ssl_ca_bundle	no	env SPARK_EXECUTOR_SSL_CA_BUNDLE_DEFAULT	Path to CA file
auth_type	no	"none"	"none" / "simple" / "basic" / "kerberos"
auth_user / auth_password	for basic	—
auth_principal / auth_keytab	for kerberos (display only)	—	Real auth uses the system cache

Snapshot semantics: when prepare_submit_job runs, it copies master / deploy_mode / spark_conf / yarn_rm_url from the current Connection into the PendingSubmission. Editing the Connection later does not retarget an in-flight submission.

Lifecycle tools use the Job's snapshotted yarn_rm_url (set at confirm time) — not the Connection's current value. If the RM moves between confirm and a follow-up get_job_logs, you have a problem; re-pointing the Connection does not help.

---

6. Job file workflow (write_job_file / read_job_file / update_job_file)

These are filesystem I/O tools, not code generators. The LLM in context writes the PySpark code; these tools just persist it where spark-submit can find it, with safety guards.

Typical lifecycle:

LLM composes code   --> write_job_file(code=...)         # write
LLM reviews code    --> read_job_file(script_path=...)      # verify
LLM revises code    --> update_job_file(script_path=...,    # overwrite
                                          content=...)
LLM submits         --> prepare_submit_job(script_path=...)

Guards (these are why the tools exist instead of just cat > file.sh):

• SQL guard — write_job_file and update_job_file both reject code containing forbidden SQL (DROP, DELETE, UPDATE, TRUNCATE, ALTER, GRANT, REVOKE, …). The check is conservative; if it false-positives, rewrite to use only SELECT/INSERT or restructure.
• Path restriction — update_job_file only writes under SPARK_EXECUTOR_JOBS_DIR. This blocks accidentally overwriting host-mounted configs or the service's own files.
• Size cap — 1 MB on both read_job_file and update_job_file. Use read_job_file for small scripts; for huge ones, mount the host directory and skip these tools.

For pre-existing scripts on the host: mount the host dir into the container (e.g. -v /host/scripts:/app/scripts:ro in docker run) and pass the in-container path directly to prepare_submit_job — no need to round-trip through the file tools.

---

7. Error reference

Status	When	What to do
400	prepare_submit_job without explicit defaults; cancel_pending_job on terminal state; update_pending_job on non-PENDING; prepare_submit_job with missing/non-file script_path; SQL policy violation	Read the message — it almost always names the fix. Re-submit with corrected args.
404	KeyError from tool layer: unknown connection, pending_id, job_id, or application_id	The ID you sent doesn't match anything in the store. For lifecycle tools, the error message explicitly suggests both ID forms.
422	Pydantic validation: missing field, wrong type	Usually a typo or missing required arg. The response body lists every field error.
500	spark-submit subprocess failed and was not a SparkSubmitError (e.g. binary missing); YARN REST call failed	The tool logs the full traceback to data/logs/error/... — check there.

The service translates exceptions at the FastAPI layer:

• KeyError → 404 (with the KeyError's string as detail)
• ValueError → 400
• Pydantic validation → 422

So when the server returns 404, the body detail field is the same "Unknown job_id: …" / "Unknown connection: …" string the tool layer raised.

---

8. Common pitfalls

Pitfall	Why it bites	What to do
Skipping the prepare step	confirm_submit_job requires a pending_id; calling it without one returns 400	Always prepare first
Passing implicit defaults	Server returns 400 listing what you didn't confirm	Pass queue, executor_memory, executor_cores, num_executors every time
script_path doesn't exist in the container	prepare_submit_job returns 400; the agent generated the path in its own context but the container's filesystem is different	Either call write_job_file first, or mount a host dir
SQL injection in the script	write_job_file / update_job_file / prepare_submit_job all 400	Only use SELECT/INSERT; never DROP/DELETE/UPDATE
cancel_pending_job on a running YARN app	Returns 400 — submissions are terminal once submitted	Use kill_job(job_id=...) instead, which talks to YARN REST
Re-pointing a Connection and expecting in-flight jobs to follow	They don't — the Job snapshots yarn_rm_url at confirm time	Re-issue with a new prepare_submit_job if you actually need a different RM
Calling lifecycle tools with the wrong ID type	Used to fail; fixed in the 2026-06-29 update — both IDs are accepted	Just pass whichever you have; the service tries job_id first
Auto-confirming without the user's explicit "yes"	Triggers an irreversible YARN job the user may not have wanted — wastes cluster time, may write to wrong data paths	Per §0.1, always present a one-screen summary and wait for the user to say "confirm" / "yes" / equivalent before calling confirm_submit_job. Silence, "hmm", or "let me think" are NOT consent.
Picking connection / queue / executor defaults on the user's behalf	The user knows their cluster's cost / SLA / data layout better than you do; your defaults may not match their intent	Per §0.1, ASK at Checkpoint 1 (connection + app_name) and Checkpoint 2 (queue / memory / cores / num_executors). Surface the defaults you'd use so they can approve or override, don't just pick.
"Session not found" at the MCP layer	Multi-worker gunicorn + in-process session state	Out of scope for this skill; the service warns on startup if GUNICORN_WORKERS>1; set GUNICORN_WORKERS=1 or front with a sticky LB

---

9. End-to-end example (with the three checkpoints)

User: "Run a word count on s3a://my-bucket/books/*.txt and write the result back to s3a://my-bucket/wc/."

# ============================================================
# CHECKPOINT 1 — ask the user for connection + app_name FIRST
# ============================================================
#
# Agent: "I see you have these saved connections:
#         - prod   (yarn RM http://rm.prod.example.com:8088, cluster, kerberos)
#         - dev    (yarn RM http://rm.dev.example.com:8088,  client,  none)
#         Which one should I submit to? And what app_name should I label
#         the YARN app with?"
# User:  "prod, call it wordcount-books"
#
# (If neither connection is right, ask for the full profile and call
#  save_connection before continuing — see §2.1 for the fields.)

# 1. Compose the PySpark code IN YOUR CONTEXT
pyspark_code = '''
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("wordcount").getOrCreate()
sc = spark.sparkContext
files = sc.wholeTextFiles("s3a://my-bucket/books/")
words = files.flatMap(lambda f: f[1].lower().split()) \\
             .map(lambda w: (w, 1)) \\
             .reduceByKey(lambda a, b: a + b)
df = words.toDF(["word", "count"])
df.write.mode("overwrite").parquet("s3a://my-bucket/wc/")
'''

# 2. Persist to disk
script_path = write_job_file(code=pyspark_code)["script_path"]

# 3. Sanity-check (optional but recommended)
contents = read_job_file(script_path=script_path)["content"]
# ... review in context ...

# ============================================================
# CHECKPOINT 2 — ask the user for submit parameters
# ============================================================
#
# Agent: "I'll use these unless you say otherwise:
#         queue=default, executor_memory=2G, executor_cores=2,
#         num_executors=2. Any overrides (different queue, more memory,
#         jars/py-files to add)?"
# User:  "use research queue, 4G memory, 2 executors"
#
# (If the user says "defaults are fine", that's still an explicit
#  approval — you may then call prepare_submit_job with those values.)

# 4. Prepare (snapshot — no YARN call yet)
pending = prepare_submit_job(
    connection="prod",
    script_path=script_path,
    app_name="wordcount-books",
    queue="research",
    executor_memory="4G",
    executor_cores=2,
    num_executors=2,
)
pending_id = pending["pending_id"]

# ============================================================
# CHECKPOINT 3 — present summary, wait for explicit "confirm"
# ============================================================
#
# Agent: "About to run on YARN:
#         • connection:  prod (yarn, cluster, http://rm.prod.example.com:8088)
#         • app_name:    wordcount-books
#         • script:      <script_path>
#         • queue:       research
#         • resources:   2G memory × 2 cores × 2 executors
#         This will invoke spark-submit and start a real YARN application.
#         Confirm? (y / n)"
# User:  "yes, go"
#   ─── only NOW may you call confirm_submit_job ───

# 5. Confirm (NOW spark-submit runs)
result = confirm_submit_job(pending_id=pending_id)
job_id          = result["job_id"]            # STORE
application_id  = result["application_id"]    # STORE
tracking_url    = result["tracking_url"]      # STORE

# 6. Poll
import time
while True:
    st = get_job_status(job_id=job_id).state
    if st in ("SUCCEEDED", "FAILED", "KILLED", "FINISHED"):
        break
    time.sleep(30)

# 7. Final result
res = get_job_result(job_id=job_id)
print(res.final_status, res.diagnostics)

# If something went wrong and you want to see why:
# logs = get_job_logs(job_id=job_id, tail_chars=20000)

# (If you need to kill the job: ASK THE USER FIRST, then call
#  kill_job(job_id=job_id). Don't kill running jobs without an explicit
#  "kill it" — the user may have wanted to let it finish.)

---

10. Reference paths (for the curious)

This skill is the operation guide. The architectural / implementation details live in:

• CLAUDE.md — project conventions, persistence layout, two-step submit rationale, MCP arg pattern, MCP session affinity caveat.
• docs/superpowers/plans/2026-06-24-spark-executor-mcp.md — the original implementation plan with Stage 1 / 2 / 3 breakdown.
• spark_executor/server.py — the route + exception-handler layer.
• spark_executor/tools/requests.py — the Pydantic body models; useful to read when the MCP 422 response confuses you.
• spark_executor/core/job_store.py — Job persistence; explains the dual-ID contract and the file-backed JSON layout.