Add rdd command for individual RDD details

Completes full SHS REST API coverage - all 20 endpoints now have
CLI commands.

Co-authored-by: Copilot <[email protected]>

Author: yaooqinn

spark_history_cli/cli.py

@@ -147,6 +147,7 @@ def repl(state: CliState):
         "summary": "Application overview (config + workload)",
         "processes": "List miscellaneous processes",
         "rdds": "List cached RDDs for current app",
+        "rdd <id>": "Show RDD details",
         "env": "Show application environment",
         "logs <path>": "Download event logs to file",
         "version": "Show Spark version",
@@ -458,6 +459,15 @@ def repl(state: CliState):
                 headers, rows = fmt.format_rdd_list(rdds)
                 output_table(skin, headers, rows)
 
+            elif cmd == "rdd":
+                if not args or not args[0].isdigit():
+                    skin.error("Usage: rdd <rdd-id>")
+                else:
+                    app_id = state.session.require_app()
+                    data = client.get_rdd(app_id, int(args[0]))
+                    info = fmt.format_rdd_detail(data)
+                    output_status_block(skin, info, title=f"RDD {args[0]}")
+
             elif cmd == "env":
                 app_id = state.resolve_app_id(None)
                 env = client.get_environment(app_id)
@@ -979,6 +989,28 @@ def cmd_rdds(state: CliState):
         output_table(skin, headers, rows)
 
 
+@cli.command("rdd")
+@click.argument("rdd_id", type=int)
+@pass_state
+def cmd_rdd(state: CliState, rdd_id: int):
+    """Show details for a specific cached RDD.
+
+    Examples:
+
+      spark-history-cli -a <app> rdd 42
+    """
+    client = state.ensure_client()
+    app_id = state.resolve_app_id(None)
+    data = client.get_rdd(app_id, rdd_id)
+    if state.json_mode:
+        output_json(data)
+    else:
+        from spark_history_cli.utils.repl_skin import ReplSkin
+        skin = ReplSkin("spark_history", version=__version__)
+        info = fmt.format_rdd_detail(data)
+        output_status_block(skin, info, title=f"RDD {rdd_id}")
+
+
 @cli.command("env")
 @pass_state
 def cmd_env(state: CliState):

spark_history_cli/core/formatters.py

@@ -496,6 +496,23 @@ def format_rdd_list(rdds: list[dict]) -> tuple[list[str], list[list[str]]]:
     return headers, rows
 
 
+def format_rdd_detail(rdd: dict) -> dict[str, str]:
+    """Format RDD storage detail as a status block."""
+    info = {
+        "RDD ID": str(rdd.get("id", "")),
+        "Name": rdd.get("name", ""),
+        "Storage Level": rdd.get("storageLevel", ""),
+        "Partitions (cached/total)": f"{rdd.get('numCachedPartitions', 0)}/{rdd.get('numPartitions', 0)}",
+        "Memory Used": _bytes(rdd.get("memoryUsed")),
+        "Disk Used": _bytes(rdd.get("diskUsed")),
+    }
+    # Add per-partition data distribution if available
+    partitions = rdd.get("dataDistribution", rdd.get("partitions", []))
+    if partitions:
+        info["Data Distribution"] = f"{len(partitions)} entries"
+    return info
+
+
 # ── SQL Plan Formatters ───────────────────────────────────────────────