Add analyze CLI command and harden capture.py (PR 4) (#11)

- Add `sparkparse analyze` command: calls get_parsed_metrics then
  to_plan_summary; supports --format json/text and --out-file
- Add --version eager callback; remove dead `welcome` command
- Add help strings to all typer.Option/Argument calls in get and viz
- Replace print() with logging in capture.py
- Narrow action type to Literal["viz","get","analyze"] in capture.py
- Add "analyze" action in SparkparseCapture.__exit__; stores result in
  self._analysis
- Rename SparkSession app name from "temp" to "sparkparse_capture"

https://claude.ai/code/session_01U1F1a7vVfbxnBeU6oNwnxz

Co-authored-by: Claude <noreply@anthropic.com>

Author: jeffbrennan

sparkparse/app.py

@@ -1,32 +1,83 @@
+import json
+import sys
+from enum import StrEnum
+from pathlib import Path
+from typing import Annotated
+
 import typer
 
+from sparkparse.analyze import to_plan_summary
 from sparkparse.dashboard import init_dashboard, run_app
 from sparkparse.models import OutputFormat, ParsedLogDataFrames
 from sparkparse.parse import get_parsed_metrics
 
+__version__ = "0.1.0"
+
 app = typer.Typer(pretty_exceptions_enable=False)
 
 
+class AnalysisFormat(StrEnum):
+    json = "json"
+    text = "text"
+
+
+def _version_callback(value: bool) -> None:
+    if value:
+        typer.echo(f"sparkparse {__version__}")
+        raise typer.Exit()
+
+
+@app.callback()
+def main(
+    version: Annotated[
+        bool | None,
+        typer.Option(
+            "--version",
+            callback=_version_callback,
+            is_eager=True,
+            help="Show version and exit.",
+        ),
+    ] = None,
+) -> None:
+    pass
+
+
 @app.command("viz")
 def viz_parsed_metrics(
-    log_dir: str = "data/logs/raw",
+    log_dir: Annotated[
+        str, typer.Argument(help="Directory containing raw Spark event logs.")
+    ] = "data/logs/raw",
     force_port: bool = typer.Option(
-        default=False, help="force kill processes using port 8050"
+        default=False, help="Force kill any process using port 8050 before starting."
     ),
 ) -> None:
+    """Launch the interactive Dash dashboard for a directory of Spark event logs."""
     app = init_dashboard(log_dir)
     run_app(app=app, force_port=force_port)
 
 
 @app.command("get")
 def get(
-    log_dir: str = "data/logs/raw",
-    log_file: str | None = None,
-    out_dir: str | None = "data/logs/parsed",
-    out_name: str | None = None,
-    out_format: OutputFormat | None = OutputFormat.csv,
-    verbose: bool = False,
+    log_dir: Annotated[
+        str, typer.Argument(help="Directory containing raw Spark event logs.")
+    ] = "data/logs/raw",
+    log_file: Annotated[
+        str | None,
+        typer.Option(help="Parse a single log file instead of the whole directory."),
+    ] = None,
+    out_dir: Annotated[
+        str | None, typer.Option(help="Directory to write parsed output files.")
+    ] = "data/logs/parsed",
+    out_name: Annotated[
+        str | None,
+        typer.Option(help="Base name for output files (defaults to log file stem)."),
+    ] = None,
+    out_format: Annotated[
+        OutputFormat | None, typer.Option(help="Output file format.")
+    ] = OutputFormat.csv,
+    verbose: Annotated[bool, typer.Option(help="Print extra parsing details.")] = False,
 ) -> ParsedLogDataFrames:
+    """Parse Spark event logs and write structured DataFrames to disk."""
     return get_parsed_metrics(
         log_dir=log_dir,
         log_file=log_file,
@@ -37,10 +88,72 @@ def get(
     )
 
 
-@app.command()
-def welcome():
-    typer.echo("Welcome to sparkparse CLI")
-    typer.echo("Use --help to see available commands")
+@app.command("analyze")
+def analyze(
+    log_dir: Annotated[
+        str, typer.Argument(help="Directory containing raw Spark event logs.")
+    ] = "data/logs/raw",
+    log_file: Annotated[
+        str | None,
+        typer.Option(help="Analyze a single log file instead of the whole directory."),
+    ] = None,
+    out_file: Annotated[
+        Path | None,
+        typer.Option(help="Write analysis JSON to this file instead of stdout."),
+    ] = None,
+    format: Annotated[
+        AnalysisFormat,
+        typer.Option(help="Output format: 'json' (default) or human-readable 'text'."),
+    ] = AnalysisFormat.json,
+) -> None:
+    """Analyze Spark event logs and emit a token-efficient summary suitable for LLM piping."""
+    dfs = get_parsed_metrics(
+        log_dir=log_dir,
+        log_file=log_file,
+        out_dir=None,
+        out_name=None,
+        out_format=None,
+        verbose=False,
+    )
+
+    log_name = Path(log_file).stem if log_file else Path(log_dir).name
+    summary = to_plan_summary(dfs, log_name)
+
+    if format == AnalysisFormat.json:
+        output = json.dumps(summary, indent=2, default=str)
+    else:
+        lines: list[str] = [f"Log: {summary['log_name']}"]
+        totals = summary.get("totals", {})
+        lines.append(f"Queries: {len(summary.get('queries', []))}")
+        lines.append(f"Bytes read: {totals.get('bytes_read', 0):,}")
+        lines.append(f"Bytes written: {totals.get('bytes_written', 0):,}")
+        lines.append(f"Shuffle bytes read: {totals.get('shuffle_bytes_read', 0):,}")
+        lines.append(
+            f"Shuffle bytes written: {totals.get('shuffle_bytes_written', 0):,}"
+        )
+        lines.append(f"Memory spilled: {totals.get('memory_bytes_spilled', 0):,}")
+        lines.append(f"Disk spilled: {totals.get('disk_bytes_spilled', 0):,}")
+        for q in summary.get("queries", []):
+            lines.append(
+                f"\nQuery {q['query_id']} ({q['query_function']})"
+                f"  duration={q['duration_seconds']:.1f}s"
+                f"  nodes={len(q['nodes'])}"
+            )
+            for n in q["nodes"]:
+                dur = (
+                    f"{n['duration_minutes']:.3f}min"
+                    if n.get("duration_minutes") is not None
+                    else "?"
+                )
+                lines.append(f"  [{n['node_id']}] {n['node_type']}  {dur}")
+        output = "\n".join(lines)
+
+    if out_file is not None:
+        out_file.parent.mkdir(parents=True, exist_ok=True)
+        out_file.write_text(output)
+        typer.echo(f"Analysis written to {out_file}", err=True)
+    else:
+        sys.stdout.write(output + "\n")
 
 
 if __name__ == "__main__":

sparkparse/capture.py

@@ -1,4 +1,5 @@
 import functools
+import logging
 import os
 import shutil
 import subprocess
@@ -8,13 +9,18 @@
 import webbrowser
 from collections.abc import Callable
 from pathlib import Path
-from typing import Any, TypeVar, overload
+from typing import Any, Literal, TypeVar, overload
 
 from pyspark.sql import SparkSession
 
+from sparkparse.analyze import to_plan_summary
 from sparkparse.app import get
 from sparkparse.models import ParsedLogDataFrames
 
+_log = logging.getLogger(__name__)
+
+CaptureAction = Literal["viz", "get", "analyze"]
+
 R = TypeVar("R")
 F = TypeVar("F", bound=Callable[..., Any])
 
@@ -25,7 +31,7 @@ class SparkparseCapture:
 
     def __init__(
         self,
-        action: str,
+        action: CaptureAction,
         spark: SparkSession,
         temp_dir: str | None = None,
         headless: bool = False,
@@ -38,6 +44,7 @@ def __init__(
         self._should_cleanup = temp_dir is None
         self._headless = headless
         self._parsed_logs = None
+        self._analysis: dict[str, Any] | None = None
 
     def __call__(
         self, func: Callable[..., R]
@@ -84,8 +91,8 @@ def __enter__(self):
 
         self.spark = builder.getOrCreate()
 
-        print(f"Log enabled: {self.spark.conf.get('spark.eventLog.enabled')}")
-        print(f"Log dir config: {self.spark.conf.get('spark.eventLog.dir')}")
+        _log.info("Log enabled: %s", self.spark.conf.get("spark.eventLog.enabled"))
+        _log.info("Log dir config: %s", self.spark.conf.get("spark.eventLog.dir"))
         return self
 
     def _run_dashboard_in_background(self):
@@ -140,25 +147,32 @@ def __exit__(self, exc_type, *args):
                 raise ValueError("log directory is not set")
             result = get(log_dir=self._log_dir)
             self._parsed_logs = result
-
-            if (
-                self._should_cleanup
-                and self._log_dir is not None
-                and os.path.exists(self._log_dir)
-            ):
-                shutil.rmtree(self._log_dir)
+        elif self.action == "analyze":
+            if self._log_dir is None:
+                raise ValueError("log directory is not set")
+            result = get(log_dir=self._log_dir)
+            self._parsed_logs = result
+            log_name = Path(self._log_dir).name
+            self._analysis = to_plan_summary(result, log_name)
         else:
             raise ValueError(f"Invalid action: {self.action}")
 
+        if (
+            self._should_cleanup
+            and self._log_dir is not None
+            and os.path.exists(self._log_dir)
+        ):
+            shutil.rmtree(self._log_dir)
+
 
 def capture_context(
-    action: str = "viz",
+    action: CaptureAction = "viz",
     temp_dir: str | None = None,
     spark: SparkSession | None = None,
     headless: bool = False,
 ) -> SparkparseCapture:
     if spark is None:
-        _spark = SparkSession.builder.appName("temp").getOrCreate()  # type: ignore
+        _spark = SparkSession.builder.appName("sparkparse_capture").getOrCreate()  # type: ignore
     else:
         _spark = spark
 
@@ -169,7 +183,7 @@ def capture_context(
 def capture(
     func: Callable[..., R],
     *,
-    action: str = ...,
+    action: CaptureAction = ...,
     temp_dir: str | None = ...,
     spark: SparkSession | None = ...,
     headless: bool = ...,
@@ -180,7 +194,7 @@ def capture(
 def capture(
     func: None = None,
     *,
-    action: str = ...,
+    action: CaptureAction = ...,
     temp_dir: str | None = ...,
     spark: SparkSession | None = ...,
     headless: bool = ...,
@@ -190,7 +204,7 @@ def capture(
 def capture(
     func=None,
     *,
-    action: str = "viz",
+    action: CaptureAction = "viz",
     temp_dir: str | None = None,
     spark: SparkSession | None = None,
     headless: bool = False,
@@ -199,7 +213,7 @@ def decorator(
         func: Callable[..., R],
     ) -> Callable[..., tuple[Any, SparkparseCapture]]:
         if spark is None:
-            _spark = SparkSession.builder.appName("temp").getOrCreate()  # type: ignore
+            _spark = SparkSession.builder.appName("sparkparse_capture").getOrCreate()  # type: ignore
         else:
             _spark = spark