From ad6eef516607deca63f3b3e791e0d98e006a9b02 Mon Sep 17 00:00:00 2001 From: Jeff Brennan Date: Sat, 20 Jun 2026 13:15:38 -0400 Subject: [PATCH] add perf history and alerts --- pyproject.toml | 1 + sparkparse/alerts.py | 192 +++++++++++++++++++ sparkparse/app.py | 66 ++++++- sparkparse/capture.py | 68 ++++++- sparkparse/common.py | 5 +- sparkparse/history.py | 190 +++++++++++++++++++ sparkparse/models.py | 188 ++++++------------- sparkparse/storage.py | 11 ++ tests/test_alerts.py | 422 ++++++++++++++++++++++++++++++++++++++++++ tests/test_history.py | 190 +++++++++++++++++++ tests/test_storage.py | 1 - uv.lock | 100 +++++++++- 12 files changed, 1300 insertions(+), 134 deletions(-) create mode 100644 sparkparse/alerts.py create mode 100644 sparkparse/history.py create mode 100644 tests/test_alerts.py create mode 100644 tests/test_history.py diff --git a/pyproject.toml b/pyproject.toml index 3219aef..51da01c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,6 +26,7 @@ s3 = ["s3fs>=2024.1.0"] azure = ["adlfs>=2024.1.0"] gcs = ["gcsfs>=2024.1.0"] cloud = ["s3fs>=2024.1.0", "adlfs>=2024.1.0", "gcsfs>=2024.1.0"] +delta = ["deltalake>=0.25"] [tool.hatch.build.targets.wheel] diff --git a/sparkparse/alerts.py b/sparkparse/alerts.py new file mode 100644 index 0000000..022351c --- /dev/null +++ b/sparkparse/alerts.py @@ -0,0 +1,192 @@ +"""Regression detection against run history. + +Alert rules are defined in TOML (or passed as Python dicts) and evaluated +against the current ``RunRecord`` and historical records for the same +``log_name``. Three condition types are supported: + +- ``threshold`` — fire if the current metric exceeds a fixed value. +- ``pct_increase`` — fire if ``(current - baseline) / baseline > threshold``, + where baseline is the mean of the last ``window`` runs. +- ``absolute_increase`` — fire if ``current - baseline > threshold``. + +Triggered alerts dispatch via ``on_trigger``: ``"log"`` emits a log record, +``"raise"`` raises ``SparkparseAlertError``, ``"file"`` appends a JSON record +to ``alert_output_path``. +""" + +from __future__ import annotations + +import datetime +import json +import logging +import tomllib +from pathlib import Path +from typing import Literal + +import polars as pl +from pydantic import BaseModel + +from sparkparse.models import RunRecord +from sparkparse.storage import append_text, ensure_dir, is_cloud_path, open_file + +logger = logging.getLogger(__name__) + +_VALID_METRICS = set(RunRecord.model_fields.keys()) - {"run_id", "run_at", "log_name"} + + +class SparkparseAlertError(Exception): + """Raised when an alert with ``on_trigger = "raise"`` fires.""" + + def __init__( + self, + alert_name: str, + metric: str, + current: float, + baseline: float | None, + ) -> None: + self.alert_name = alert_name + self.metric = metric + self.current = current + self.baseline = baseline + super().__init__( + f"Alert '{alert_name}' triggered: {metric}={current} (baseline={baseline})" + ) + + +class AlertConfig(BaseModel): + name: str + log_name: str + metric: str + condition: Literal["pct_increase", "absolute_increase", "threshold"] + threshold: float + window: int = 10 + severity: Literal["warning", "critical"] = "warning" + on_trigger: Literal["log", "raise", "file"] = "log" + + +def load_alert_config(path: str) -> list[AlertConfig]: + """Load alert rules from a TOML file. + + The TOML file must contain an ``[[alerts]]`` table per rule. Local paths + and cloud URIs are both supported via ``storage.open_file``. + """ + with open_file(path, "rb") as f: + data = tomllib.load(f) + + raw_alerts = data.get("alerts", []) + return [AlertConfig(**a) for a in raw_alerts] + + +def _compute_baseline(history: pl.DataFrame, alert: AlertConfig, current_run_id: str) -> float: + """Mean of the alert metric over the last ``window`` runs, excluding the + current run. Returns 0.0 when no history is available. + """ + if history.is_empty() or "run_id" not in history.columns: + return 0.0 + + hist = ( + history.filter( + (pl.col("log_name") == alert.log_name) & (pl.col("run_id") != current_run_id) + ) + .sort("run_at", descending=True) + .head(alert.window) + ) + + if hist.is_empty(): + return 0.0 + + mean_val = hist[alert.metric].mean() + return float(mean_val) if mean_val is not None else 0.0 + + +def _dispatch(alert: AlertConfig, alert_dict: dict, alert_output_path: str | None) -> None: + if alert.on_trigger == "raise": + raise SparkparseAlertError( + alert_name=alert.name, + metric=alert.metric, + current=alert_dict["current"], + baseline=alert_dict.get("baseline"), + ) + + if alert.on_trigger == "log": + msg = f"Alert '{alert.name}' triggered: {alert.metric}={alert_dict['current']}" + if alert_dict.get("baseline") is not None: + msg += f" (baseline={alert_dict['baseline']})" + if alert.severity == "critical": + logger.error(msg) + else: + logger.warning(msg) + + if alert.on_trigger == "file": + if alert_output_path is None: + logger.error( + "Alert '%s' has on_trigger='file' but no alert_output_path set", + alert.name, + ) + return + if not is_cloud_path(alert_output_path): + ensure_dir(Path(alert_output_path).parent) + append_text(alert_output_path, json.dumps(alert_dict, default=str) + "\n") + + +def check_alerts( + record: RunRecord, + history: pl.DataFrame, + alerts: list[AlertConfig], + alert_output_path: str | None = None, +) -> list[dict]: + """Evaluate all alert rules against the current record and history. + + Fires ``on_trigger`` actions for any triggered alerts. Returns a list of + triggered alert dicts (empty if none triggered). + """ + triggered: list[dict] = [] + + for alert in alerts: + if alert.log_name != record.log_name: + continue + + if alert.metric not in _VALID_METRICS: + logger.warning( + "Alert '%s' references unknown metric '%s', skipping", + alert.name, + alert.metric, + ) + continue + + current = float(getattr(record, alert.metric)) + baseline: float | None = None + + if alert.condition == "threshold": + fired = current > alert.threshold + else: + baseline = _compute_baseline(history, alert, record.run_id) + if alert.condition == "pct_increase": + if baseline == 0: + fired = current > 0 + else: + fired = (current - baseline) / baseline > alert.threshold + elif alert.condition == "absolute_increase": + fired = current - baseline > alert.threshold + else: + fired = False + + if not fired: + continue + + alert_dict = { + "alert_name": alert.name, + "log_name": alert.log_name, + "metric": alert.metric, + "condition": alert.condition, + "threshold": alert.threshold, + "severity": alert.severity, + "current": current, + "baseline": baseline, + "triggered_at": datetime.datetime.now(datetime.UTC).isoformat(), + } + + _dispatch(alert, alert_dict, alert_output_path) + triggered.append(alert_dict) + + return triggered diff --git a/sparkparse/app.py b/sparkparse/app.py index c060383..0a0547c 100644 --- a/sparkparse/app.py +++ b/sparkparse/app.py @@ -6,9 +6,10 @@ import typer +from sparkparse import alerts, history from sparkparse.analyze import to_plan_summary from sparkparse.dashboard import init_dashboard, run_app -from sparkparse.models import OutputFormat, ParsedLogDataFrames +from sparkparse.models import OutputFormat, ParsedLogDataFrames, RunRecord from sparkparse.parse import get_parsed_metrics from sparkparse.storage import ( get_path_name, @@ -27,6 +28,11 @@ class AnalysisFormat(StrEnum): text = "text" +class HistoryFormat(StrEnum): + table = "table" + json = "json" + + def _version_callback(value: bool) -> None: if value: typer.echo(f"sparkparse {__version__}") @@ -161,5 +167,63 @@ def analyze( sys.stdout.write(output + "\n") +@app.command("history") +def history_cmd( + history_path: Annotated[ + str, typer.Argument(help="Path to history store (Delta dir or JSONL file).") + ], + log_name: Annotated[ + str | None, + typer.Option(help="Filter records to this log_name only."), + ] = None, + last: Annotated[ + int | None, typer.Option(help="Show only the last N runs (most recent first).") + ] = None, + format: Annotated[ + HistoryFormat, + typer.Option(help="Output format: 'table' (default) or 'json'."), + ] = HistoryFormat.table, +) -> None: + """Query the run history store for past job metrics.""" + df = history.read(history_path, log_name=log_name, last_n=last) + + if df.is_empty(): + typer.echo("No history records found.") + return + + if format == HistoryFormat.json: + typer.echo(json.dumps(df.to_dicts(), default=str, indent=2)) + else: + typer.echo(str(df)) + + +@app.command("check-alerts") +def check_alerts_cmd( + history_path: Annotated[ + str, typer.Argument(help="Path to history store (Delta dir or JSONL file).") + ], + log_name: Annotated[str, typer.Argument(help="Stable job identifier to check alerts for.")], + alert_config: Annotated[str, typer.Argument(help="Path to alert configuration TOML file.")], + alert_output_path: Annotated[ + str | None, + typer.Option(help="File to write triggered alerts (for on_trigger='file' rules)."), + ] = None, +) -> None: + """Run alert checks against the latest run in history.""" + hist_df = history.read(history_path, log_name=log_name) + + if hist_df.is_empty(): + typer.echo(f"No history records found for log_name '{log_name}'.", err=True) + raise typer.Exit(1) + + latest_row = hist_df.sort("run_at").row(-1, named=True) + latest = RunRecord(**latest_row) + + rules = alerts.load_alert_config(alert_config) + triggered = alerts.check_alerts(latest, hist_df, rules, alert_output_path) + + typer.echo(json.dumps(triggered, indent=2, default=str)) + + if __name__ == "__main__": app() diff --git a/sparkparse/capture.py b/sparkparse/capture.py index 0a33dff..2abfc69 100644 --- a/sparkparse/capture.py +++ b/sparkparse/capture.py @@ -11,9 +11,10 @@ from pyspark.sql import SparkSession +from sparkparse import alerts, history from sparkparse.analyze import to_plan_summary from sparkparse.app import get -from sparkparse.models import ParsedLogDataFrames +from sparkparse.models import ParsedLogDataFrames, RunRecord from sparkparse.storage import ( copy_file, ensure_dir, @@ -43,6 +44,9 @@ def __init__( spark: SparkSession, temp_dir: str | None = None, headless: bool = False, + history_path: str | None = None, + log_name: str | None = None, + alert_config: str | None = None, ) -> None: self.action = action self.temp_dir = temp_dir @@ -53,6 +57,11 @@ def __init__( self._headless = headless self._parsed_logs = None self._analysis: dict[str, Any] | None = None + self._history_path = history_path + self._log_name = log_name + self._alert_config = alert_config + self._last_record: RunRecord | None = None + self._triggered_alerts: list[dict] = [] def __call__(self, func: Callable[..., R]) -> Callable[..., tuple[R, "SparkparseCapture"]]: @functools.wraps(func) @@ -126,6 +135,28 @@ def _run_dashboard_in_background(self): time.sleep(2) webbrowser.open("http://127.0.0.1:8050/") + def _record_history_and_alerts(self) -> None: + if self._parsed_logs is None or self._history_path is None: + return + + effective_log_name = self._log_name or get_path_name(self._log_dir) + record = history.record_from_dfs(self._parsed_logs, effective_log_name) + self._last_record = record + + history.append(record, self._history_path) + _log.info("Appended run record %s to %s", record.run_id, self._history_path) + + if self._alert_config is not None: + rules = alerts.load_alert_config(self._alert_config) + hist_df = history.read(self._history_path, effective_log_name) + self._triggered_alerts = alerts.check_alerts(record, hist_df, rules) + if self._triggered_alerts: + _log.warning( + "%d alert(s) triggered for %s", + len(self._triggered_alerts), + effective_log_name, + ) + def __exit__(self, exc_type, *args): self.spark.stop() self.spark = self._orig_spark @@ -163,6 +194,9 @@ def __exit__(self, exc_type, *args): else: raise ValueError(f"Invalid action: {self.action}") + if self._history_path is not None: + self._record_history_and_alerts() + if self._should_cleanup and self._log_dir is not None and path_exists(self._log_dir): remove_dir(self._log_dir) @@ -172,13 +206,24 @@ def capture_context( temp_dir: str | None = None, spark: SparkSession | None = None, headless: bool = False, + history_path: str | None = None, + log_name: str | None = None, + alert_config: str | None = None, ) -> SparkparseCapture: if spark is None: _spark = SparkSession.builder.appName("sparkparse_capture").getOrCreate() # type: ignore else: _spark = spark - return SparkparseCapture(action, temp_dir=temp_dir, spark=_spark, headless=headless) + return SparkparseCapture( + action, + temp_dir=temp_dir, + spark=_spark, + headless=headless, + history_path=history_path, + log_name=log_name, + alert_config=alert_config, + ) @overload @@ -189,6 +234,9 @@ def capture( temp_dir: str | None = ..., spark: SparkSession | None = ..., headless: bool = ..., + history_path: str | None = ..., + log_name: str | None = ..., + alert_config: str | None = ..., ) -> Callable[..., tuple[R, SparkparseCapture]]: ... @@ -200,6 +248,9 @@ def capture( temp_dir: str | None = ..., spark: SparkSession | None = ..., headless: bool = ..., + history_path: str | None = ..., + log_name: str | None = ..., + alert_config: str | None = ..., ) -> Callable[[Callable[..., R]], Callable[..., tuple[R, SparkparseCapture]]]: ... @@ -210,6 +261,9 @@ def capture( temp_dir: str | None = None, spark: SparkSession | None = None, headless: bool = False, + history_path: str | None = None, + log_name: str | None = None, + alert_config: str | None = None, ) -> Any: def decorator( func: Callable[..., R], @@ -219,7 +273,15 @@ def decorator( else: _spark = spark - cap = SparkparseCapture(action, spark=_spark, temp_dir=temp_dir, headless=headless) + cap = SparkparseCapture( + action, + spark=_spark, + temp_dir=temp_dir, + headless=headless, + history_path=history_path, + log_name=log_name, + alert_config=alert_config, + ) return cap(func) if func is None: diff --git a/sparkparse/common.py b/sparkparse/common.py index b579e68..753d3a0 100644 --- a/sparkparse/common.py +++ b/sparkparse/common.py @@ -29,7 +29,10 @@ def timeit_wrapper(*args, **kwargs): def write_dataframe( - df: pl.DataFrame, out_path: str | Path, out_format: OutputFormat, overwrite: bool = True + df: pl.DataFrame, + out_path: str | Path, + out_format: OutputFormat, + overwrite: bool = True, ) -> None: out_path_str = str(out_path) cloud = is_cloud_path(out_path_str) diff --git a/sparkparse/history.py b/sparkparse/history.py new file mode 100644 index 0000000..6715903 --- /dev/null +++ b/sparkparse/history.py @@ -0,0 +1,190 @@ +"""Append-only run history for tracking Spark job metrics over time. + +Each call to ``record_from_dfs`` derives a compact ``RunRecord`` — the numeric +snapshot you would want to plot as a time series — from a +``ParsedLogDataFrames``. ``append`` writes it to a persistent store; ``read`` +queries it back for trend analysis and alert baseline computation. + +Storage format is selected automatically: Delta Lake (append-only, ACID, +supports concurrent writers) when ``deltalake`` is importable, JSONL otherwise. +Format can be forced via the ``format`` parameter. +""" + +from __future__ import annotations + +import datetime +import importlib.util +import logging +import uuid +from pathlib import Path + +import polars as pl + +from sparkparse.analyze import find_cartesian_joins, find_largest_scans +from sparkparse.models import ParsedLogDataFrames, RunRecord +from sparkparse.storage import ( + append_text, + ensure_dir, + is_cloud_path, + open_file, + path_exists, +) + +logger = logging.getLogger(__name__) + +HistoryFormat = str # "auto" | "delta" | "jsonl" + +_TS_FORMAT = "%Y-%m-%dT%H:%M:%S" + + +def _delta_available() -> bool: + return importlib.util.find_spec("deltalake") is not None + + +def _resolve_format(format: HistoryFormat) -> str: + if format not in ("auto", "delta", "jsonl"): + raise ValueError(f"Unsupported history format: {format!r}") + if format == "auto": + return "delta" if _delta_available() else "jsonl" + if format == "delta" and not _delta_available(): + raise ImportError( + "Delta format requested but deltalake is not installed. " + "Install it with `uv add deltalake` or `pip install deltalake`." + ) + return format + + +def _resolve_read_format(format: HistoryFormat, history_path: str) -> str: + """For reads, detect JSONL from the path extension before falling back to + the deltalake availability check. This lets ``read`` work on a JSONL file + even when deltalake is installed. + """ + if format != "auto": + return _resolve_format(format) + if str(history_path).endswith(".jsonl"): + return "jsonl" + return "delta" if _delta_available() else "jsonl" + + +def record_from_dfs(dfs: ParsedLogDataFrames, log_name: str) -> RunRecord: + """Derive a ``RunRecord`` from parsed DataFrames. + + Reuses ``analyze.find_cartesian_joins`` and ``analyze.find_largest_scans`` + so the metric definitions stay consistent with the rest of the codebase. + """ + dag = dfs.dag + combined = dfs.combined + + start_ts = dag["query_start_timestamp"].str.to_datetime(format=_TS_FORMAT).min() + end_ts = dag["query_end_timestamp"].str.to_datetime(format=_TS_FORMAT).max() + if start_ts is not None and end_ts is not None: + duration_s = (end_ts - start_ts).total_seconds() + else: + duration_s = 0.0 + + totals = combined.select( + pl.sum("bytes_read").alias("bytes_read"), + pl.sum("bytes_written").alias("bytes_written"), + pl.sum("shuffle_bytes_read").alias("shuffle_bytes_read"), + pl.sum("shuffle_bytes_written").alias("shuffle_bytes_written"), + pl.sum("memory_bytes_spilled").alias("memory_bytes_spilled"), + pl.sum("disk_bytes_spilled").alias("disk_bytes_spilled"), + ).row(0, named=True) + + n_cartesian = find_cartesian_joins(dfs).height + + max_node_dur = dag["node_duration_minutes"].max() + if max_node_dur is None: + max_node_dur = 0.0 + + largest_scans = find_largest_scans(dfs, n=1) + if largest_scans.is_empty(): + max_scan_bytes = 0 + else: + val = largest_scans["bytes_read"][0] + max_scan_bytes = int(val) if val is not None else 0 + + return RunRecord( + run_id=uuid.uuid4().hex, + run_at=datetime.datetime.now(datetime.UTC), + log_name=log_name, + duration_s=duration_s, + bytes_read=int(totals["bytes_read"] or 0), + bytes_written=int(totals["bytes_written"] or 0), + shuffle_bytes=int( + (totals["shuffle_bytes_read"] or 0) + (totals["shuffle_bytes_written"] or 0) + ), + spill_bytes=int( + (totals["memory_bytes_spilled"] or 0) + (totals["disk_bytes_spilled"] or 0) + ), + n_queries=dag["query_id"].n_unique(), + n_stages=combined["stage_id"].n_unique(), + n_tasks=combined.height, + n_cartesian_joins=n_cartesian, + max_node_duration_min=float(max_node_dur), + max_scan_bytes=max_scan_bytes, + ) + + +def append(record: RunRecord, history_path: str, format: HistoryFormat = "auto") -> None: + """Append a ``RunRecord`` to the history store at ``history_path``. + + For Delta, ``history_path`` is a directory. For JSONL, it is a file path. + Local and cloud URIs are both supported via ``storage`` helpers. + """ + resolved = _resolve_format(format) + + if resolved == "delta": + from deltalake import write_deltalake + + df = pl.DataFrame([record.model_dump()]) + table = df.to_arrow() + write_deltalake(history_path, table, mode="append") + logger.info("Appended run %s to Delta table at %s", record.run_id, history_path) + else: + if not is_cloud_path(history_path): + ensure_dir(Path(history_path).parent) + append_text(history_path, record.model_dump_json() + "\n") + logger.info("Appended run %s to JSONL at %s", record.run_id, history_path) + + +def read( + history_path: str, + log_name: str | None = None, + last_n: int | None = None, + format: HistoryFormat = "auto", +) -> pl.DataFrame: + """Read history records, optionally filtered by ``log_name`` and/or limited + to the last ``N`` runs (most recent by ``run_at``). + """ + resolved = _resolve_read_format(format, history_path) + + if resolved == "delta": + from deltalake import DeltaTable + + dt = DeltaTable(history_path) + df = pl.from_arrow(dt.to_pyarrow_table()) + else: + if not path_exists(history_path): + return pl.DataFrame() + if is_cloud_path(history_path): + with open_file(history_path, "rb") as f: + df = pl.read_ndjson(f) + else: + df = pl.read_ndjson(history_path) + + if df.is_empty(): + return df + + if df["run_at"].dtype == pl.String: + df = df.with_columns(pl.col("run_at").str.to_datetime()) + + if log_name is not None: + df = df.filter(pl.col("log_name") == log_name) + + df = df.sort("run_at") + + if last_n is not None and last_n > 0: + df = df.tail(last_n) + + return df diff --git a/sparkparse/models.py b/sparkparse/models.py index bfce3c5..3f5f3af 100644 --- a/sparkparse/models.py +++ b/sparkparse/models.py @@ -1,3 +1,4 @@ +import datetime import json from enum import StrEnum, auto from typing import Annotated, Any @@ -157,9 +158,7 @@ class ScanDetailLocation(BaseModel): class ScanDetail(BaseModel): - output: Annotated[ - list[str] | None, Field(alias="Output"), BeforeValidator(str_to_list) - ] + output: Annotated[list[str] | None, Field(alias="Output"), BeforeValidator(str_to_list)] batched: bool = Field(alias="Batched") location: ScanDetailLocation = Field(alias="Location") read_schema: str = Field(alias="ReadSchema") @@ -182,16 +181,12 @@ def deserialize_scan_detail(s: str) -> ScanDetail: class ColumnarToRowDetail(BaseModel): - input: Annotated[ - list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list) - ] + input: Annotated[list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list)] class ProjectDetail(BaseModel): input: Annotated[list[str], Field(alias="Input"), BeforeValidator(str_to_list)] - output: Annotated[ - list[str] | None, Field(alias="Output"), BeforeValidator(str_to_list) - ] + output: Annotated[list[str] | None, Field(alias="Output"), BeforeValidator(str_to_list)] class Function(BaseModel): @@ -200,9 +195,7 @@ class Function(BaseModel): class HashAggregateDetail(BaseModel): - input: Annotated[ - list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list) - ] + input: Annotated[list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list)] keys: Annotated[list[str] | None, Field(alias="Keys"), BeforeValidator(str_to_list)] functions: list[Function] | None = Field(alias="Functions") aggregate_attributes: Annotated[ @@ -210,9 +203,7 @@ class HashAggregateDetail(BaseModel): Field(alias="Aggregate Attributes"), BeforeValidator(str_to_list), ] - results: Annotated[ - list[str] | None, Field(alias="Results"), BeforeValidator(str_to_list) - ] + results: Annotated[list[str] | None, Field(alias="Results"), BeforeValidator(str_to_list)] @field_validator("functions", mode="before") @classmethod @@ -242,9 +233,7 @@ class ExchangeArgument(BaseModel): class ExchangeDetail(BaseModel): - input: Annotated[ - list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list) - ] + input: Annotated[list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list)] arguments: ExchangeArgument = Field(alias="Arguments") @field_validator("arguments", mode="before") @@ -258,9 +247,7 @@ def parse_exchange_argument_str(cls, value: Any) -> ExchangeArgument: plan_identifier=int(value.split("plan_id=")[1].removesuffix("]")), ) - hash_partition_section = ( - value.split("), ")[0].removeprefix("hashpartitioning(").split(", ") - ) + hash_partition_section = value.split("), ")[0].removeprefix("hashpartitioning(").split(", ") cols = [i for i in hash_partition_section if "#" in i] n_partitions = hash_partition_section[-1].strip() @@ -276,9 +263,7 @@ def parse_exchange_argument_str(cls, value: Any) -> ExchangeArgument: class ShuffleQueryStageDetail(BaseModel): - output: Annotated[ - list[str] | None, Field(alias="Output"), BeforeValidator(str_to_list) - ] + output: Annotated[list[str] | None, Field(alias="Output"), BeforeValidator(str_to_list)] stage_order: int = Field(alias="Arguments") @@ -288,9 +273,7 @@ class AQEShuffleReadArgument(StrEnum): class AQEShuffleReadDetail(BaseModel): - input: Annotated[ - list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list) - ] + input: Annotated[list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list)] arguments: AQEShuffleReadArgument = Field(alias="Arguments") @@ -319,9 +302,7 @@ def parse_sort_argument_col_str(col_section: str) -> list[SortArgumentCol]: class SortDetail(BaseModel): - input: Annotated[ - list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list) - ] + input: Annotated[list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list)] arguments: SortArgument = Field(alias="Arguments") @field_validator("arguments", mode="before") @@ -333,9 +314,7 @@ def parse_sort_argument_str(cls, value: Any) -> SortArgument: global_sort = value.split("]")[1].strip() == "true" sort_order = value.split(",")[-1].strip() - return SortArgument( - cols=cols, global_sort=global_sort, sort_order=int(sort_order) - ) + return SortArgument(cols=cols, global_sort=global_sort, sort_order=int(sort_order)) class JoinType(StrEnum): @@ -349,12 +328,8 @@ class JoinType(StrEnum): class SortMergeJoinDetail(BaseModel): - left_keys: Annotated[ - list[str] | None, Field(alias="Left keys"), BeforeValidator(str_to_list) - ] - right_keys: Annotated[ - list[str] | None, Field(alias="Right keys"), BeforeValidator(str_to_list) - ] + left_keys: Annotated[list[str] | None, Field(alias="Left keys"), BeforeValidator(str_to_list)] + right_keys: Annotated[list[str] | None, Field(alias="Right keys"), BeforeValidator(str_to_list)] join_type: JoinType = Field(alias="Join type") join_condition: str | None = Field(alias="Join condition", default=None) @@ -377,9 +352,7 @@ class WindowDetailArgument(BaseModel): class WindowDetail(BaseModel): - input: Annotated[ - list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list) - ] + input: Annotated[list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list)] arguments: WindowDetailArgument = Field(alias="Arguments") @field_validator("arguments", mode="before") @@ -389,18 +362,14 @@ def parse_window_detail_argument_str(cls, value: Any) -> WindowDetailArgument: window_function_col = function_section.split("(")[1].removesuffix(")") window_function_col = None if window_function_col == "" else window_function_col - window_function = Function( - function=function_section.split("(")[0], col=window_function_col - ) + window_function = Function(function=function_section.split("(")[0], col=window_function_col) window_specification = value.split("windowspecdefinition(")[1].split("], ")[0] pre_frame_section = window_specification.split(", specifiedwindowframe")[0] partition_cols = [ i for i in pre_frame_section.split(", ") if not ("DESC" in i or "ASC" in i) ] - order_cols = [ - i for i in pre_frame_section.split(", ") if "DESC" in i or "ASC" in i - ] + order_cols = [i for i in pre_frame_section.split(", ") if "DESC" in i or "ASC" in i] order_cols_parsed = [] for col in order_cols: @@ -412,9 +381,7 @@ def parse_window_detail_argument_str(cls, value: Any) -> WindowDetailArgument: ) ) - window_frame = window_specification.split("specifiedwindowframe(")[1].split( - "], " - )[0] + window_frame = window_specification.split("specifiedwindowframe(")[1].split("], ")[0] window_frame = "specifiedwindowframe(" + window_frame.replace(")))", "))") @@ -442,16 +409,12 @@ class WindowGroupLimitArgument(BaseModel): class WindowGroupLimitDetail(BaseModel): - input: Annotated[ - list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list) - ] + input: Annotated[list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list)] arguments: WindowGroupLimitArgument = Field(alias="Arguments") @field_validator("arguments", mode="before") @classmethod - def parse_window_group_limit_argument_str( - cls, value: Any - ) -> WindowGroupLimitArgument: + def parse_window_group_limit_argument_str(cls, value: Any) -> WindowGroupLimitArgument: partition_cols = value.split("], ")[0].removeprefix("[").strip().split(", ") order_section = value.split("], ")[1].removeprefix("[").strip().split(", ") @@ -513,16 +476,12 @@ class FilterDetailCondition(BaseModel): class FilterDetail(BaseModel): - input: Annotated[ - list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list) - ] + input: Annotated[list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list)] condition: str = Field(alias="Condition") class CoalesceDetail(BaseModel): - input: Annotated[ - list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list) - ] + input: Annotated[list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list)] n_partitions: int = Field(alias="Arguments") @@ -548,12 +507,8 @@ class InsertIntoHadoopFsRelationCommandDetailArguments(BaseModel): class InsertIntoHadoopFsRelationCommandDetail(BaseModel): - input: Annotated[ - list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list) - ] - arguments: InsertIntoHadoopFsRelationCommandDetailArguments = Field( - alias="Arguments" - ) + input: Annotated[list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list)] + arguments: InsertIntoHadoopFsRelationCommandDetailArguments = Field(alias="Arguments") @field_validator("arguments", mode="before") @classmethod @@ -594,10 +549,8 @@ def deserialize_insert_into_hadoop_fs_relation_command_detail( s: str, ) -> InsertIntoHadoopFsRelationCommandDetail: data = json.loads(s)["detail"] - data["arguments"] = ( - InsertIntoHadoopFsRelationCommandDetailArguments.model_construct( - **data["arguments"] - ) + data["arguments"] = InsertIntoHadoopFsRelationCommandDetailArguments.model_construct( + **data["arguments"] ) return InsertIntoHadoopFsRelationCommandDetail.model_construct(**data) @@ -608,16 +561,12 @@ class LocalTableScanArguments(BaseModel): class LocalTableScanDetail(BaseModel): - output: Annotated[ - list[str] | None, Field(alias="Output"), BeforeValidator(str_to_list) - ] + output: Annotated[list[str] | None, Field(alias="Output"), BeforeValidator(str_to_list)] arguments: LocalTableScanArguments = Field(alias="Arguments") @field_validator("arguments", mode="before") @classmethod - def parse_local_table_scan_arguments_str( - cls, value: Any - ) -> LocalTableScanArguments: + def parse_local_table_scan_arguments_str(cls, value: Any) -> LocalTableScanArguments: contents = value.split(", [")[0] input_col_section = "[" + value.split(", [")[1] input_cols = str_to_list(input_col_section) @@ -625,15 +574,11 @@ def parse_local_table_scan_arguments_str( class WriteFilesDetail(BaseModel): - input: Annotated[ - list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list) - ] + input: Annotated[list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list)] class LocalLimitDetail(BaseModel): - input: Annotated[ - list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list) - ] + input: Annotated[list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list)] limit: int = Field(alias="Arguments") @@ -643,9 +588,7 @@ class GlobalLimitArguments(BaseModel): class GlobalLimitDetail(BaseModel): - input: Annotated[ - list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list) - ] + input: Annotated[list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list)] arguments: GlobalLimitArguments = Field(alias="Arguments") @field_validator("arguments", mode="before") @@ -670,16 +613,12 @@ class BroadcastExchangeArguments(BaseModel): class BroadcastExchangeDetail(BaseModel): - input: Annotated[ - list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list) - ] + input: Annotated[list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list)] arguments: BroadcastExchangeArguments = Field(alias="Arguments") @field_validator("arguments", mode="before") @classmethod - def parse_broadcast_exchange_arguments_str( - cls, value: Any - ) -> BroadcastExchangeArguments: + def parse_broadcast_exchange_arguments_str(cls, value: Any) -> BroadcastExchangeArguments: value_split = value.split(", ") mode = BroadcastExchangeMode(value_split[0].split("(")[0]) plan_identifier = int( @@ -712,19 +651,13 @@ def parse_broadcast_exchange_arguments_str( class BroadcastQueryStageDetail(BaseModel): - output: Annotated[ - list[str] | None, Field(alias="Output"), BeforeValidator(str_to_list) - ] + output: Annotated[list[str] | None, Field(alias="Output"), BeforeValidator(str_to_list)] stage_order: int = Field(alias="Arguments") class BroadcastHashJoinDetail(BaseModel): - left_keys: Annotated[ - list[str], Field(alias="Left keys"), BeforeValidator(str_to_list) - ] - right_keys: Annotated[ - list[str], Field(alias="Right keys"), BeforeValidator(str_to_list) - ] + left_keys: Annotated[list[str], Field(alias="Left keys"), BeforeValidator(str_to_list)] + right_keys: Annotated[list[str], Field(alias="Right keys"), BeforeValidator(str_to_list)] join_type: JoinType = Field(alias="Join type") join_condition: str | None = Field(alias="Join condition", default=None) @@ -742,9 +675,7 @@ class TakeOrderedAndProjectDetailArguments(BaseModel): class TakeOrderedAndProjectDetail(BaseModel): - input: Annotated[ - list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list) - ] + input: Annotated[list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list)] arguments: TakeOrderedAndProjectDetailArguments = Field(alias="Arguments") @field_validator("arguments", mode="before") @@ -757,9 +688,7 @@ def parse_take_ordered_and_project_arguments_str( cols = parse_sort_argument_col_str(cols_raw) output = str_to_list(value.split("], [")[-1]) - return TakeOrderedAndProjectDetailArguments( - limit=limit, cols=cols, output=output - ) + return TakeOrderedAndProjectDetailArguments(limit=limit, cols=cols, output=output) class BroadcastNestedLoopJoinDetail(BaseModel): @@ -774,9 +703,7 @@ def parse_join_condition_str(cls, value: Any) -> str | None: class ReusedExchangeDetail(BaseModel): - output: Annotated[ - list[str] | None, Field(alias="Output"), BeforeValidator(str_to_list) - ] + output: Annotated[list[str] | None, Field(alias="Output"), BeforeValidator(str_to_list)] reuses_node_id: int = Field(alias="reuses_node_id") @@ -787,9 +714,7 @@ class GenerateDetailArguments(BaseModel): class GenerateDetail(BaseModel): - input: Annotated[ - list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list) - ] + input: Annotated[list[str] | None, Field(alias="Input"), BeforeValidator(str_to_list)] arguments: GenerateDetailArguments = Field(alias="Arguments") @field_validator("arguments", mode="before") @@ -805,16 +730,12 @@ def parse_generate_detail_arguments_str(cls, value: Any) -> GenerateDetailArgume class TableCacheQueryStageDetail(BaseModel): - output: Annotated[ - list[str] | None, Field(alias="Output"), BeforeValidator(str_to_list) - ] + output: Annotated[list[str] | None, Field(alias="Output"), BeforeValidator(str_to_list)] stage_order: int = Field(alias="Arguments") class InMemoryTableScanDetail(BaseModel): - output: Annotated[ - list[str] | None, Field(alias="Output"), BeforeValidator(str_to_list) - ] + output: Annotated[list[str] | None, Field(alias="Output"), BeforeValidator(str_to_list)] class InMemoryRelationDetail(BaseModel): @@ -823,9 +744,7 @@ class InMemoryRelationDetail(BaseModel): class AdaptiveSparkPlanDetail(BaseModel): - output: Annotated[ - list[str] | None, Field(alias="Output"), BeforeValidator(str_to_list) - ] + output: Annotated[list[str] | None, Field(alias="Output"), BeforeValidator(str_to_list)] is_final_plan: bool = Field(alias="Arguments") @field_validator("is_final_plan", mode="before") @@ -915,9 +834,7 @@ class PushBasedShuffle(BaseModel): merged_local_chunks_fetched: int = Field(alias="Merged Local Chunks Fetched") merged_remote_bytes_read: int = Field(alias="Merged Remote Bytes Read") merged_local_bytes_read: int = Field(alias="Merged Local Bytes Read") - merged_remote_requests_duration: int = Field( - alias="Merged Remote Requests Duration" - ) + merged_remote_requests_duration: int = Field(alias="Merged Remote Requests Duration") class ShuffleReadMetrics(BaseModel): @@ -1050,3 +967,20 @@ class ParsedLogDataFrames(BaseModel): NodeType.InMemoryRelation: InMemoryRelationDetail, NodeType.AdaptiveSparkPlan: AdaptiveSparkPlanDetail, } + + +class RunRecord(BaseModel): + run_id: str + run_at: datetime.datetime + log_name: str + duration_s: float + bytes_read: int + bytes_written: int + shuffle_bytes: int + spill_bytes: int + n_queries: int + n_stages: int + n_tasks: int + n_cartesian_joins: int + max_node_duration_min: float + max_scan_bytes: int diff --git a/sparkparse/storage.py b/sparkparse/storage.py index ccc31e6..572f00f 100644 --- a/sparkparse/storage.py +++ b/sparkparse/storage.py @@ -69,6 +69,17 @@ def write_text(path: str | Path, content: str) -> None: f.write(content) +def append_text(path: str | Path, content: str) -> None: + """Append ``content`` to ``path`` as UTF-8 text, local or cloud. + + Creates the file if it does not exist. For cloud object stores there is + no true append primitive, so each call rewrites the object; Delta is the + recommended format for cloud history stores with frequent appends. + """ + with open_file(path, "a") as f: + f.write(content) + + def read_text(path: str | Path) -> str: """Read and return the full text content of ``path``.""" with open_file(path, "r") as f: diff --git a/tests/test_alerts.py b/tests/test_alerts.py new file mode 100644 index 0000000..d23a73a --- /dev/null +++ b/tests/test_alerts.py @@ -0,0 +1,422 @@ +import datetime +import io +import json +import uuid +from pathlib import Path +from unittest.mock import patch + +import polars as pl +import pytest + +from sparkparse.alerts import ( + AlertConfig, + SparkparseAlertError, + check_alerts, + load_alert_config, +) +from sparkparse.models import RunRecord + + +def _make_record(**overrides) -> RunRecord: + defaults = dict( + run_id=uuid.uuid4().hex, + run_at=datetime.datetime.now(datetime.UTC), + log_name="test_job", + duration_s=100.0, + bytes_read=1000, + bytes_written=500, + shuffle_bytes=200, + spill_bytes=0, + n_queries=2, + n_stages=3, + n_tasks=50, + n_cartesian_joins=0, + max_node_duration_min=1.5, + max_scan_bytes=800, + ) + defaults.update(overrides) + return RunRecord(**defaults) + + +def _history_df(records: list[RunRecord]) -> pl.DataFrame: + return pl.DataFrame([r.model_dump() for r in records]) + + +def test_load_alert_config_from_toml(tmp_path: Path): + toml_content = """ +[[alerts]] +name = "duration_regression" +log_name = "nightly_job" +metric = "duration_s" +condition = "pct_increase" +threshold = 0.20 +window = 5 +severity = "warning" +on_trigger = "log" + +[[alerts]] +name = "spill_alert" +log_name = "nightly_job" +metric = "spill_bytes" +condition = "threshold" +threshold = 1073741824 +severity = "critical" +on_trigger = "raise" +""" + toml_path = tmp_path / "alerts.toml" + toml_path.write_text(toml_content) + + alerts = load_alert_config(str(toml_path)) + assert len(alerts) == 2 + assert alerts[0].name == "duration_regression" + assert alerts[0].condition == "pct_increase" + assert alerts[0].window == 5 + assert alerts[1].name == "spill_alert" + assert alerts[1].condition == "threshold" + assert alerts[1].severity == "critical" + assert alerts[1].on_trigger == "raise" + + +def test_load_alert_config_cloud_path(): + toml_bytes = b""" +[[alerts]] +name = "cloud_alert" +log_name = "my_job" +metric = "duration_s" +condition = "threshold" +threshold = 300.0 +""" + fake_file = io.BytesIO(toml_bytes) + with patch("sparkparse.alerts.open_file", return_value=fake_file): + alerts = load_alert_config("s3://bucket/alerts.toml") + assert len(alerts) == 1 + assert alerts[0].name == "cloud_alert" + + +def test_load_alert_config_empty(): + fake_file = io.BytesIO(b"") + with patch("sparkparse.alerts.open_file", return_value=fake_file): + alerts = load_alert_config("s3://bucket/empty.toml") + assert alerts == [] + + +def test_check_alerts_threshold_fires(): + record = _make_record(spill_bytes=200) + alert = AlertConfig( + name="spill_check", + log_name="test_job", + metric="spill_bytes", + condition="threshold", + threshold=100, + on_trigger="log", + ) + result = check_alerts(record, pl.DataFrame(), [alert]) + assert len(result) == 1 + assert result[0]["alert_name"] == "spill_check" + assert result[0]["current"] == 200 + assert result[0]["baseline"] is None + + +def test_check_alerts_threshold_does_not_fire(): + record = _make_record(spill_bytes=50) + alert = AlertConfig( + name="spill_check", + log_name="test_job", + metric="spill_bytes", + condition="threshold", + threshold=100, + on_trigger="log", + ) + result = check_alerts(record, pl.DataFrame(), [alert]) + assert result == [] + + +def test_check_alerts_pct_increase_fires(): + current = _make_record(duration_s=130.0) + history = _history_df( + [ + _make_record(duration_s=100.0), + _make_record(duration_s=100.0), + _make_record(duration_s=100.0), + ] + ) + alert = AlertConfig( + name="duration_regression", + log_name="test_job", + metric="duration_s", + condition="pct_increase", + threshold=0.20, + window=5, + on_trigger="log", + ) + result = check_alerts(current, history, [alert]) + assert len(result) == 1 + assert result[0]["baseline"] == 100.0 + assert result[0]["current"] == 130.0 + + +def test_check_alerts_pct_increase_does_not_fire(): + current = _make_record(duration_s=110.0) + history = _history_df( + [ + _make_record(duration_s=100.0), + _make_record(duration_s=100.0), + ] + ) + alert = AlertConfig( + name="duration_regression", + log_name="test_job", + metric="duration_s", + condition="pct_increase", + threshold=0.20, + on_trigger="log", + ) + result = check_alerts(current, history, [alert]) + assert result == [] + + +def test_check_alerts_pct_increase_baseline_zero_fires(): + current = _make_record(spill_bytes=100) + history = _history_df([_make_record(spill_bytes=0)]) + alert = AlertConfig( + name="new_spill", + log_name="test_job", + metric="spill_bytes", + condition="pct_increase", + threshold=0.5, + on_trigger="log", + ) + result = check_alerts(current, history, [alert]) + assert len(result) == 1 + assert result[0]["baseline"] == 0.0 + + +def test_check_alerts_pct_increase_baseline_zero_no_fire(): + current = _make_record(spill_bytes=0) + history = _history_df([_make_record(spill_bytes=0)]) + alert = AlertConfig( + name="no_spill", + log_name="test_job", + metric="spill_bytes", + condition="pct_increase", + threshold=0.5, + on_trigger="log", + ) + result = check_alerts(current, history, [alert]) + assert result == [] + + +def test_check_alerts_absolute_increase_fires(): + current = _make_record(duration_s=200.0) + history = _history_df([_make_record(duration_s=100.0)]) + alert = AlertConfig( + name="abs_increase", + log_name="test_job", + metric="duration_s", + condition="absolute_increase", + threshold=50, + on_trigger="log", + ) + result = check_alerts(current, history, [alert]) + assert len(result) == 1 + assert result[0]["baseline"] == 100.0 + assert result[0]["current"] == 200.0 + + +def test_check_alerts_absolute_increase_does_not_fire(): + current = _make_record(duration_s=120.0) + history = _history_df([_make_record(duration_s=100.0)]) + alert = AlertConfig( + name="abs_increase", + log_name="test_job", + metric="duration_s", + condition="absolute_increase", + threshold=50, + on_trigger="log", + ) + result = check_alerts(current, history, [alert]) + assert result == [] + + +def test_check_alerts_window_larger_than_history(): + current = _make_record(duration_s=200.0) + history = _history_df( + [ + _make_record(duration_s=90.0), + _make_record(duration_s=110.0), + ] + ) + alert = AlertConfig( + name="big_window", + log_name="test_job", + metric="duration_s", + condition="pct_increase", + threshold=0.20, + window=10, + on_trigger="log", + ) + result = check_alerts(current, history, [alert]) + assert len(result) == 1 + assert result[0]["baseline"] == 100.0 + + +def test_check_alerts_excludes_current_run(): + current = _make_record(duration_s=130.0, run_id="current_run") + history = _history_df( + [ + _make_record(duration_s=100.0, run_id="old_1"), + _make_record(duration_s=130.0, run_id="current_run"), + ] + ) + alert = AlertConfig( + name="exclude_current", + log_name="test_job", + metric="duration_s", + condition="pct_increase", + threshold=0.20, + on_trigger="log", + ) + result = check_alerts(current, history, [alert]) + assert len(result) == 1 + assert result[0]["baseline"] == 100.0 + + +def test_check_alerts_filters_by_log_name(): + record = _make_record(log_name="job_b", duration_s=500.0) + history = _history_df([_make_record(log_name="job_a", duration_s=100.0)]) + alert = AlertConfig( + name="job_a_alert", + log_name="job_a", + metric="duration_s", + condition="threshold", + threshold=200, + on_trigger="log", + ) + result = check_alerts(record, history, [alert]) + assert result == [] + + +def test_check_alerts_empty_history_threshold(): + record = _make_record(duration_s=300.0) + alert = AlertConfig( + name="threshold_only", + log_name="test_job", + metric="duration_s", + condition="threshold", + threshold=200, + on_trigger="log", + ) + result = check_alerts(record, pl.DataFrame(), [alert]) + assert len(result) == 1 + + +def test_on_trigger_raise(): + record = _make_record(spill_bytes=200) + alert = AlertConfig( + name="critical_spill", + log_name="test_job", + metric="spill_bytes", + condition="threshold", + threshold=100, + on_trigger="raise", + ) + with pytest.raises(SparkparseAlertError) as exc_info: + check_alerts(record, pl.DataFrame(), [alert]) + assert exc_info.value.alert_name == "critical_spill" + assert exc_info.value.metric == "spill_bytes" + assert exc_info.value.current == 200 + + +def test_on_trigger_log(caplog): + record = _make_record(spill_bytes=200) + alert = AlertConfig( + name="spill_warning", + log_name="test_job", + metric="spill_bytes", + condition="threshold", + threshold=100, + severity="warning", + on_trigger="log", + ) + with caplog.at_level("WARNING", logger="sparkparse.alerts"): + result = check_alerts(record, pl.DataFrame(), [alert]) + assert len(result) == 1 + assert any("spill_warning" in r.message for r in caplog.records) + + +def test_on_trigger_log_critical(caplog): + record = _make_record(spill_bytes=200) + alert = AlertConfig( + name="spill_critical", + log_name="test_job", + metric="spill_bytes", + condition="threshold", + threshold=100, + severity="critical", + on_trigger="log", + ) + with caplog.at_level("ERROR", logger="sparkparse.alerts"): + check_alerts(record, pl.DataFrame(), [alert]) + assert any("spill_critical" in r.message for r in caplog.records) + assert any(r.levelname == "ERROR" for r in caplog.records) + + +def test_on_trigger_file(tmp_path: Path): + alert_path = str(tmp_path / "alerts_out.jsonl") + record = _make_record(spill_bytes=200) + alert = AlertConfig( + name="file_spill", + log_name="test_job", + metric="spill_bytes", + condition="threshold", + threshold=100, + on_trigger="file", + ) + check_alerts(record, pl.DataFrame(), [alert], alert_output_path=alert_path) + + content = Path(alert_path).read_text().strip() + alert_dict = json.loads(content) + assert alert_dict["alert_name"] == "file_spill" + assert alert_dict["current"] == 200 + + +def test_on_trigger_file_without_path(caplog): + record = _make_record(spill_bytes=200) + alert = AlertConfig( + name="orphan_file", + log_name="test_job", + metric="spill_bytes", + condition="threshold", + threshold=100, + on_trigger="file", + ) + with caplog.at_level("ERROR", logger="sparkparse.alerts"): + result = check_alerts(record, pl.DataFrame(), [alert]) + assert len(result) == 1 + assert any("no alert_output_path" in r.message for r in caplog.records) + + +def test_multiple_alerts_some_fire(): + record = _make_record(duration_s=200.0, spill_bytes=50) + history = _history_df([_make_record(duration_s=100.0)]) + alerts = [ + AlertConfig( + name="duration_alert", + log_name="test_job", + metric="duration_s", + condition="pct_increase", + threshold=0.20, + on_trigger="log", + ), + AlertConfig( + name="spill_alert", + log_name="test_job", + metric="spill_bytes", + condition="threshold", + threshold=100, + on_trigger="log", + ), + ] + result = check_alerts(record, history, alerts) + assert len(result) == 1 + assert result[0]["alert_name"] == "duration_alert" diff --git a/tests/test_history.py b/tests/test_history.py new file mode 100644 index 0000000..c53df19 --- /dev/null +++ b/tests/test_history.py @@ -0,0 +1,190 @@ +import datetime +import uuid +from pathlib import Path + +import pytest + +from sparkparse.clean import log_to_combined_df, log_to_dag_df +from sparkparse.history import append, read, record_from_dfs +from sparkparse.models import ParsedLogDataFrames, RunRecord +from sparkparse.parse import parse_log + +DATA_DIR = Path(__file__).parent / "data" / "full_logs" + + +@pytest.fixture(scope="module") +def dfs_nested() -> ParsedLogDataFrames: + log_path = DATA_DIR / "nested_final_plans" + result = parse_log(log_path) + dag = log_to_dag_df(result) + combined = log_to_combined_df(result, dag, log_path.stem) + return ParsedLogDataFrames(dag=dag, combined=combined) + + +@pytest.fixture(scope="module") +def dfs_loop_join() -> ParsedLogDataFrames: + log_path = DATA_DIR / "nested_loop_join" + result = parse_log(log_path) + dag = log_to_dag_df(result) + combined = log_to_combined_df(result, dag, log_path.stem) + return ParsedLogDataFrames(dag=dag, combined=combined) + + +@pytest.fixture(scope="module") +def dfs_complex() -> ParsedLogDataFrames: + log_path = DATA_DIR / "complex_transformation_medium" + result = parse_log(log_path) + dag = log_to_dag_df(result) + combined = log_to_combined_df(result, dag, log_path.stem) + return ParsedLogDataFrames(dag=dag, combined=combined) + + +def _make_record(**overrides) -> RunRecord: + defaults = dict( + run_id=uuid.uuid4().hex, + run_at=datetime.datetime.now(datetime.UTC), + log_name="test_job", + duration_s=100.0, + bytes_read=1000, + bytes_written=500, + shuffle_bytes=200, + spill_bytes=0, + n_queries=2, + n_stages=3, + n_tasks=50, + n_cartesian_joins=0, + max_node_duration_min=1.5, + max_scan_bytes=800, + ) + defaults.update(overrides) + return RunRecord(**defaults) + + +def test_record_from_dfs_valid(dfs_nested): + record = record_from_dfs(dfs_nested, "nested_final_plans") + assert isinstance(record, RunRecord) + assert record.log_name == "nested_final_plans" + assert len(record.run_id) == 32 # uuid4 hex + assert isinstance(record.run_at, datetime.datetime) + assert record.duration_s > 0 + assert record.bytes_read >= 0 + assert record.bytes_written >= 0 + assert record.shuffle_bytes >= 0 + assert record.spill_bytes >= 0 + assert record.n_queries > 0 + assert record.n_stages > 0 + assert record.n_tasks > 0 + assert record.n_cartesian_joins >= 0 + assert record.max_node_duration_min >= 0 + assert record.max_scan_bytes >= 0 + + +def test_record_from_dfs_duration_wall_clock(dfs_nested): + record = record_from_dfs(dfs_nested, "nested_final_plans") + total_query_duration = dfs_nested.dag["query_duration_seconds"].sum() + assert 0 < record.duration_s <= total_query_duration + + +def test_record_from_dfs_cartesian_count(dfs_loop_join): + record = record_from_dfs(dfs_loop_join, "nested_loop_join") + assert record.n_cartesian_joins > 0 + + +def test_record_from_dfs_complex(dfs_complex): + record = record_from_dfs(dfs_complex, "complex_transformation_medium") + assert record.n_queries > 0 + assert record.n_tasks > 0 + + +def test_append_read_roundtrip_jsonl(tmp_path: Path): + history_path = str(tmp_path / "history.jsonl") + records = [ + _make_record(log_name="job_a", duration_s=10.0), + _make_record(log_name="job_a", duration_s=20.0), + _make_record(log_name="job_a", duration_s=30.0), + ] + for r in records: + append(r, history_path, format="jsonl") + + df = read(history_path, format="jsonl") + assert df.height == 3 + assert df["log_name"].unique().to_list() == ["job_a"] + durations = df.sort("run_at")["duration_s"].to_list() + assert durations == sorted(durations) + + +def test_read_filter_by_log_name(tmp_path: Path): + history_path = str(tmp_path / "history.jsonl") + append(_make_record(log_name="job_a"), history_path, format="jsonl") + append(_make_record(log_name="job_b"), history_path, format="jsonl") + append(_make_record(log_name="job_a"), history_path, format="jsonl") + + df = read(history_path, log_name="job_a", format="jsonl") + assert df.height == 2 + assert (df["log_name"] == "job_a").all() + + +def test_read_last_n(tmp_path: Path): + history_path = str(tmp_path / "history.jsonl") + base_time = datetime.datetime(2024, 1, 1, tzinfo=datetime.UTC) + for i in range(5): + append( + _make_record( + log_name="job_a", + duration_s=float(i), + run_at=base_time + datetime.timedelta(hours=i), + run_id=f"run_{i:03d}", + ), + history_path, + format="jsonl", + ) + + df = read(history_path, last_n=2, format="jsonl") + assert df.height == 2 + run_ids = df.sort("run_at")["run_id"].to_list() + assert run_ids == ["run_003", "run_004"] + + +def test_read_empty_path(tmp_path: Path): + history_path = str(tmp_path / "nonexistent.jsonl") + df = read(history_path, format="jsonl") + assert df.is_empty() + + +def test_append_read_roundtrip_delta(tmp_path: Path): + pytest.importorskip("deltalake") + history_path = str(tmp_path / "delta_table") + records = [ + _make_record(log_name="job_a", duration_s=10.0), + _make_record(log_name="job_a", duration_s=20.0), + ] + for r in records: + append(r, history_path, format="delta") + + df = read(history_path, format="delta") + assert df.height == 2 + assert (df["log_name"] == "job_a").all() + + +def test_auto_format_falls_back_to_jsonl(tmp_path: Path, monkeypatch): + monkeypatch.setattr("sparkparse.history._delta_available", lambda: False) + history_path = str(tmp_path / "auto.jsonl") + record = _make_record(log_name="job_a") + append(record, history_path, format="auto") + + assert Path(history_path).exists() + content = Path(history_path).read_text().strip() + assert content.startswith("{") + df = read(history_path, format="auto") + assert df.height == 1 + + +def test_auto_format_uses_delta_when_available(tmp_path: Path): + pytest.importorskip("deltalake") + history_path = str(tmp_path / "auto_delta") + record = _make_record(log_name="job_a") + append(record, history_path, format="auto") + + assert (Path(history_path) / "_delta_log").exists() + df = read(history_path, format="auto") + assert df.height == 1 diff --git a/tests/test_storage.py b/tests/test_storage.py index 11f25b8..a5a3888 100644 --- a/tests/test_storage.py +++ b/tests/test_storage.py @@ -162,7 +162,6 @@ def test_join_path(base: str, parts: tuple[str, ...], expected: str) -> None: assert join_path(base, *parts) == expected - @pytest.fixture def mock_fsspec(): fake = MagicMock(name="fsspec") diff --git a/uv.lock b/uv.lock index 8fd3b1f..a3d53be 100644 --- a/uv.lock +++ b/uv.lock @@ -208,6 +208,70 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/81/29/5ecc3a15d5a33e31b26c11426c45c501e439cb865d0bff96315d86443b78/appnope-0.1.4-py2.py3-none-any.whl", hash = "sha256:502575ee11cd7a28c0205f379b525beefebab9d161b7c964670864014ed7213c", size = 4321, upload-time = "2024-02-06T09:43:09.663Z" }, ] +[[package]] +name = "arro3-core" +version = "0.8.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "typing-extensions", marker = "python_full_version < '3.12'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/c0/c8/fc5bacb6fc264dc61e46d4832f690015b7f6c693ff5dea8a1e53b63cb772/arro3_core-0.8.1.tar.gz", hash = "sha256:1df54a8e2c14a877f291d90de65f00bafe9cc6d5958417ff749f178f059dcd39", size = 93684, upload-time = "2026-06-11T18:01:37.508Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/ca/75/29517738623cccba1d60d0aa65b896bdd046ab4a82c5cd5a9fca115f6d7a/arro3_core-0.8.1-cp311-abi3-macosx_10_12_x86_64.whl", hash = "sha256:64d3ea60da4279e9c6a9b2e60abf7f4fefb6643544ecb2dfde899f72f1f91bad", size = 3085640, upload-time = "2026-06-11T17:59:46.597Z" }, + { url = "https://files.pythonhosted.org/packages/c0/b5/bff1842ed8dbb2134b004f78c52d977b52a6df1d9389b1284aad02bf4d93/arro3_core-0.8.1-cp311-abi3-macosx_11_0_arm64.whl", hash = "sha256:463db0b7f698abc19f4274d6df697560bc9c19463cfdbd01094bb0fa6ddd1206", size = 2800705, upload-time = "2026-06-11T17:59:48.242Z" }, + { url = "https://files.pythonhosted.org/packages/32/20/ef60404d5008f84bca50510f65d7acfe26812e61638ee19a416bf3f11530/arro3_core-0.8.1-cp311-abi3-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:881afcb9c83334ac26498b429265223c676e028cd5d88fb0c909578f8e0330de", size = 3272103, upload-time = "2026-06-11T17:59:49.641Z" }, + { url = "https://files.pythonhosted.org/packages/f5/8e/6687efc8e0414cfde6281b1f4ea3b5b44aeb318584e39113d645d1f913a2/arro3_core-0.8.1-cp311-abi3-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:d1dc03e1e5965528bfc7e26a6ee857e3b6fc5f201ff530a03dc731b9f3fb14fd", size = 3405750, upload-time = "2026-06-11T17:59:51.037Z" }, + { url = "https://files.pythonhosted.org/packages/2e/38/2567f26cd041387a2c1f381f1757b5246184de22dd8bd9dbbcb91a1b3586/arro3_core-0.8.1-cp311-abi3-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:410cdca92392be39a5580059ebf9a6f89a0410af6dc10a8baa76e9d511fa6624", size = 3467290, upload-time = "2026-06-11T17:59:52.672Z" }, + { url = "https://files.pythonhosted.org/packages/85/38/2c5af3a3e8c806188f1b3a651cbaa6c22d1f094c41e82900d40219c2b337/arro3_core-0.8.1-cp311-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:f4082c6bbff7f164619d99dffcbc2313ed69024653a9e58d9f94cc65d9226f6c", size = 3195582, upload-time = "2026-06-11T17:59:54.536Z" }, + { url = "https://files.pythonhosted.org/packages/e4/8d/d7d8686901a273743c53aae98f898f00caaedea807e28854200f2a7365e4/arro3_core-0.8.1-cp311-abi3-manylinux_2_24_aarch64.whl", hash = "sha256:40da2ee0701f217cd482e61228023ee9d8993984daccaf6ded089035b5fdc132", size = 2950909, upload-time = "2026-06-11T17:59:56.05Z" }, + { url = "https://files.pythonhosted.org/packages/5b/8d/f44198a859b13f519b331906d08079c7c281f8267d718082a3ca858c93ee/arro3_core-0.8.1-cp311-abi3-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:d79b12a83403b1a100731587d33f09a818a4ff93472f84fcdf3a38d3934dfc5e", size = 3403497, upload-time = "2026-06-11T17:59:57.651Z" }, + { url = "https://files.pythonhosted.org/packages/88/6d/431d2ef9942a30599db4a86cf13b7f1de92d340717438074a25659d382f4/arro3_core-0.8.1-cp311-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:3ff2da3c888ec504291deafd6965f2364f14f1fb63977f202f2f7680f10909f6", size = 3130901, upload-time = "2026-06-11T17:59:59.262Z" }, + { url = "https://files.pythonhosted.org/packages/ce/9e/458e8596c675fb88075137ab21d4c4a2f9e585101a97d77e910839862168/arro3_core-0.8.1-cp311-abi3-musllinux_1_2_armv7l.whl", hash = "sha256:0bbcfb60ddf9b571387b4c0aaf78171d2ee7e0d335c08514bc247685f01bc086", size = 3550071, upload-time = "2026-06-11T18:00:00.808Z" }, + { url = "https://files.pythonhosted.org/packages/93/cd/8fa1f4a49e707ea364022620ca5033ba69b6396215e0bae1fb9fa9efcfe0/arro3_core-0.8.1-cp311-abi3-musllinux_1_2_i686.whl", hash = "sha256:44f6256cae47d369fe9076e9fa61f5c2899447b76ca5bda641668ad07cf26bf6", size = 3508957, upload-time = "2026-06-11T18:00:02.341Z" }, + { url = "https://files.pythonhosted.org/packages/7f/02/0126ff3b2ff48187315a9782280c0f6412c066559e22bf206ceec2791299/arro3_core-0.8.1-cp311-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:b2517ea3fb7cb15a41e2d0e3496d91afe4703eca86a88072f917bef044c2b8ee", size = 3408923, upload-time = "2026-06-11T18:00:03.97Z" }, + { url = "https://files.pythonhosted.org/packages/d8/fd/a19cb50480a769b1cc3f347efa1a808e13b853bb3b1d94b289677115fe92/arro3_core-0.8.1-cp311-abi3-win_amd64.whl", hash = "sha256:6a96df94a4538ab9acf01873eecf35161ad0c54ab79134247e69a42cd66515dc", size = 3368045, upload-time = "2026-06-11T18:00:05.578Z" }, + { url = "https://files.pythonhosted.org/packages/5c/86/1ac6eed229482b1ef6ac6f3211c6760012aaaa026b0a385a18e42ec0ff74/arro3_core-0.8.1-cp313-cp313-pyemscripten_2025_0_wasm32.whl", hash = "sha256:71165a75a7d22c16226085ed2255afe84741369cd35eb01cc153c4687e85b49f", size = 1724964, upload-time = "2026-06-11T18:00:07.653Z" }, + { url = "https://files.pythonhosted.org/packages/4a/32/c2e4a65b3b4f7e00552d280e9c2eaaca59c19466b0f95e4e08d6a020ab2d/arro3_core-0.8.1-cp313-cp313t-macosx_10_12_x86_64.whl", hash = "sha256:dc81e079d182dfc2ba6bace815608dbab0158809b9268c4eb59816685f7871fb", size = 3071070, upload-time = "2026-06-11T18:00:09.12Z" }, + { url = "https://files.pythonhosted.org/packages/f0/32/56016de27757bcc0d66999ae2fec624eaa27bd097ee69ac7a5a0d52d407b/arro3_core-0.8.1-cp313-cp313t-macosx_11_0_arm64.whl", hash = "sha256:7695e9fca1e2c0064571bea65d9b7909a957ed9be4c81718d19356e578818649", size = 2794497, upload-time = "2026-06-11T18:00:10.872Z" }, + { url = "https://files.pythonhosted.org/packages/d2/6e/a7aac0e87d6d63672d4ebb7e8466b276fde2a6413579008fadd1e8e919c5/arro3_core-0.8.1-cp313-cp313t-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:ad0f507dfafb1e8a8c15e8a902bccdd9a9c2cf2102fc39ad35c22e23ca76a65e", size = 3272090, upload-time = "2026-06-11T18:00:12.561Z" }, + { url = "https://files.pythonhosted.org/packages/1b/fd/8b96e57eea8ebf5fa4e2556b06c84a29ea25ded924f56d300807ff377a94/arro3_core-0.8.1-cp313-cp313t-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:3fc91a59cb27c7660134b9ed0c067d53c0f5bb1722f16f2cf7d4c446ce1086c4", size = 3399753, upload-time = "2026-06-11T18:00:13.992Z" }, + { url = "https://files.pythonhosted.org/packages/80/8b/76f1385ecd0448fcc60902a4474f3f00bd2197fe6d7fa4434bb6a0b3dee9/arro3_core-0.8.1-cp313-cp313t-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:71bc86880256cca5f22802ca444a92b3e9edbb5c9e5b74df1cbfcfe2ca788097", size = 3468912, upload-time = "2026-06-11T18:00:15.502Z" }, + { url = "https://files.pythonhosted.org/packages/0f/15/e1c3aefee23d926be9f75f125d6d94f189f28826aa01713498f1de60b080/arro3_core-0.8.1-cp313-cp313t-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:94528515f59146b5b4421e468bc269f682754ff082434cccf0ecbfb0d74d0b38", size = 3193338, upload-time = "2026-06-11T18:00:17.064Z" }, + { url = "https://files.pythonhosted.org/packages/de/3d/bfa3963bd09941f69d810a387bbea61deb8a091101c1765216266ec61394/arro3_core-0.8.1-cp313-cp313t-manylinux_2_24_aarch64.whl", hash = "sha256:e348fa5349e6655df4b1ea281ab71e0794c749ba842da54248eb942201f93fb8", size = 2950166, upload-time = "2026-06-11T18:00:18.607Z" }, + { url = "https://files.pythonhosted.org/packages/21/07/67bdc66582294d5914b8f7462907b1e44377ccb77a0f7e7b7760c9f81fd1/arro3_core-0.8.1-cp313-cp313t-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:d97170d56690655699977f4535120f170c5047813341c2b0264abc3af00625dc", size = 3401097, upload-time = "2026-06-11T18:00:20.135Z" }, + { url = "https://files.pythonhosted.org/packages/6d/a3/ab0d46e8fe2337f0ce1789949c5078e63d532fd628af4f254702e1359989/arro3_core-0.8.1-cp313-cp313t-musllinux_1_2_aarch64.whl", hash = "sha256:6f71fdd46e1a86bc1db0733361130e51447d2d3c7d5fca13937381e2ba085dca", size = 3128454, upload-time = "2026-06-11T18:00:21.598Z" }, + { url = "https://files.pythonhosted.org/packages/c2/6e/e90cea2555aaa429c16173cbc121b41fb6a656e70ae9f21edbe3f9e73eb0/arro3_core-0.8.1-cp313-cp313t-musllinux_1_2_armv7l.whl", hash = "sha256:3e2c2e4c6b36a0f493f07222eadf7c628155f4a306c650a14c4d6d343870043c", size = 3550294, upload-time = "2026-06-11T18:00:24.344Z" }, + { url = "https://files.pythonhosted.org/packages/15/38/f3db72e411fbaea83844d19b1965abbdf3dd387083f99a715d3b3752b461/arro3_core-0.8.1-cp313-cp313t-musllinux_1_2_i686.whl", hash = "sha256:90d31db38bc7cbbe6061d23ae11b3e830b55ce60ba6f0550c70915bab4cafe41", size = 3507806, upload-time = "2026-06-11T18:00:26.143Z" }, + { url = "https://files.pythonhosted.org/packages/47/73/5d011dad78ac58aaa61e9cc65d349b879afec6781c5aa78f80c839fe359f/arro3_core-0.8.1-cp313-cp313t-musllinux_1_2_x86_64.whl", hash = "sha256:9d28d87625d6c743fee1d2e20252531db6cff6c5218b89055a9f7492af99c14c", size = 3408910, upload-time = "2026-06-11T18:00:27.843Z" }, + { url = "https://files.pythonhosted.org/packages/6a/43/be86ea01c2d53d32376602cb2efa9a9b7a25d98e14ad4b5bf0d7da8b0564/arro3_core-0.8.1-cp313-cp313t-win_amd64.whl", hash = "sha256:b394218ad71ce266088e0e65bfad0542556adf0eb93197b66afaa2009358662f", size = 3356174, upload-time = "2026-06-11T18:00:29.395Z" }, + { url = "https://files.pythonhosted.org/packages/7a/0f/2f7b5b458fd38ecf9277fd59919e4c90047795d1abdfd47f0081c48c28fa/arro3_core-0.8.1-cp314-cp314-pyemscripten_2026_0_wasm32.whl", hash = "sha256:a7c227982c0fd762271d550a6242e54ff7d2aa7aff73c918e127042cf9601ec0", size = 1710676, upload-time = "2026-06-11T18:00:30.947Z" }, + { url = "https://files.pythonhosted.org/packages/4d/9e/b64c9325173b7d87030955007af55348b866a1c02545382261d8966d8bb1/arro3_core-0.8.1-cp314-cp314t-macosx_10_12_x86_64.whl", hash = "sha256:5861781e6363db0707e1a4ba5166025a1113d921ccff870a0a097f27ab2dece1", size = 3072327, upload-time = "2026-06-11T18:00:32.313Z" }, + { url = "https://files.pythonhosted.org/packages/3c/bf/2c58a549b2409439fcac4ce2abe31ef060d05114a1e5412ce3088e2a2cde/arro3_core-0.8.1-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:8c1d9ff8b2848bf48bbcad962271b2adeda922457cf23591464c12af9b0ddbf6", size = 2794872, upload-time = "2026-06-11T18:00:33.92Z" }, + { url = "https://files.pythonhosted.org/packages/1f/21/dee8d1c9309820783fe30ad29149508638743cdc61d2851f0132b370ac49/arro3_core-0.8.1-cp314-cp314t-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:246057dc9d283a283cb05b228f710e1ada3b4a8c483e519b0ec8b0d8499204db", size = 3272432, upload-time = "2026-06-11T18:00:35.563Z" }, + { url = "https://files.pythonhosted.org/packages/b6/b2/dbc6a3d5ece2cb11c3041821c8322df500b8c8030534d9f791b66b29c090/arro3_core-0.8.1-cp314-cp314t-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:5d55178ad56c637f278286370532e58256eff48f71d3c36a4e5f8652eb7300b3", size = 3400822, upload-time = "2026-06-11T18:00:37.129Z" }, + { url = "https://files.pythonhosted.org/packages/a4/44/0ca0e96eebc7f96af4ad8b78c545a9477e8db271aa090043bb8f797e7f58/arro3_core-0.8.1-cp314-cp314t-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:7a917195d7553a188bbb9fa815d430ff70ee53357fcc6fdb395d0e4436a66243", size = 3469220, upload-time = "2026-06-11T18:00:39.375Z" }, + { url = "https://files.pythonhosted.org/packages/8e/70/3e9c63e9e4499d373304ceb1315afc8dd326395dd0692ccdebcfec703dab/arro3_core-0.8.1-cp314-cp314t-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:2435dc4057d604650a34d195e6977ef40b26af0b9dea8f1b842a924a270de2fc", size = 3193813, upload-time = "2026-06-11T18:00:41.281Z" }, + { url = "https://files.pythonhosted.org/packages/a2/a8/48842a836d7fc2bdac16ecaf4e60c9fd98f46d1e9c80f45b85b071ccaffb/arro3_core-0.8.1-cp314-cp314t-manylinux_2_24_aarch64.whl", hash = "sha256:c8b90ad6fbdd3507c20bdbc2b23c88929a3f7c49a8a43697f316e084b5664289", size = 2950818, upload-time = "2026-06-11T18:00:42.987Z" }, + { url = "https://files.pythonhosted.org/packages/bb/46/635ca59c20b972f0575b9fd1b7debebd94bcba1d6fd68271bfff65d3ae91/arro3_core-0.8.1-cp314-cp314t-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:8918dd0fcf5c3650332678ddbfb7c13ce1f0534f9669ab6839c38069885318cf", size = 3401202, upload-time = "2026-06-11T18:00:44.654Z" }, + { url = "https://files.pythonhosted.org/packages/4a/03/020d032cf4937cca0652f434613d327f8f11e10b4a3fbd4ccf48846c3158/arro3_core-0.8.1-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:6cc8c7c9b81cad2b4eaaeb29da25854bddb5bb373a278f1f1046b277fa7f2ee6", size = 3129352, upload-time = "2026-06-11T18:00:46.322Z" }, + { url = "https://files.pythonhosted.org/packages/b2/55/ce1c840af64e8c5d4f8b684a4c00c5f6d260659c3d2f732c6074eb5597fe/arro3_core-0.8.1-cp314-cp314t-musllinux_1_2_armv7l.whl", hash = "sha256:2aa0bda5251e93c67318aa8315e563b02fe39d5577b72654c335d69fce802d03", size = 3550475, upload-time = "2026-06-11T18:00:48.021Z" }, + { url = "https://files.pythonhosted.org/packages/19/79/fa8f1a53cf10fe33029619d14f06525e08ea35c1bcdce42157b230098e1b/arro3_core-0.8.1-cp314-cp314t-musllinux_1_2_i686.whl", hash = "sha256:cd462e7ef7735c341c27f754756b244d90a9485818c5ea090fa46009bd8ae252", size = 3507877, upload-time = "2026-06-11T18:00:49.798Z" }, + { url = "https://files.pythonhosted.org/packages/a3/e5/bf8349b2e5613e6c0caf2f4f0758d61dc273649c2bdcfc96978ab9bc1686/arro3_core-0.8.1-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:480c33d62d66adf92fd731e77728f81b7bd3fcde4e2eb313f5616b5097b1875f", size = 3409360, upload-time = "2026-06-11T18:00:51.393Z" }, + { url = "https://files.pythonhosted.org/packages/a4/0d/e13efbbc448bc817f8343bf12606d22f83011682c821ccd32dda304816be/arro3_core-0.8.1-cp314-cp314t-win_amd64.whl", hash = "sha256:51bea8dc0bc0230b6af8d1b258cf2a7e4f69770ed99062592ce1b43b17732727", size = 3355557, upload-time = "2026-06-11T18:00:52.922Z" }, + { url = "https://files.pythonhosted.org/packages/de/bc/ab3081cc65cd38a5ceda6dd112ed9d70a481e2810c8f92d37441edab6a8a/arro3_core-0.8.1-pp311-pypy311_pp73-macosx_10_12_x86_64.whl", hash = "sha256:d70c68665525744154dc1f65b5038fe97600a5204e8a875b3fe7675455363024", size = 3076025, upload-time = "2026-06-11T18:01:16.781Z" }, + { url = "https://files.pythonhosted.org/packages/0b/5b/b278236c7425f36fe1d25087964ca143d2a11f99b69b19f3c050cf3a0f4b/arro3_core-0.8.1-pp311-pypy311_pp73-macosx_11_0_arm64.whl", hash = "sha256:b89c867f65852de7fa0586a4883195a2cca1ee3a7bca8c95c309a02ce6be15b5", size = 2797246, upload-time = "2026-06-11T18:01:18.648Z" }, + { url = "https://files.pythonhosted.org/packages/1c/a5/a3a96bef25a7d0d7fd49ea3762d901a17d908fda774c868abe9aa065e0f2/arro3_core-0.8.1-pp311-pypy311_pp73-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:66d685f7c53137798e05322a17dabc09445a818c9c8959d21e9fb3e084371f51", size = 3269959, upload-time = "2026-06-11T18:01:20.317Z" }, + { url = "https://files.pythonhosted.org/packages/f7/2f/4da148dea5890ae24eade70d3f9dfc93d191f501bf425d906c21dee89510/arro3_core-0.8.1-pp311-pypy311_pp73-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:5a7563856e609dd24be03efde72edd5b1c16dd1fb3d67fd84970df6d9ca36b96", size = 3400877, upload-time = "2026-06-11T18:01:22.112Z" }, + { url = "https://files.pythonhosted.org/packages/7b/9a/2ed015d4a03c131b9e3da322ee8681c5921320a90218865a711d7fe39768/arro3_core-0.8.1-pp311-pypy311_pp73-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:b7379e14f8f49e99d53e364c26533293d2136920a3cf818138718a9a9bc114d0", size = 3466119, upload-time = "2026-06-11T18:01:23.909Z" }, + { url = "https://files.pythonhosted.org/packages/25/a8/154478c9b3093602eaec8a5d71b657205177f0ac9fd9da2eef9ddfb42fe0/arro3_core-0.8.1-pp311-pypy311_pp73-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:af88a17346e9e7897b6d1bc165f686b87ea928b4c7e89666156b2281f0236cee", size = 3190211, upload-time = "2026-06-11T18:01:25.786Z" }, + { url = "https://files.pythonhosted.org/packages/5b/65/60c6323bbaffd2399f03336a90d7c02450a5b9120348808eb0260651de56/arro3_core-0.8.1-pp311-pypy311_pp73-manylinux_2_24_aarch64.whl", hash = "sha256:1e9e04cbbba877f31073146798b328537b63675f65d0b7e2154754d5203260ed", size = 2950345, upload-time = "2026-06-11T18:01:27.423Z" }, + { url = "https://files.pythonhosted.org/packages/21/81/3a0c786ec86a8c3aee946e4588b188f08a3ccbba71c4e0db144a0151b56c/arro3_core-0.8.1-pp311-pypy311_pp73-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:35c46c49442853f8a1b09d37ceabe4ce31e2a276d4a73c407ba3a25fcf1b9221", size = 3398153, upload-time = "2026-06-11T18:01:29.036Z" }, + { url = "https://files.pythonhosted.org/packages/f6/52/e230d0af9881056e527f91e58c518b85000e9ae49f8706c3d0e027cbadd4/arro3_core-0.8.1-pp311-pypy311_pp73-musllinux_1_2_aarch64.whl", hash = "sha256:4e4af152b4ccf7694d634573db906e1a9b28f5beda84bac5adeb60f7e98262a8", size = 3128123, upload-time = "2026-06-11T18:01:30.733Z" }, + { url = "https://files.pythonhosted.org/packages/62/79/9bc36279f1ef2754fde18e1c3c282b2cabe6093b7ca8bd8c0c6c735685ed/arro3_core-0.8.1-pp311-pypy311_pp73-musllinux_1_2_armv7l.whl", hash = "sha256:7bf8cb1053cb51529437544eedbc6643819912eec15c0491d7304d6bad0fb151", size = 3547345, upload-time = "2026-06-11T18:01:32.603Z" }, + { url = "https://files.pythonhosted.org/packages/f9/33/cee8670131009d3a23e140dd0a6cd454111dac54a9a2e83567a6f972ad46/arro3_core-0.8.1-pp311-pypy311_pp73-musllinux_1_2_i686.whl", hash = "sha256:212dfde17bd26193754aaee6024f7e99624163e0d44966745e8752af79c4c454", size = 3504672, upload-time = "2026-06-11T18:01:34.207Z" }, + { url = "https://files.pythonhosted.org/packages/3f/5f/daa7a8af9ad53eebbdb818078d5034525663f298ec9bfdb1933cfd0cc694/arro3_core-0.8.1-pp311-pypy311_pp73-musllinux_1_2_x86_64.whl", hash = "sha256:81426cb122d6bfccacd2e0ae5df8b847dc3c3e07b5f336547cb18f873d186f54", size = 3404906, upload-time = "2026-06-11T18:01:36.097Z" }, +] + [[package]] name = "asttokens" version = "3.0.0" @@ -632,6 +696,36 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/4e/8c/f3147f5c4b73e7550fe5f9352eaa956ae838d5c51eb58e7a25b9f3e2643b/decorator-5.2.1-py3-none-any.whl", hash = "sha256:d316bb415a2d9e2d2b3abcc4084c6502fc09240e292cd76a76afc106a1c8e04a", size = 9190, upload-time = "2025-02-24T04:41:32.565Z" }, ] +[[package]] +name = "deltalake" +version = "1.6.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "arro3-core" }, + { name = "deprecated" }, +] +wheels = [ + { url = "https://files.pythonhosted.org/packages/fe/e1/b26c473480347ba82ffe4abde5452d4aa36313cc267e08d9a8ae9c7083fd/deltalake-1.6.0-cp310-abi3-macosx_10_12_x86_64.whl", hash = "sha256:e980c834c3657d0476a0e98763a7f6c6934717175d3ef86208dd974eceb35b15", size = 40940249, upload-time = "2026-05-18T14:25:43.718Z" }, + { url = "https://files.pythonhosted.org/packages/e4/22/ae4148a1d4a0b3cfaea1eed627610a96b79f3d25443237a26af1acb43a67/deltalake-1.6.0-cp310-abi3-macosx_11_0_arm64.whl", hash = "sha256:d15cdc95a816dc1744fb208da54bb8dd57548ac4635c1b955727d57123a2e599", size = 37520755, upload-time = "2026-05-18T14:30:38.908Z" }, + { url = "https://files.pythonhosted.org/packages/0f/5d/a9fa27de555540fc11e83bd9e3de5c356f3e9e1019694f9ce32340b9e09f/deltalake-1.6.0-cp310-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:cd22cf2301c8a3d06819813726ba3ecf68ac16ba12a7c4ac4b1c9b7177461da7", size = 41804855, upload-time = "2026-05-18T13:56:40.982Z" }, + { url = "https://files.pythonhosted.org/packages/f1/bc/e09d13df887783018580d2383765cbd3aaf10c45ab428dd38e9c91053b75/deltalake-1.6.0-cp310-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:6780a3702f501c902fa7deecd7501a4f0e52d69321835bf6f87d72b13679bfd4", size = 40545361, upload-time = "2026-05-18T17:52:57.876Z" }, + { url = "https://files.pythonhosted.org/packages/4f/08/30720d394ecc394465c74ca2342cc51f9fe6f1867985b3df5cec3b6e8c63/deltalake-1.6.0-cp310-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:a2da86d4964cfd1db988158248cdf46acaea2d5ac7cc0eccee04bff565265153", size = 40538718, upload-time = "2026-05-18T13:46:27.813Z" }, + { url = "https://files.pythonhosted.org/packages/68/d0/9e0c87894641ce9b921e4d3f576ccedce68907cc55ba5004fc4291260289/deltalake-1.6.0-cp310-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:5fc975a93a043619175b86fbdf0951bf15ec4ac6e499d981dfc8c9024748d831", size = 41806788, upload-time = "2026-05-18T13:57:09.396Z" }, + { url = "https://files.pythonhosted.org/packages/0a/5b/a1e6d8d35e5b2f8460e45e9171af50c461283c7390c3977f4b7f623193b1/deltalake-1.6.0-cp310-abi3-win_amd64.whl", hash = "sha256:cdc15e2ad80376363ad4d13f475b00c23e4192b5e65e8e421b8aa7cf9df533fa", size = 44143517, upload-time = "2026-05-18T14:23:12.694Z" }, +] + +[[package]] +name = "deprecated" +version = "1.3.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "wrapt" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/49/85/12f0a49a7c4ffb70572b6c2ef13c90c88fd190debda93b23f026b25f9634/deprecated-1.3.1.tar.gz", hash = "sha256:b1b50e0ff0c1fddaa5708a2c6b0a6588bb09b892825ab2b214ac9ea9d92a5223", size = 2932523, upload-time = "2025-10-30T08:19:02.757Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/84/d0/205d54408c08b13550c733c4b85429e7ead111c7f0014309637425520a9a/deprecated-1.3.1-py2.py3-none-any.whl", hash = "sha256:597bfef186b6f60181535a29fbe44865ce137a5079f295b479886c82729d5f3f", size = 11298, upload-time = "2025-10-30T08:19:00.758Z" }, +] + [[package]] name = "executing" version = "2.2.0" @@ -2290,6 +2384,9 @@ cloud = [ { name = "gcsfs" }, { name = "s3fs" }, ] +delta = [ + { name = "deltalake" }, +] gcs = [ { name = "gcsfs" }, ] @@ -2312,6 +2409,7 @@ requires-dist = [ { name = "dash-ag-grid", specifier = ">=31.3.0" }, { name = "dash-bootstrap-components", specifier = ">=1.6.0,<2" }, { name = "dash-cytoscape", specifier = ">=1.0.2" }, + { name = "deltalake", marker = "extra == 'delta'", specifier = ">=0.25" }, { name = "falsa", specifier = ">=0.0.3" }, { name = "fsspec", specifier = ">=2024.1.0" }, { name = "gcsfs", marker = "extra == 'cloud'", specifier = ">=2024.1.0" }, @@ -2324,7 +2422,7 @@ requires-dist = [ { name = "s3fs", marker = "extra == 'cloud'", specifier = ">=2024.1.0" }, { name = "s3fs", marker = "extra == 's3'", specifier = ">=2024.1.0" }, ] -provides-extras = ["s3", "azure", "gcs", "cloud"] +provides-extras = ["s3", "azure", "gcs", "cloud", "delta"] [package.metadata.requires-dev] dev = [