diff --git a/CHANGELOG.md b/CHANGELOG.md index fa5564d3..1e0ba39c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,12 @@ All notable changes to Lakebench are documented here. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). +## [1.7.12] - 2026-10-23 + +### Added + +- **`platform.compute.spark.executor_scratch` and the per-stage `bronze_executor_scratch`, `silver_executor_scratch` and `gold_executor_scratch` set each executor's scratch PVC (whole gibibytes, for example `200Gi`).** Every Spark job's scratch size was fixed by its built-in profile, with no key to raise it, so a job that outgrew its PVC could only fail: the AML scorer's 50Gi filled at scale 20 ("No space left on device" in the coverage aggregation's spill) and the run went unscored. A stage key sizes that stage's job in the run's mode; `executor_scratch` sizes every other job, including the scorer and time-travel reads. A set size is exact (not scaled by cores, scale or executor count), is counted by the capacity check and the scratch preflight, and is recorded in `config_snapshot.scratch.size_per_job` and the perf fingerprint, so runs with different scratch sizes do not compare as the same configuration. Setting one turns `platform.storage.scratch.enabled` on when it is unset; with it set to `false` the config is refused, since no executor would get a PVC. A config that sets none of these keys requests the same PVCs as before. + ## [1.7.11] - 2026-10-23 ### Breaking diff --git a/docs/component-spark.md b/docs/component-spark.md index 723259dc..1e3048dc 100644 --- a/docs/component-spark.md +++ b/docs/component-spark.md @@ -84,13 +84,15 @@ Spark settings live under `platform.compute.spark`, `platform.storage.scratch`, | `bronze_executors`, `silver_executors`, `gold_executors` | `null` | Executor count of the stage's job in the run's mode (bronze-verify or bronze-ingest, silver-build or silver-stream, gold-finalize or gold-refresh). Set, it bypasses the scaling formula and, in continuous mode, the shared-cluster Lakebench cap. | | `bronze_executor_cores`, `silver_executor_cores`, `gold_executor_cores` | `null` | Continuous mode only: executor size, 1-16 cores. Memory and scratch per core follow the profile; the count is unchanged, so total cores grow. Batch jobs ignore it. | | `bronze_executor_memory`, `silver_executor_memory`, `gold_executor_memory` | `null` | Continuous mode only: heap per executor (e.g. `96g`), applied after the cores. The overhead keeps the profile's ratio to the heap; the capacity check and the node fit count the larger pod. Batch jobs ignore it. | -| `platform.storage.scratch.enabled` | `false` | Portworx scratch PVCs for executors. Unset, a batch run at scale 50 and above turns it on. | +| `executor_scratch` | `null` | Scratch PVC per executor for every Spark job, in whole gibibytes (e.g. `200Gi`). Replaces the profile's size outright, in either mode. | +| `bronze_executor_scratch`, `silver_executor_scratch`, `gold_executor_scratch` | `null` | Scratch PVC per executor for the stage's job in the run's mode; wins over `executor_scratch` for that stage. | +| `platform.storage.scratch.enabled` | `false` | Portworx scratch PVCs for executors. Unset, a batch run at scale 50 and above, or a config that sets a scratch size, turns it on. Set to `false` with a scratch size, the config is refused. | | `platform.storage.scratch.storage_class` | `px-csi-scratch` | Scratch StorageClass. Must be `repl=1`. | | `spark.conf` | `{}` | Spark conf merged over the job defaults (below). | - Executor count overrides above 28 are refused by the commands that change data; `destroy` and read commands drop them with a note. - Unset `_executor_cores`: bronze-ingest and silver-stream grow to 8 or 16 cores when the offered load needs more executors than the cap (keeping total cores), unless the stage's `_executors` count is set. gold-refresh is never grown. -- Batch jobs take only a count override. The removed `platform.compute.spark.driver` and `.executor` blocks and `scratch.size` are listed under 1.7.0 in the [changelog](../CHANGELOG.md#170---2026-10-06). +- Batch jobs take a count and a scratch size override. The removed `platform.compute.spark.driver` and `.executor` blocks and `scratch.size` are listed under 1.7.0 in the [changelog](../CHANGELOG.md#170---2026-10-06). ### Spark conf (S3A, shuffle, memory) @@ -196,7 +198,7 @@ When enabled, each executor gets a dynamically provisioned PVC at `/tmp/spark-lo - **Batch, per-scale profile:** the stage's `scratch_gib_per_scale` x scale / executors, rounded up, between 50Gi and the profile's `scratch_size` (the ceiling). Examples: Customer 360 silver-build gets 8 x 50Gi at scale 1, 8 x 75Gi at scale 10 and 18 x 300Gi at scale 100; AML bronze-verify with 16 executors at scale 10 gets 50Gi each. - **No per-scale need** (c360 bronze-verify, 50Gi, and the streaming jobs): `scratch_size`. A widened streaming executor's PVC grows with its cores. -- To change scratch, change the executor count (`platform.compute.spark._executors`). There is no size field. +- **Set in the config:** `platform.compute.spark._executor_scratch`, else `executor_scratch`, replaces the size above for every executor of the job. It is exact: not scaled by cores, scale or executor count. `executor_scratch` is the only control for jobs outside the three stages, such as the AML scorer. - `metrics.json` records the size per job as `config_snapshot.scratch.size_per_job`. - The StorageClass must use `repl=1`; replication only doubles storage for recomputable shuffle data. - `lakebench deploy` only checks that the StorageClass exists. A cluster admin creates it once with `lakebench admin install --component scratch-storage-class`. diff --git a/docs/configuration.md b/docs/configuration.md index a8d0b8aa..a19b00f5 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -87,7 +87,7 @@ The removed keys, with what to do instead (generated from each model's | `platform.compute.spark.silver_stream_executors` | use silver_executors, which sizes silver-stream in continuous mode and silver-build in batch mode. | | `platform.storage.s3.secret_ref` | lakebench never reads an existing Secret: deploy writes the S3 Secret from access_key and secret_key. Set those instead (use ${VAR} substitution, for example access_key: "${S3_ACCESS_KEY}", to keep the keys out of the file). | | `platform.storage.scratch.create_storage_class` | lakebench creates no StorageClass; a cluster admin runs 'lakebench admin install --component scratch-storage-class' once. | -| `platform.storage.scratch.size` | per-job scratch comes from the job profiles (silver-build: 60 GiB x scale / executors, 50-300Gi). | +| `platform.storage.scratch.size` | each executor's scratch PVC is platform.compute.spark.executor_scratch (or a stage's *_executor_scratch), else its job profile's size. | | `secret_ref` | lakebench never reads an existing Secret: deploy writes the S3 Secret from access_key and secret_key. Set those instead. | | `version` | there is one config schema and nothing read the number; delete the key. | | `workload.customer360.channels` | the customer360 generator never read it; delete the key. | @@ -287,7 +287,7 @@ platform: scratch: # enabled: true # Portworx scratch PVCs; unset, it turns on for a batch run at scale 50 and above - storage_class: px-csi-scratch # PVC size per executor: the job profile's + storage_class: px-csi-scratch # PVC size per executor: compute.spark.*executor_scratch, else the job profile's compute: spark: @@ -301,8 +301,7 @@ platform: # Per-stage executor counts (null = auto). One key per stage; the run's # mode picks the job: bronze-verify or bronze-ingest, silver-build or - # silver-stream, gold-finalize or gold-refresh. Per-executor memory and - # PVC are fixed from proven profiles. + # silver-stream, gold-finalize or gold-refresh. bronze_executors: null silver_executors: null gold_executors: null @@ -313,6 +312,12 @@ platform: silver_executor_cores: null gold_executor_cores: null + # Scratch PVC per executor (null = the job profile's size). Exact, in + # either mode; a stage key wins for its stage. Setting one turns + # scratch on when platform.storage.scratch.enabled is unset. + executor_scratch: null # e.g. 200Gi + silver_executor_scratch: null + # Global driver overrides (null = profile default). driver_memory: null driver_cores: null @@ -588,6 +593,10 @@ Scratch PVCs for Spark shuffle data. Only needed with Portworx or similar CSI. | `platform.compute.spark.bronze_executor_memory` | string or null | `null` | advanced | Continuous mode: override bronze-ingest heap per executor (a Spark size, for example `32g`); batch mode ignores it. The memory overhead keeps the profile's ratio to the heap, and the capacity check and the node fit count the larger pod. Null = the profile's per-core share of the executor size. | | `platform.compute.spark.silver_executor_memory` | string or null | `null` | advanced | Continuous mode: override silver-stream heap per executor (a Spark size, for example `96g`); batch mode ignores it. The memory overhead keeps the profile's ratio to the heap, and the capacity check and the node fit count the larger pod. For a financial run that leaves `silver_max_rows_per_trigger` unset, a larger heap raises the derived cap up to its measured bound. Null = the profile's per-core share of the executor size. | | `platform.compute.spark.gold_executor_memory` | string or null | `null` | advanced | Continuous mode: override gold-refresh heap per executor (a Spark size, for example `96g`); batch mode ignores it. The memory overhead keeps the profile's ratio to the heap, and the capacity check and the node fit count the larger pod. Null = the profile's per-core share of the executor size. | +| `platform.compute.spark.executor_scratch` | string or null | `null` | advanced | Scratch PVC per executor for every Spark job (whole gibibytes, for example `200Gi`). A stage's `*_executor_scratch` wins for that stage's jobs. Setting any scratch size turns `platform.storage.scratch.enabled` on when it is unset; with it set to `false` the config is refused. Null = each job profile's size. | +| `platform.compute.spark.bronze_executor_scratch` | string or null | `null` | advanced | Scratch PVC per executor for the bronze stage (for example `200Gi`): bronze-verify in batch mode, bronze-ingest in continuous mode. Null = `executor_scratch`, else the profile's size. | +| `platform.compute.spark.silver_executor_scratch` | string or null | `null` | advanced | Scratch PVC per executor for the silver stage (for example `300Gi`): silver-build in batch mode, silver-stream in continuous mode. Null = `executor_scratch`, else the profile's size. | +| `platform.compute.spark.gold_executor_scratch` | string or null | `null` | advanced | Scratch PVC per executor for the gold stage (for example `200Gi`): gold-finalize in batch mode, gold-refresh in continuous mode. Null = `executor_scratch`, else the profile's size. | | `platform.compute.spark.driver_memory` | string or null | `null` | advanced | Global driver memory override (e.g., `16g`). Null = profile default. | | `platform.compute.spark.driver_cores` | integer or null | `null` | advanced | Override driver cores (1--16). Null = profile default (typically 4). | diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index c0de4c19..b14c8d90 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -485,8 +485,11 @@ platform: storage_class: px-csi-scratch ``` -The per-executor scratch size comes from the job profile and should not be -reduced; see [Job Profiles](component-spark.md#job-profiles). Silver-build +The per-executor scratch size comes from the job profile unless +`platform.compute.spark.executor_scratch` or a stage's `*_executor_scratch` +sets it; do not set it below the profile's. To fix a "No space left on device" +in one job, raise that stage's key, or `executor_scratch` for a job outside +the three stages. See [Job Profiles](component-spark.md#job-profiles). Silver-build asks for 60 GiB x scale / executors per executor, between 50Gi and 300Gi. Raising `platform.compute.spark.silver_executors` makes each PVC smaller but does not shrink the total (scale 100: 28 x 215Gi = 6,020 Gi, against 18 x diff --git a/src/lakebench/__init__.py b/src/lakebench/__init__.py index d526d145..a9b99c6c 100644 --- a/src/lakebench/__init__.py +++ b/src/lakebench/__init__.py @@ -1,3 +1,3 @@ """Lakebench - CLI tool for deploying and benchmarking lakehouse architectures on Kubernetes.""" -__version__ = "1.7.11" +__version__ = "1.7.12" diff --git a/src/lakebench/cli/_init.py b/src/lakebench/cli/_init.py index 1ff75fb4..f12e62ea 100644 --- a/src/lakebench/cli/_init.py +++ b/src/lakebench/cli/_init.py @@ -217,6 +217,7 @@ def first_day_dict( "platform.compute.spark.bronze_executor_memory", "platform.compute.spark.silver_executor_memory", "platform.compute.spark.gold_executor_memory", + "platform.compute.spark.executor_scratch", "platform.compute.spark.driver_memory", "platform.compute.spark.driver_cores", "platform.compute.postgres.storage", diff --git a/src/lakebench/config/autosizer.py b/src/lakebench/config/autosizer.py index bbb477b0..56337ae6 100644 --- a/src/lakebench/config/autosizer.py +++ b/src/lakebench/config/autosizer.py @@ -426,10 +426,11 @@ def resolve_auto_sizing( # -- Schema-specific overrides -- # Workload schemas that differ from Customer360 on baseline resource shape # override defaults here. Financial raises the Spark Thrift memory. - # Per-executor scratch PVCs are job.py:scratch_executor_size (the stage's - # scratch_gib_per_scale x scale over its executors, up to the profile's - # scratch_size) with _SCHEMA_PROFILE_OVERRIDES on top, applied at - # manifest-build time in modules/pipeline_engines/spark/job.py. + # Per-executor scratch PVCs are job.py:executor_scratch_size: the size the + # config sets, else the stage's scratch_gib_per_scale x scale over its + # executors, up to the profile's scratch_size, with + # _SCHEMA_PROFILE_OVERRIDES on top, applied at manifest-build time in + # modules/pipeline_engines/spark/job.py. # -- Spark Thrift on Delta -- # Runs before the schema overrides so the financial 24g heap still wins. delta_change = _apply_delta_thrift_default(config, cluster_capacity) @@ -587,12 +588,20 @@ def _apply_delta_thrift_default( def _autoscratch_applies(config: LakebenchConfig) -> bool: from lakebench.config.schema import PipelineMode - return ( + return _scratch_size_set(config) or ( config.architecture.pipeline.mode == PipelineMode.BATCH and config.architecture.workload.datagen.scale >= _AUTOSCRATCH_MIN_SCALE ) +def _scratch_size_set(config: LakebenchConfig) -> bool: + """Whether the config sets a ``platform.compute.spark.*executor_scratch``.""" + from lakebench.config.schema import SCRATCH_OVERRIDE_KEYS + + spark = config.platform.compute.spark + return any(getattr(spark, key) is not None for key in SCRATCH_OVERRIDE_KEYS) + + def scratch_will_be_enabled(config: LakebenchConfig) -> bool: """Whether the run uses scratch PVCs, before or after auto-sizing.""" scratch = config.platform.storage.scratch @@ -602,7 +611,8 @@ def scratch_will_be_enabled(config: LakebenchConfig) -> bool: def _apply_scratch_autoenable(config: LakebenchConfig) -> str | None: - """Enable scratch PVCs on a batch run at scale 50 or above. + """Enable scratch PVCs on a batch run at scale 50 or above, or when the + config sets an executor scratch size. Only sets ``platform.storage.scratch.enabled`` when the user did not set it. ``storage_class`` and ``provisioner`` keep their defaults. ``plan``, @@ -616,6 +626,11 @@ def _apply_scratch_autoenable(config: LakebenchConfig) -> str | None: scratch = config.platform.storage.scratch if not _set_if_default(scratch, "enabled", True): return None + if _scratch_size_set(config): + return ( + "platform.storage.scratch.enabled=True (the config sets an executor " + "scratch size, which needs a scratch PVC per executor)" + ) return ( f"platform.storage.scratch.enabled=True (scale {scale:g}: " "silver-build shuffle exceeds pod ephemeral storage without a " diff --git a/src/lakebench/config/schema.py b/src/lakebench/config/schema.py index a6dbd6dc..3abbf4dc 100644 --- a/src/lakebench/config/schema.py +++ b/src/lakebench/config/schema.py @@ -600,9 +600,8 @@ class ScratchStorageConfig(ConfigModel): "lakebench creates no StorageClass; a cluster admin runs " "'lakebench admin install --component scratch-storage-class' once." ), - # Nothing sized a PVC from it: each executor's scratch volume is the - # job profile's scratch_size (modules/pipeline_engines/spark/job.py). - "size": "per-job scratch comes from the job profiles (silver-build: 60 GiB x scale / executors, 50-300Gi).", + # Nothing sized a PVC from it; the executor scratch keys do. + "size": "each executor's scratch PVC is platform.compute.spark.executor_scratch (or a stage's *_executor_scratch), else its job profile's size.", } _removed_defaults: ClassVar[dict[str, Any]] = {"size": "100Gi"} @@ -715,6 +714,15 @@ def _refuse_install(self, info: ValidationInfo) -> SparkOperatorConfig: # Spark; a bare number (MiB to Spark) and 0 are refused here too, so the # capacity check never misreads one. _SPARK_MEMORY_SIZE = re.compile(r"[1-9][0-9]*[kmgt]b?", re.IGNORECASE) +#: A scratch PVC size: whole gibibytes, as the job profiles write them. +_SCRATCH_SIZE = re.compile(r"[1-9][0-9]*Gi") +#: The ``platform.compute.spark`` keys that size an executor's scratch PVC. +SCRATCH_OVERRIDE_KEYS = ( + "executor_scratch", + "bronze_executor_scratch", + "silver_executor_scratch", + "gold_executor_scratch", +) #: The largest per-job executor override: the proven executor ceiling (32 @@ -768,8 +776,7 @@ class SparkComputeConfig(ConfigModel): # Per-stage executor overrides (None = auto). A stage key sizes the stage's # job in the run's mode: bronze-verify or bronze-ingest, silver-build or - # silver-stream, gold-finalize or gold-refresh. Per-executor memory and - # scratch stay with the proven job profiles. + # silver-stream, gold-finalize or gold-refresh. bronze_executors: int | None = Field( default=None, ge=1, @@ -821,6 +828,25 @@ class SparkComputeConfig(ConfigModel): description="Continuous mode: override gold-refresh heap per executor (a Spark size, for example `96g`); batch mode ignores it. The memory overhead keeps the profile's ratio to the heap, and the capacity check and the node fit count the larger pod. Null = the profile's per-core share of the executor size.", ) + # Executor scratch PVC overrides (None = the job profile's size). A set + # value is each executor's PVC exactly, in either mode. + executor_scratch: str | None = Field( + default=None, + description="Scratch PVC per executor for every Spark job (whole gibibytes, for example `200Gi`). A stage's `*_executor_scratch` wins for that stage's jobs. Setting any scratch size turns `platform.storage.scratch.enabled` on when it is unset; with it set to `false` the config is refused. Null = each job profile's size.", + ) + bronze_executor_scratch: str | None = Field( + default=None, + description="Scratch PVC per executor for the bronze stage (for example `200Gi`): bronze-verify in batch mode, bronze-ingest in continuous mode. Null = `executor_scratch`, else the profile's size.", + ) + silver_executor_scratch: str | None = Field( + default=None, + description="Scratch PVC per executor for the silver stage (for example `300Gi`): silver-build in batch mode, silver-stream in continuous mode. Null = `executor_scratch`, else the profile's size.", + ) + gold_executor_scratch: str | None = Field( + default=None, + description="Scratch PVC per executor for the gold stage (for example `200Gi`): gold-finalize in batch mode, gold-refresh in continuous mode. Null = `executor_scratch`, else the profile's size.", + ) + # Driver resource overrides (None = use profile defaults). # These are global - they apply to all Spark jobs. Use when cluster nodes # have limited memory or when running at extreme scales (500+). @@ -859,6 +885,30 @@ def _driver_memory_is_a_spark_size(cls, value: str | None, info: ValidationInfo) emit_note(text + "; ignored. Commands that change data refuse it.", kind="removed") return None + @field_validator( + "executor_scratch", + "bronze_executor_scratch", + "silver_executor_scratch", + "gold_executor_scratch", + mode="after", + ) + @classmethod + def _scratch_is_whole_gi(cls, value: str | None, info: ValidationInfo) -> str | None: + """A scratch size the PVC request and the capacity check cannot both + read: commands that change data refuse it; teardown and read commands + drop it with a note.""" + if value is None or _SCRATCH_SIZE.fullmatch(value.strip()): + return None if value is None else value.strip() + text = ( + f"platform.compute.spark.{info.field_name} {value!r} is not a scratch size " + "Lakebench can request: write whole gibibytes above 0 (for example 200Gi)" + ) + purpose = purpose_from_context(info.context) + if purpose is None or purpose in CHANGES_DATA: + raise ValueError(text) + emit_note(text + "; ignored. Commands that change data refuse it.", kind="removed") + return None + @field_validator( "bronze_executors", "silver_executors", @@ -1016,6 +1066,29 @@ class PlatformConfig(ConfigModel): deps: DepsConfig = Field(default_factory=DepsConfig) """The deployment's dependency server, `lb-deps`.""" + @model_validator(mode="after") + def _scratch_size_needs_scratch(self, info: ValidationInfo) -> PlatformConfig: + """A scratch size with scratch turned off would size nothing: the + executors spill to pod ephemeral storage. Commands that change data + refuse it; teardown and read commands note it.""" + scratch = self.storage.scratch + if "enabled" not in scratch.model_fields_set or scratch.enabled: + return self + spark = self.compute.spark + set_keys = [k for k in SCRATCH_OVERRIDE_KEYS if getattr(spark, k) is not None] + if not set_keys: + return self + text = ( + f"platform.compute.spark.{set_keys[0]} sizes a scratch PVC, but " + "platform.storage.scratch.enabled is false, so no executor gets one: " + "remove the size or turn scratch on" + ) + purpose = purpose_from_context(info.context) + if purpose is None or purpose in CHANGES_DATA: + raise ValueError(text) + emit_note(text + ". Commands that change data refuse it.", kind="dead") + return self + # ============================================================================= # Data Architecture Configuration (Layer 2) diff --git a/src/lakebench/metrics/collector.py b/src/lakebench/metrics/collector.py index d664c108..74d09fb4 100644 --- a/src/lakebench/metrics/collector.py +++ b/src/lakebench/metrics/collector.py @@ -2925,7 +2925,7 @@ def build_config_snapshot( "scratch": { "enabled": scratch.enabled, "storage_class": scratch.storage_class, - # Each executor's scratch PVC is its job profile's size. + # Each executor's scratch PVC: the configured size, else its job profile's. "size_per_job": scratch_size_per_job(cfg, _continuous), }, "spark": { diff --git a/src/lakebench/metrics/fingerprint_inputs.py b/src/lakebench/metrics/fingerprint_inputs.py index 4900ae58..531c2720 100644 --- a/src/lakebench/metrics/fingerprint_inputs.py +++ b/src/lakebench/metrics/fingerprint_inputs.py @@ -209,12 +209,13 @@ def user_conf_digest(spark_conf: dict[str, Any], user: dict[str, str]) -> str | def scratch_size_per_job(cfg: Any, continuous: bool) -> dict[str, Any]: - """``{job_type: scratch_size}`` from the job profiles, schema overrides applied.""" + """``{job_type: scratch_size}``: the size the config sets, else the job + profile's, schema overrides applied.""" from lakebench.modules.pipeline_engines.spark.job import ( executor_count, executor_override, + executor_scratch_size, get_job_profile, - scratch_executor_size, stage_profile, ) @@ -229,7 +230,7 @@ def scratch_size_per_job(cfg: Any, continuous: bool) -> dict[str, Any]: continue # A batch stage's scratch follows its executor count. executors = executor_override(jt, cfg) or executor_count(profile, scale) - out[jt] = scratch_executor_size(profile, scale, executors) + out[jt] = executor_scratch_size(jt, profile, cfg, scale, executors) return out @@ -287,13 +288,16 @@ def _catalog_block(cfg: Any) -> dict[str, Any]: return {"type": kind, "resources": resources.model_dump(mode="json") if resources else None} -def _scratch_of(profile: dict[str, Any] | None, cfg: Any, executors: int) -> str | None: - """The scratch PVC each of *executors* executors gets under *profile*.""" - from lakebench.modules.pipeline_engines.spark.job import scratch_executor_size +def _scratch_of( + job_type: str, profile: dict[str, Any] | None, cfg: Any, executors: int +) -> str | None: + """The scratch PVC each of *executors* executors of *job_type* gets.""" + from lakebench.modules.pipeline_engines.spark.job import executor_scratch_size if not profile: return None - return scratch_executor_size(profile, float(cfg.architecture.workload.datagen.scale), executors) + scale = float(cfg.architecture.workload.datagen.scale) + return executor_scratch_size(job_type, profile, cfg, scale, executors) def _build(cfg: Any, continuous: bool) -> dict[str, Any]: @@ -329,7 +333,7 @@ def _build(cfg: Any, continuous: bool) -> dict[str, Any]: "executor_memory_overhead": executor["memoryOverhead"], "executor_instances": executor["instances"], "scratch_size": _scratch_of( - get_job_profile(jt, schema), cfg, int(executor["instances"]) + jt, get_job_profile(jt, schema), cfg, int(executor["instances"]) ), } confs[jt] = owned_conf(spec["sparkConf"], user) diff --git a/src/lakebench/modules/pipeline_engines/spark/job.py b/src/lakebench/modules/pipeline_engines/spark/job.py index c4f82041..fbbb4014 100644 --- a/src/lakebench/modules/pipeline_engines/spark/job.py +++ b/src/lakebench/modules/pipeline_engines/spark/job.py @@ -872,6 +872,33 @@ def executor_override(job_type: str, config: Any | None) -> int | None: return int(value) if value is not None else None +def scratch_override(job_type: str, config: Any | None) -> str | None: + """The scratch PVC *config* sets for each of *job_type*'s executors, or + None (the profile's size applies): the stage's ``*_executor_scratch`` + where the stage key applies, else ``executor_scratch``.""" + if config is None: + return None + spark = config.platform.compute.spark + if job_type in EXECUTOR_OVERRIDE_FIELDS and stage_key_applies(job_type, config): + stage = EXECUTOR_OVERRIDE_FIELDS[job_type][0].split("_")[0] + value = { + "bronze": spark.bronze_executor_scratch, + "silver": spark.silver_executor_scratch, + "gold": spark.gold_executor_scratch, + }[stage] + if value is not None: + return str(value) + return str(spark.executor_scratch) if spark.executor_scratch is not None else None + + +def executor_scratch_size( + job_type: str, profile: dict[str, Any], config: Any | None, scale: float, executors: int +) -> str: + """Each of *job_type*'s executors' scratch PVC: the size *config* sets, + else the profile's (``scratch_executor_size``).""" + return scratch_override(job_type, config) or scratch_executor_size(profile, scale, executors) + + def bronze_ingest_checkpoint_uri(cfg) -> str: """Checkpoint location of the continuous bronze-ingest stream. @@ -1080,7 +1107,7 @@ def _job_requirement( gib = 1024**3 memory_bytes = executors * exec_total_bytes + driver_mem scratch_gb = executors * int( - scratch_executor_size(profile, scale, executors).removesuffix("Gi") or 0 + executor_scratch_size(job_type, profile, config, scale, executors).removesuffix("Gi") or 0 ) return JobRequirement( @@ -3041,7 +3068,7 @@ def _mounts() -> list[dict[str, Any]]: scratch = cfg.platform.storage.scratch if scratch.enabled and scratch.storage_class: # Dynamic PVC for executor local storage - pvc_size = scratch_executor_size(profile, scale, executor_count) + pvc_size = executor_scratch_size(job_type.value, profile, cfg, scale, executor_count) spark_conf.update( { "spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-1.options.claimName": "OnDemand", diff --git a/tests/test_executor_scratch.py b/tests/test_executor_scratch.py new file mode 100644 index 00000000..1632a4cd --- /dev/null +++ b/tests/test_executor_scratch.py @@ -0,0 +1,128 @@ +"""The executor scratch keys size the PVC each executor gets, and the run +records the size it requested.""" + +from __future__ import annotations + +from pathlib import Path +from unittest.mock import MagicMock + +import pytest +import yaml + +from lakebench.config import LakebenchConfig +from lakebench.config._load_context import LoadPurpose +from lakebench.config.autosizer import resolve_auto_sizing, scratch_will_be_enabled +from lakebench.config.loader import load_config +from lakebench.metrics.fingerprint_inputs import scratch_size_per_job +from lakebench.modules.pipeline_engines.spark.job import ( + compute_peak_requirements, + scratch_override, +) +from lakebench.spark.job import JobType, SparkJobManager + +_SIZE_KEY = ( + "spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-1.options.sizeLimit" +) + + +def _config(spark: dict, *, scratch: dict | None = None, mode: str = "batch") -> LakebenchConfig: + return LakebenchConfig( + name="t", + platform={ + "storage": { + "s3": {"endpoint": "http://minio:9000", "access_key": "a", "secret_key": "b"}, + "scratch": {"enabled": True} if scratch is None else scratch, + }, + "compute": {"spark": spark}, + }, + architecture={ + "workload": {"schema": "financial", "datagen": {"scale": 20}}, + "pipeline": {"mode": mode}, + "catalog": {"type": "polaris", "polaris": {"client_secret": "test-only-secret"}}, + }, + ) + + +def _pvc(cfg: LakebenchConfig, job: JobType) -> str: + k8s = MagicMock() + k8s.get_cluster_capacity.return_value = None + return SparkJobManager(cfg, k8s)._build_manifest(job)["spec"]["sparkConf"][_SIZE_KEY] + + +def test_executor_scratch_sizes_the_aml_scorer(): + """The AML scorer's profile gives 50Gi, which a scale-20 run filled; + executor_scratch is the one key that reaches it.""" + assert _pvc(_config({}), JobType.SCORE_FINANCIAL) == "50Gi" + assert _pvc(_config({"executor_scratch": "200Gi"}), JobType.SCORE_FINANCIAL) == "200Gi" + + +def test_a_stage_key_wins_for_its_stage_only(): + cfg = _config({"executor_scratch": "200Gi", "silver_executor_scratch": "400Gi"}) + assert _pvc(cfg, JobType.SILVER_BUILD) == "400Gi" + assert _pvc(cfg, JobType.GOLD_FINALIZE) == "200Gi" + assert _pvc(cfg, JobType.SCORE_FINANCIAL) == "200Gi" + + +def test_the_size_is_exact_not_scaled_by_cores(): + cfg = _config( + {"silver_executor_scratch": "120Gi", "silver_executor_cores": 16}, mode="continuous" + ) + assert _pvc(cfg, JobType.SILVER_STREAM) == "120Gi" + + +def test_a_continuous_runs_bronze_verify_preflight_takes_only_the_global_key(): + cfg = _config( + {"executor_scratch": "200Gi", "bronze_executor_scratch": "80Gi"}, mode="continuous" + ) + assert scratch_override("bronze-ingest", cfg) == "80Gi" + assert scratch_override("bronze-verify", cfg) == "200Gi" + + +def test_the_capacity_plan_counts_the_set_size(): + cfg = _config({"silver_executor_scratch": "400Gi"}) + peak = compute_peak_requirements(20, "batch", schema_type="financial", config=cfg) + sb = next(r for r in peak.per_job if r.job_type == "silver-build") + assert sb.scratch_gb == sb.executors * 400 + + +def test_the_run_record_holds_the_set_size(): + cfg = _config({"executor_scratch": "200Gi", "silver_executor_scratch": "400Gi"}) + sizes = scratch_size_per_job(cfg, continuous=False) + assert sizes["silver-build"] == "400Gi" + assert sizes["bronze-verify"] == sizes["gold-finalize"] == "200Gi" + assert scratch_size_per_job(_config({}), continuous=False)["silver-build"] != "400Gi" + + +def _write(tmp_path: Path, spark: dict, scratch: dict | None = None) -> Path: + raw = _config({}).model_dump(mode="json", by_alias=True, exclude_none=True) + raw["platform"]["compute"]["spark"].update(spark) + if scratch is not None: + raw["platform"]["storage"]["scratch"] = scratch + path = tmp_path / "c.yaml" + path.write_text(yaml.safe_dump(raw)) + return path + + +@pytest.mark.parametrize("value", ["200G", "200g", "0Gi", "1.5Gi", "200Mi"]) +def test_a_size_not_in_whole_gi_is_refused_and_dropped_for_teardown(tmp_path, value): + path = _write(tmp_path, {"executor_scratch": value}) + with pytest.raises(Exception, match="not a scratch size"): + load_config(path, purpose=LoadPurpose.RUN, print_notes=False) + cfg = load_config(path, purpose=LoadPurpose.TEARDOWN, print_notes=False) + assert cfg.platform.compute.spark.executor_scratch is None + + +def test_a_size_with_scratch_turned_off_is_refused(tmp_path): + path = _write(tmp_path, {"gold_executor_scratch": "200Gi"}, scratch={"enabled": False}) + with pytest.raises(Exception, match="scratch.enabled is false"): + load_config(path, purpose=LoadPurpose.RUN, print_notes=False) + assert load_config(path, purpose=LoadPurpose.TEARDOWN, print_notes=False) + + +def test_a_size_turns_scratch_on_when_it_is_unset(): + cfg = _config({"executor_scratch": "200Gi"}, scratch={}, mode="continuous") + assert scratch_will_be_enabled(cfg) + resolve_auto_sizing(cfg) + assert cfg.platform.storage.scratch.enabled + plain = _config({}, scratch={}, mode="continuous") + assert not scratch_will_be_enabled(plain)