diff --git a/pyproject.toml b/pyproject.toml index ed5d9050..0c175c4e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,6 +12,7 @@ requires-python = ">=3.10,<3.13" dependencies = [ "pyspark>=3.4.0,<4.0.0", "pandas>=1.3.4,<3.0.0", + "pyarrow>=4.0.0", "networkx>=2.6.3,<3.0.0", "pyfarmhash>=0.3.2,<0.4.0", "keras>=3.0.0,<4.0.0", diff --git a/src/kamae/spark/estimators/conditional_standard_scale.py b/src/kamae/spark/estimators/conditional_standard_scale.py index b6edfb2f..30f6ae65 100644 --- a/src/kamae/spark/estimators/conditional_standard_scale.py +++ b/src/kamae/spark/estimators/conditional_standard_scale.py @@ -25,6 +25,7 @@ from pyspark.ml.param import Param, Params, TypeConverters from pyspark.sql import Column, DataFrame from pyspark.sql.types import ArrayType, DataType, DoubleType, FloatType +from pyspark.storagelevel import StorageLevel from kamae.keras.core.backend import ALL_BACKENDS from kamae.spark.params import ( @@ -378,22 +379,37 @@ def _fit(self, dataset: DataFrame) -> "ConditionalStandardScaleTransformer": mask_val = self.getMaskValues()[i] dataset = dataset.filter(mask_op(F.col(mask_col), mask_val)) - # Collect a single row to driver and get the length. - # We assume all subsequent rows have the same length. - row = dataset.select(input_col).first() - if row is None: - raise ValueError("No data left after application of mask conditions.") - array_size = np.array((row[0])).shape[-1] - - # Calculate the moments - if self.getScalingFunction().lower() == "standard": - return self._fit_standard( - dataset, input_col, input_column_dtype, array_size - ) - elif self.getScalingFunction().lower() == "binary": - return self._fit_binary(dataset, input_col, input_column_dtype, array_size) - else: - raise ValueError(f"Unknown scaling function: {self.getScalingFunction()}.") + # Persist so the array-size probe and the moments aggregation reuse a + # materialised result instead of re-scanning the (masked) upstream lineage + # twice. Guarded so we do not double-persist data the caller already cached. + already_cached = dataset.storageLevel.useMemory or dataset.storageLevel.useDisk + if not already_cached: + dataset = dataset.persist(StorageLevel.MEMORY_AND_DISK) + + try: + # Collect a single row to driver and get the length. + # We assume all subsequent rows have the same length. + row = dataset.select(input_col).first() + if row is None: + raise ValueError("No data left after application of mask conditions.") + array_size = np.array((row[0])).shape[-1] + + # Calculate the moments + if self.getScalingFunction().lower() == "standard": + return self._fit_standard( + dataset, input_col, input_column_dtype, array_size + ) + elif self.getScalingFunction().lower() == "binary": + return self._fit_binary( + dataset, input_col, input_column_dtype, array_size + ) + else: + raise ValueError( + f"Unknown scaling function: {self.getScalingFunction()}." + ) + finally: + if not already_cached: + dataset.unpersist() def _fit_binary( self, diff --git a/src/kamae/spark/estimators/single_feature_array_standard_scale.py b/src/kamae/spark/estimators/single_feature_array_standard_scale.py index 4c209893..a197ae4c 100644 --- a/src/kamae/spark/estimators/single_feature_array_standard_scale.py +++ b/src/kamae/spark/estimators/single_feature_array_standard_scale.py @@ -19,6 +19,7 @@ from pyspark import keyword_only from pyspark.sql import DataFrame from pyspark.sql.types import ArrayType, DataType, DoubleType, FloatType +from pyspark.storagelevel import StorageLevel from kamae.keras.core.backend import ALL_BACKENDS from kamae.spark.params import ( @@ -113,32 +114,45 @@ def _fit(self, dataset: DataFrame) -> "StandardScaleTransformer": Got {input_column_type} instead.""" ) - # Collect a single row to driver and get the length. - # We assume all subsequent rows have the same length. - array_size = np.array((dataset.select(self.getInputCol()).first()[0])).shape[-1] - - # Flatten the array to a single array. - # Will do nothing if the array is not nested. - flattened_array_col = flatten_nested_arrays( - column=F.col(self.getInputCol()), column_data_type=input_column_type - ) - - mean_and_stddev_dict: Dict[str, float] = ( - dataset.select(F.explode(flattened_array_col).alias(self.getInputCol())) - .withColumn( - "mask", - F.when( - F.col(self.getInputCol()) == F.lit(self.getMaskValue()), 1 - ).otherwise(0), + # Persist so the array-size probe and the moments aggregation reuse a + # materialised result instead of re-scanning the upstream lineage twice. + # Guarded so we do not double-persist data the caller already cached. + already_cached = dataset.storageLevel.useMemory or dataset.storageLevel.useDisk + if not already_cached: + dataset = dataset.persist(StorageLevel.MEMORY_AND_DISK) + + try: + # Collect a single row to driver and get the length. + # We assume all subsequent rows have the same length. + array_size = np.array( + (dataset.select(self.getInputCol()).first()[0]) + ).shape[-1] + + # Flatten the array to a single array. + # Will do nothing if the array is not nested. + flattened_array_col = flatten_nested_arrays( + column=F.col(self.getInputCol()), column_data_type=input_column_type ) - .filter(F.col("mask") == F.lit(0)) - .agg( - F.mean(self.getInputCol()).alias("mean"), - F.stddev_pop(self.getInputCol()).alias("stddev"), + + mean_and_stddev_dict: Dict[str, float] = ( + dataset.select(F.explode(flattened_array_col).alias(self.getInputCol())) + .withColumn( + "mask", + F.when( + F.col(self.getInputCol()) == F.lit(self.getMaskValue()), 1 + ).otherwise(0), + ) + .filter(F.col("mask") == F.lit(0)) + .agg( + F.mean(self.getInputCol()).alias("mean"), + F.stddev_pop(self.getInputCol()).alias("stddev"), + ) + .first() + .asDict() ) - .first() - .asDict() - ) + finally: + if not already_cached: + dataset.unpersist() mean: List[float] = [mean_and_stddev_dict["mean"] for _ in range(array_size)] stddev: List[float] = [ mean_and_stddev_dict["stddev"] for _ in range(array_size) diff --git a/src/kamae/spark/estimators/standard_scale.py b/src/kamae/spark/estimators/standard_scale.py index a1c654ea..379dce37 100644 --- a/src/kamae/spark/estimators/standard_scale.py +++ b/src/kamae/spark/estimators/standard_scale.py @@ -23,6 +23,7 @@ from pyspark import keyword_only from pyspark.sql import DataFrame from pyspark.sql.types import ArrayType, DataType, DoubleType, FloatType +from pyspark.storagelevel import StorageLevel from kamae.keras.core.backend import ALL_BACKENDS from kamae.spark.params import ( @@ -113,41 +114,54 @@ def _fit(self, dataset: DataFrame) -> "StandardScaleTransformer": else: input_col = F.col(self.getInputCol()) - # Collect a single row to driver and get the length. - # We assume all subsequent rows have the same length. - array_size = np.array((dataset.select(input_col).first()[0])).shape[-1] - - element_struct = construct_nested_elements_for_scaling( - column=input_col, - column_datatype=input_column_type, - array_dim=array_size, - ) - - mean_cols = [ - F.mean( - F.when( - F.col(f"element_struct.element_{i}") == F.lit(self.getMaskValue()), - F.lit(None), - ).otherwise(F.col(f"element_struct.element_{i}")) - ).alias(f"mean_{i}") - for i in range(1, array_size + 1) - ] - - stddev_cols = [ - F.stddev_pop( - F.when( - F.col(f"element_struct.element_{i}") == F.lit(self.getMaskValue()), - F.lit(None), - ).otherwise(F.col(f"element_struct.element_{i}")) - ).alias(f"stddev_{i}") - for i in range(1, array_size + 1) - ] - - metric_cols = mean_cols + stddev_cols - - mean_and_stddev_dict = ( - dataset.select(element_struct).agg(*metric_cols).first().asDict() - ) + # Persist so the array-size probe and the moments aggregation reuse a + # materialised result instead of re-scanning the upstream lineage twice. + # Guarded so we do not double-persist data the caller already cached. + already_cached = dataset.storageLevel.useMemory or dataset.storageLevel.useDisk + if not already_cached: + dataset = dataset.persist(StorageLevel.MEMORY_AND_DISK) + + try: + # Collect a single row to driver and get the length. + # We assume all subsequent rows have the same length. + array_size = np.array((dataset.select(input_col).first()[0])).shape[-1] + + element_struct = construct_nested_elements_for_scaling( + column=input_col, + column_datatype=input_column_type, + array_dim=array_size, + ) + + mean_cols = [ + F.mean( + F.when( + F.col(f"element_struct.element_{i}") + == F.lit(self.getMaskValue()), + F.lit(None), + ).otherwise(F.col(f"element_struct.element_{i}")) + ).alias(f"mean_{i}") + for i in range(1, array_size + 1) + ] + + stddev_cols = [ + F.stddev_pop( + F.when( + F.col(f"element_struct.element_{i}") + == F.lit(self.getMaskValue()), + F.lit(None), + ).otherwise(F.col(f"element_struct.element_{i}")) + ).alias(f"stddev_{i}") + for i in range(1, array_size + 1) + ] + + metric_cols = mean_cols + stddev_cols + + mean_and_stddev_dict = ( + dataset.select(element_struct).agg(*metric_cols).first().asDict() + ) + finally: + if not already_cached: + dataset.unpersist() mean = [mean_and_stddev_dict[f"mean_{i}"] for i in range(1, array_size + 1)] stddev = [mean_and_stddev_dict[f"stddev_{i}"] for i in range(1, array_size + 1)] diff --git a/src/kamae/spark/pipeline/pipeline.py b/src/kamae/spark/pipeline/pipeline.py index 1da0d7ea..57b5aee4 100644 --- a/src/kamae/spark/pipeline/pipeline.py +++ b/src/kamae/spark/pipeline/pipeline.py @@ -12,15 +12,17 @@ # See the License for the specific language governing permissions and # limitations under the License. -from typing import TYPE_CHECKING, List, Optional, Type +import warnings +from typing import TYPE_CHECKING, List, Optional, Set, Type import networkx as nx from pyspark import keyword_only from pyspark.ml import Pipeline -from pyspark.ml.param import Params +from pyspark.ml.param import Param, Params, TypeConverters from pyspark.ml.pipeline import PipelineReader, PipelineSharedReadWrite, PipelineWriter from pyspark.ml.util import DefaultParamsReader, MLWriter from pyspark.sql import DataFrame +from pyspark.storagelevel import StorageLevel from kamae.graph import PipelineGraph from kamae.spark.estimators import BaseEstimator @@ -38,17 +40,87 @@ class KamaeSparkPipeline(Pipeline): KamaeSparkPipeline is a subclass of pyspark.ml.Pipeline that is used to chain together BaseTransformers. It maintains the same functionality as pyspark.ml.Pipeline e.g. serialisation. + + Four opt-in fit optimisations are available, all defaulting off (fit behaviour + unchanged): `checkpointInterval` reliably checkpoints every N stages to bound + logical-plan depth (requires a checkpoint dir); `cacheIntermediateData` persists + the working DataFrame at each estimator-fit boundary to avoid re-scanning the + upstream lineage; `pruneInputColumns` drops input columns no stage consumes; + `cacheEstimatorInput` projects to the columns still read downstream at the first + estimator boundary and persists that narrow frame once, so independent sibling + estimators reuse it instead of re-scanning the wide input. """ + checkpointInterval = Param( + Params._dummy(), + "checkpointInterval", + "Stages between reliable checkpoint(eager=True) calls during fit, to bound " + "logical-plan depth. Requires a checkpoint dir. None (default) disables it.", + typeConverter=TypeConverters.toInt, + ) + + cacheIntermediateData = Param( + Params._dummy(), + "cacheIntermediateData", + "If True, persist the working DataFrame (MEMORY_AND_DISK) at each " + "estimator-fit boundary to avoid re-scanning the upstream lineage. False " + "(default) disables it.", + typeConverter=TypeConverters.toBoolean, + ) + + pruneInputColumns = Param( + Params._dummy(), + "pruneInputColumns", + "If True, drop input columns no stage consumes before fitting. False " + "(default) disables it.", + typeConverter=TypeConverters.toBoolean, + ) + + cacheEstimatorInput = Param( + Params._dummy(), + "cacheEstimatorInput", + "If True, at the first estimator-fit boundary project the working DataFrame " + "to the columns still read downstream and persist (MEMORY_AND_DISK) that " + "narrow frame once, reused by all subsequent estimators. Competes with " + "cacheIntermediateData; enable at most one. False (default) disables it.", + typeConverter=TypeConverters.toBoolean, + ) + @keyword_only - def __init__(self, *, stages: Optional[List["KamaePipelineStage"]] = None) -> None: + def __init__( + self, + *, + stages: Optional[List["KamaePipelineStage"]] = None, + checkpointInterval: Optional[int] = None, + cacheIntermediateData: bool = False, + pruneInputColumns: bool = False, + cacheEstimatorInput: bool = False, + ) -> None: """ Initialises the KamaeSparkPipeline object. :param stages: List of LayerTransformers to chain together. + :param checkpointInterval: Number of stages between reliable + checkpoint(eager=True) calls during fit. None (default) disables it. + :param cacheIntermediateData: If True, persist the working DataFrame at + each estimator-fit boundary to avoid re-scanning the upstream lineage. + False (default) disables it. + :param pruneInputColumns: If True, drop input columns no stage consumes + before fitting. False (default) disables it. + :param cacheEstimatorInput: If True, project to the columns still read + downstream at the first estimator boundary and persist that narrow frame + once for reuse by subsequent estimators. False (default) disables it. :returns: None - class instantiated. """ + kwargs = self._input_kwargs super().__init__(stages=stages) + self._setDefault( + checkpointInterval=None, + cacheIntermediateData=False, + pruneInputColumns=False, + cacheEstimatorInput=False, + ) + self.setParams(**kwargs) def setStages(self, value: List["KamaePipelineStage"]) -> "KamaeSparkPipeline": """ @@ -67,18 +139,115 @@ def getStages(self) -> List["KamaePipelineStage"]: """ return self.getOrDefault("stages") + def setCheckpointInterval(self, value: Optional[int]) -> "KamaeSparkPipeline": + """ + Sets the `checkpointInterval` parameter. + + :param value: Positive number of stages between reliable checkpoint calls + during fit. None disables checkpointing. + :returns: KamaeSparkPipeline object with checkpointInterval set. + :raises ValueError: If value is not None and not a positive integer. + """ + if value is not None and value < 1: + raise ValueError( + "checkpointInterval must be a positive integer or None, got " + f"{value}." + ) + return self._set(checkpointInterval=value) + + def getCheckpointInterval(self) -> Optional[int]: + """ + Gets the value of the `checkpointInterval` parameter. + + :returns: The checkpointInterval value. + """ + return self.getOrDefault(self.checkpointInterval) + + def setCacheIntermediateData(self, value: bool) -> "KamaeSparkPipeline": + """ + Sets the `cacheIntermediateData` parameter. + + :param value: Whether to persist the working DataFrame at each + estimator-fit boundary during fit. + :returns: KamaeSparkPipeline object with cacheIntermediateData set. + """ + return self._set(cacheIntermediateData=value) + + def getCacheIntermediateData(self) -> bool: + """ + Gets the value of the `cacheIntermediateData` parameter. + + :returns: The cacheIntermediateData value. + """ + return self.getOrDefault(self.cacheIntermediateData) + + def setPruneInputColumns(self, value: bool) -> "KamaeSparkPipeline": + """ + Sets the `pruneInputColumns` parameter. + + :param value: Whether to drop input columns no stage consumes before fitting. + :returns: KamaeSparkPipeline object with pruneInputColumns set. + """ + return self._set(pruneInputColumns=value) + + def getPruneInputColumns(self) -> bool: + """ + Gets the value of the `pruneInputColumns` parameter. + + :returns: The pruneInputColumns value. + """ + return self.getOrDefault(self.pruneInputColumns) + + def setCacheEstimatorInput(self, value: bool) -> "KamaeSparkPipeline": + """ + Sets the `cacheEstimatorInput` parameter. + + :param value: Whether to project and persist a narrow estimator-input + frame once at the first estimator boundary during fit. + :returns: KamaeSparkPipeline object with cacheEstimatorInput set. + """ + return self._set(cacheEstimatorInput=value) + + def getCacheEstimatorInput(self) -> bool: + """ + Gets the value of the `cacheEstimatorInput` parameter. + + :returns: The cacheEstimatorInput value. + """ + return self.getOrDefault(self.cacheEstimatorInput) + @keyword_only def setParams( - self, *, stages: Optional["KamaePipelineStage"] = None + self, + *, + stages: Optional["KamaePipelineStage"] = None, + checkpointInterval: Optional[int] = None, + cacheIntermediateData: bool = False, + pruneInputColumns: bool = False, + cacheEstimatorInput: bool = False, ) -> "KamaeSparkPipeline": """ Sets the keyword arguments of the pipeline. + Routes each supplied param through its setter so setter-level validation + (e.g. checkpointInterval) runs. + :param stages: List of pipeline stages. - :returns: KamaeSparkPipeline object with stages set. + :param checkpointInterval: Number of stages between reliable + checkpoint(eager=True) calls during fit. None (default) disables it. + :param cacheIntermediateData: If True, persist the working DataFrame at + each estimator-fit boundary. False (default) disables it. + :param pruneInputColumns: If True, drop input columns no stage consumes + before fitting. False (default) disables it. + :param cacheEstimatorInput: If True, project to the columns still read + downstream at the first estimator boundary and persist that narrow frame + once for reuse by subsequent estimators. False (default) disables it. + :returns: KamaeSparkPipeline object with params set. """ - kwargs = self._input_kwargs - return self._set(**kwargs) + for param_name, param_value in self._input_kwargs.items(): + setter = getattr(self, f"set{param_name[0].upper()}{param_name[1:]}") + setter(param_value) + return self def expand_pipeline_stages(self) -> List["KamaePipelineStage"]: """ @@ -132,6 +301,99 @@ def collect_estimator_parents( ] return estimator_parent_stages + @staticmethod + def collect_required_input_columns( + stages: List["KamaePipelineStage"], + ) -> Set[str]: + """ + Collects every column potentially read by any stage in the pipeline. + + Generous by design: unions canonical inputs with the value(s) of every param + whose name ends in `Col`/`Cols`, so aux columns read during fit (e.g. + maskCols, relevanceCol, queryIdCol) are not missed. Over-inclusion is + harmless (names not matching `dataset.columns` are ignored); omission would + wrongly drop data the pipeline needs at fit time. + + :param stages: List of pipeline stages. + :returns: Set of column names potentially read by at least one stage. + """ + required_input_columns: Set[str] = set() + for stage in stages: + inputs, _ = stage.get_layer_inputs_outputs() + required_input_columns.update(inputs) + for param in stage.params: + if not (param.name.endswith("Col") or param.name.endswith("Cols")): + continue + if not stage.isDefined(param): + continue + value = stage.getOrDefault(param) + if isinstance(value, str): + required_input_columns.add(value) + elif isinstance(value, (list, tuple)): + required_input_columns.update( + item for item in value if isinstance(item, str) + ) + return required_input_columns + + def prune_unused_input_columns( + self, + dataset: DataFrame, + stages: List["KamaePipelineStage"], + ) -> DataFrame: + """ + Projects the input DataFrame down to only the columns the pipeline reads. + + Returned unchanged if there are no unused columns to drop. + + :param dataset: Input DataFrame to prune. + :param stages: Expanded pipeline stages. + :returns: DataFrame projected to the columns the pipeline consumes. + """ + required_input_columns = self.collect_required_input_columns(stages) + columns_to_keep = [c for c in dataset.columns if c in required_input_columns] + if columns_to_keep: + return dataset.select(*columns_to_keep) + return dataset + + @staticmethod + def _validate_stage_types(stages: List["KamaePipelineStage"]) -> None: + """ + Ensures every expanded stage is a recognised estimator or transformer. + + :param stages: Expanded pipeline stages. + :raises TypeError: If any stage is not a BaseEstimator or BaseTransformer. + """ + for stage in stages: + if not isinstance(stage, (BaseEstimator, BaseTransformer)): + raise TypeError( + "Cannot recognize a pipeline stage of type %s." % type(stage) + ) + + @staticmethod + def _resolve_checkpoint_enabled( + dataset: DataFrame, checkpoint_interval: Optional[int] + ) -> bool: + """ + Determines whether checkpointing is enabled and validates its prerequisites. + + Fails fast if enabled without a checkpoint dir, rather than raising mid-fit. + + :param dataset: DataFrame whose SparkContext is checked for a checkpoint dir. + :param checkpoint_interval: Configured checkpoint interval (0/None disables). + :returns: True if checkpointing is enabled, False otherwise. + :raises ValueError: If enabled but no checkpoint directory has been set. + """ + checkpoint_enabled = checkpoint_interval is not None and checkpoint_interval > 0 + if ( + checkpoint_enabled + and dataset.sparkSession.sparkContext.getCheckpointDir() is None + ): + raise ValueError( + "checkpointInterval > 0 requires a checkpoint directory. Set one via " + "spark.sparkContext.setCheckpointDir() before fitting." + ) + return checkpoint_enabled + def _fit(self, dataset: DataFrame) -> "KamaeSparkPipelineModel": """ Fits the pipeline to the dataset. Returns a KamaeSparkPipelineModel object. @@ -139,18 +401,27 @@ def _fit(self, dataset: DataFrame) -> "KamaeSparkPipelineModel": Calls the super fit method of the pyspark.ml.Pipeline class and then constructs a KamaeSparkPipelineModel uses the stages from the fit pipeline. + Optionally applies the opt-in fit optimisations (`pruneInputColumns`, + `checkpointInterval`, `cacheIntermediateData`, `cacheEstimatorInput`); see the + class docstring. All preserve data exactly, so fitted results match the + defaults-off behaviour. + + If both `cacheIntermediateData` and `cacheEstimatorInput` are enabled, + `cacheEstimatorInput` takes precedence (a warning is emitted) and + `cacheIntermediateData` is ignored, since the narrow frame is a strictly + smaller cache and the intermediate cache would evict it. + :param dataset: PySpark DataFrame to fit the pipeline to. :returns: KamaeSparkPipelineModel object. + :raises ValueError: If checkpointing is enabled but no checkpoint directory + has been set on the SparkContext. """ expanded_pipeline_stages = self.expand_pipeline_stages() + self._validate_stage_types(expanded_pipeline_stages) - for stage in expanded_pipeline_stages: - if not ( - isinstance(stage, BaseEstimator) or isinstance(stage, BaseTransformer) - ): - raise TypeError( - "Cannot recognize a pipeline stage of type %s." % type(stage) - ) + # Opt-in: drop input columns no stage consumes. Default False = no change. + if self.getPruneInputColumns(): + dataset = self.prune_unused_input_columns(dataset, expanded_pipeline_stages) # Native Spark checks for the last estimator and executes all transformers # before it, regardless whether there is a dependency between them. See here: @@ -162,19 +433,79 @@ def _fit(self, dataset: DataFrame) -> "KamaeSparkPipelineModel": estimator_parent_stages = self.collect_estimator_parents( expanded_pipeline_stages ) + # Opt-in plan-depth bounding. 0 (or None) = no change. + checkpoint_interval = self.getCheckpointInterval() + checkpoint_enabled = self._resolve_checkpoint_enabled( + dataset, checkpoint_interval + ) + cache_enabled = self.getCacheIntermediateData() + cache_estimator_input = self.getCacheEstimatorInput() + # Competing caching strategies - both would persist the wide frame, and the + # intermediate cache's per-boundary unpersist would evict the narrow frame. + # cacheEstimatorInput wins: it persists a strictly narrower frame once. + if cache_enabled and cache_estimator_input: + warnings.warn( + "cacheIntermediateData and cacheEstimatorInput are competing " + "caching strategies; cacheEstimatorInput takes precedence and " + "cacheIntermediateData is ignored.", + stacklevel=2, + ) + cache_enabled = False + last_checkpoint_index = 0 + # The single persisted frame (if any), unpersisted once superseded or done. + cached_dataset: Optional[DataFrame] = None + # The narrow estimator-input frame (if any), persisted once and reused. + estimator_input_cache: Optional[DataFrame] = None + estimator_input_cached = False # Fit each stage, appending the transformer to the list of transformers # If the stage is a parent of an estimator, transform the dataset. transformers: List[BaseTransformer] = [] - for stage in expanded_pipeline_stages: + for index, stage in enumerate(expanded_pipeline_stages): if isinstance(stage, BaseTransformer): transformers.append(stage) if stage in estimator_parent_stages: dataset = stage.transform(dataset) else: + # Opt-in: at the first estimator boundary, project to the columns + # still read downstream and persist that narrow frame once. + # Independent sibling estimators then fit against the cached narrow + # frame instead of re-scanning the wide input. The persist sits + # below each estimator's in-fit sample, so sampling is unchanged. + if cache_estimator_input and not estimator_input_cached: + estimator_input_cached = True + live_columns = self.collect_required_input_columns( + expanded_pipeline_stages[index:] + ) + keep_columns = [c for c in dataset.columns if c in live_columns] + if keep_columns and len(keep_columns) < len(dataset.columns): + estimator_input_cache = dataset.select(*keep_columns).persist( + StorageLevel.MEMORY_AND_DISK + ) + dataset = estimator_input_cache + # Truncate accumulated lineage before the fit action to bound plan + # depth. eager=True materialises now. + if ( + checkpoint_enabled + and index - last_checkpoint_index >= checkpoint_interval + ): + dataset = dataset.checkpoint(eager=True) + last_checkpoint_index = index + # Persist so the fit action and downstream transforms reuse a + # materialised frame instead of re-scanning. One frame held at a time. + if cache_enabled: + new_cached = dataset.persist(StorageLevel.MEMORY_AND_DISK) + if cached_dataset is not None: + cached_dataset.unpersist() + cached_dataset = new_cached + dataset = new_cached model = stage.fit(dataset) transformers.append(model) if stage in estimator_parent_stages: dataset = model.transform(dataset) + if cached_dataset is not None: + cached_dataset.unpersist() + if estimator_input_cache is not None: + estimator_input_cache.unpersist() return KamaeSparkPipelineModel(transformers) def copy(self, extra: Optional["ParamMap"] = None) -> "KamaeSparkPipeline": diff --git a/src/kamae/spark/transformers/bucketize.py b/src/kamae/spark/transformers/bucketize.py index b639f0cc..e9055cf7 100644 --- a/src/kamae/spark/transformers/bucketize.py +++ b/src/kamae/spark/transformers/bucketize.py @@ -16,21 +16,21 @@ # pylint: disable=invalid-name # pylint: disable=too-many-ancestors # pylint: disable=no-member -from bisect import bisect_right -from typing import List, Optional, Union +from functools import reduce +from typing import List, Optional import pyspark.sql.functions as F import tensorflow as tf from pyspark import keyword_only from pyspark.ml.param import Param, Params, TypeConverters -from pyspark.sql import DataFrame +from pyspark.sql import Column, DataFrame from pyspark.sql.types import DataType, DoubleType, FloatType, IntegerType, LongType from kamae.keras.core.backend import TENSORFLOW_ONLY from kamae.keras.tensorflow.layers import BucketizeLayer from kamae.spark.params import SingleInputSingleOutputParams from kamae.spark.utils.transform_utils import ( - single_input_single_output_scalar_udf_transform, + single_input_single_output_scalar_transform, ) from .base import BaseTransformer @@ -90,9 +90,8 @@ class BucketizeTransformer( The 0 index is reserved for masking/padding. """ - jit_compatible = True - supported_backends = TENSORFLOW_ONLY + jit_compatible = True @keyword_only def __init__( @@ -113,7 +112,7 @@ def __init__( transforming. :param outputDtype: Output data type to cast the output column to after transforming. - :param layerName: Name of the layer. Used as the name of the Keras layer + :param layerName: Name of the layer. Used as the name of the tensorflow layer in the keras model. If not set, we use the uid of the Spark transformer. :param splits: List of float values to use for bucketing. :returns: None - class instantiated. @@ -141,23 +140,26 @@ def _transform(self, dataset: DataFrame) -> DataFrame: :returns: Transformed pyspark dataframe. """ splits = self.getSplits() - # We need to create a UDF to perform binary search on the splits. - def bucketize(value: Optional[Union[float, int]]) -> Optional[int]: - # If null, keep null. There is no best bucket to place these into. - if value is None: - return None - # We add 1 because we want to reserve the 0 index for mask/padding. - return bisect_right(splits, value) + 1 + def bucketize(value: Column) -> Column: + # Bucket index equals the number of splits <= value (matching + # bisect_right on a sorted splits list), plus 1 to reserve index 0 for + # mask/padding. Nulls are kept null - there is no best bucket for them. + bucket = reduce( + lambda acc, s: acc + + F.when(value >= F.lit(s), F.lit(1)).otherwise(F.lit(0)), + splits, + F.lit(1), + ) + return F.when(value.isNull(), F.lit(None)).otherwise(bucket) input_datatype = self.get_column_datatype( dataset=dataset, column_name=self.getInputCol() ) - output_col = single_input_single_output_scalar_udf_transform( + output_col = single_input_single_output_scalar_transform( input_col=F.col(self.getInputCol()), input_col_datatype=input_datatype, - func=lambda x: bucketize(x), - udf_return_element_datatype=IntegerType(), + func=bucketize, ) return dataset.withColumn( diff --git a/src/kamae/spark/utils/transform_utils.py b/src/kamae/spark/utils/transform_utils.py index c67d8773..87425aeb 100644 --- a/src/kamae/spark/utils/transform_utils.py +++ b/src/kamae/spark/utils/transform_utils.py @@ -11,9 +11,9 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. - from typing import Callable, List +import pandas as pd import pyspark.sql.functions as F from pyspark.sql import Column from pyspark.sql.types import ArrayType, DataType @@ -150,6 +150,18 @@ def _single_input_single_output_udf_transform( func=func, nest_level=nested_level, ) + # Scalar (non-array) columns transfer as a flat Arrow batch, so a pandas_udf + # that maps the same per-element func avoids the per-row pickling of a plain + # Python UDF (~1.4x faster). Nested-array columns are kept on the row-wise UDF: + # Arrow (de)serialisation of nested lists there costs more than it saves. + if not isinstance(input_col_datatype, ArrayType): + + def _vectorized_func(series: pd.Series) -> pd.Series: + return series.map(nested_lambda_func) + + udf_func = F.pandas_udf(_vectorized_func, udf_return_type) + return udf_func(input_col) + udf_func = F.udf(nested_lambda_func, udf_return_type) return udf_func(input_col) diff --git a/tests/kamae/spark/conftest.py b/tests/kamae/spark/conftest.py index 0d356a68..ce7b2b84 100644 --- a/tests/kamae/spark/conftest.py +++ b/tests/kamae/spark/conftest.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import tempfile from typing import List, Optional import pytest @@ -43,8 +44,10 @@ def spark_session(): .config("spark.driver.memory", "2g") .getOrCreate() ) - yield spark - spark.stop() + with tempfile.TemporaryDirectory() as checkpoint_dir: + spark.sparkContext.setCheckpointDir(checkpoint_dir) + yield spark + spark.stop() @pytest.fixture(scope="module") diff --git a/tests/kamae/spark/pipeline/test_pipeline.py b/tests/kamae/spark/pipeline/test_pipeline.py index 03c7ddbd..dd5b040a 100644 --- a/tests/kamae/spark/pipeline/test_pipeline.py +++ b/tests/kamae/spark/pipeline/test_pipeline.py @@ -12,14 +12,20 @@ # See the License for the specific language governing permissions and # limitations under the License. -import os +import tempfile from shutil import rmtree +from unittest.mock import patch import pytest import tensorflow as tf +from pyspark.sql import DataFrame from pyspark.sql.types import DoubleType -from kamae.spark.estimators import StandardScaleEstimator, StringIndexEstimator +from kamae.spark.estimators import ( + ConditionalStandardScaleEstimator, + StandardScaleEstimator, + StringIndexEstimator, +) from kamae.spark.pipeline import KamaeSparkPipeline, KamaeSparkPipelineModel from kamae.spark.transformers import ( ArrayConcatenateTransformer, @@ -27,6 +33,7 @@ BucketizeTransformer, HashIndexTransformer, IdentityTransformer, + ListMeanTransformer, LogTransformer, SubtractTransformer, ) @@ -39,8 +46,7 @@ class TestPipeline: @pytest.fixture def test_dir(self): - path = "./tmp_test" - os.makedirs(path, exist_ok=True) + path = tempfile.mkdtemp() yield path rmtree(path) @@ -552,6 +558,362 @@ def test_spark_pipeline( diff = transformed_df.exceptAll(request.getfixturevalue(expected_dataframe)) assert diff.isEmpty(), f"PipelineKeras output is not the same as expected." + @pytest.mark.parametrize( + "stages", + [ + "valid_stages_1", + "valid_stages_2", + ], + ) + def test_spark_pipeline_checkpoint_is_transparent( + self, stages, example_dataframe, request + ): + """ + checkpoint(eager=True) only truncates lineage, so fitting with a positive + checkpointInterval must yield results identical to the default (None). + """ + stages = request.getfixturevalue(stages) + + baseline_model = KamaeSparkPipeline(stages=stages, checkpointInterval=None).fit( + example_dataframe + ) + checkpointed_model = KamaeSparkPipeline( + stages=stages, checkpointInterval=2 + ).fit(example_dataframe) + + baseline_df = baseline_model.transform(example_dataframe) + checkpointed_df = checkpointed_model.transform(example_dataframe) + + assert baseline_df.schema == checkpointed_df.schema + assert baseline_df.exceptAll(checkpointed_df).isEmpty() + assert checkpointed_df.exceptAll(baseline_df).isEmpty() + + def test_spark_pipeline_checkpoint_invocation( + self, valid_stages_1, example_dataframe + ): + """ + checkpoint must be invoked during fit only when checkpointInterval > 0. + """ + original_checkpoint = DataFrame.checkpoint + + with patch.object( + DataFrame, + "checkpoint", + autospec=True, + side_effect=original_checkpoint, + ) as mock_checkpoint: + KamaeSparkPipeline(stages=valid_stages_1, checkpointInterval=None).fit( + example_dataframe + ) + assert mock_checkpoint.call_count == 0 + + mock_checkpoint.reset_mock() + KamaeSparkPipeline(stages=valid_stages_1, checkpointInterval=2).fit( + example_dataframe + ) + assert mock_checkpoint.call_count > 0 + + @pytest.mark.parametrize("bad_value", [0, -1, -5]) + def test_spark_pipeline_checkpoint_interval_rejects_non_positive(self, bad_value): + """ + checkpointInterval must be a positive integer or None; 0 and negatives raise. + """ + with pytest.raises(ValueError): + KamaeSparkPipeline(checkpointInterval=bad_value) + + def test_spark_pipeline_checkpoint_bounds_plan_depth(self, spark_session): + """ + The point of checkpointInterval is to bound logical-plan depth. We build a + deep, linearly-dependent pipeline (every stage is an ancestor of the next, so + the working DataFrame is advanced at every fit and lineage keeps growing) and + capture the logical-plan size of the DataFrame handed to each estimator fit. + With a positive interval the plan must stay markedly smaller than the default. + """ + df = spark_session.createDataFrame( + [(1.0,), (4.0,), (7.0,), (2.0,), (9.0,)], + ["col0"], + ) + + num_blocks = 4 + transforms_per_block = 4 + + def build_stages(): + stages = [] + prev = "col0" + for b in range(num_blocks): + for t in range(transforms_per_block): + out = f"t_{b}_{t}" + stages.append( + SubtractTransformer( + inputCol=prev, outputCol=out, mathFloatConstant=1.0 + ) + ) + prev = out + out = f"s_{b}" + stages.append(StandardScaleEstimator(inputCol=prev, outputCol=out)) + prev = out + return stages + + def max_fit_plan_length(interval): + plan_lengths = [] + original_fit = StandardScaleEstimator.fit + + def spy_fit(estimator, dataset, *args, **kwargs): + plan = dataset._jdf.queryExecution().logical().toString() + plan_lengths.append(len(plan)) + return original_fit(estimator, dataset, *args, **kwargs) + + with patch.object(StandardScaleEstimator, "fit", spy_fit): + KamaeSparkPipeline( + stages=build_stages(), checkpointInterval=interval + ).fit(df) + return max(plan_lengths) + + baseline_max = max_fit_plan_length(None) + checkpointed_max = max_fit_plan_length(transforms_per_block + 1) + + # Checkpointing must keep the deepest fit-time plan well below the un-bounded + # baseline. A strict 2x margin is robust to Spark-version plan-string changes. + assert checkpointed_max * 2 < baseline_max, ( + f"plan not bounded: baseline_max={baseline_max}, " + f"checkpointed_max={checkpointed_max}" + ) + + def test_spark_pipeline_prunes_unused_input_columns( + self, valid_stages_1, example_dataframe + ): + """ + prune_unused_input_columns must keep the columns the pipeline reads + (col1/col2/col3 via ArrayConcatenate, col4 via StringIndex) and drop the + unused col5 and col1_col2_col3, while preserving every row. + """ + pipeline = KamaeSparkPipeline(stages=valid_stages_1) + + # The required set is generous (a superset), so assert containment rather + # than equality - it also carries output/param strings that harmlessly do + # not match any input-DataFrame column. + required = pipeline.collect_required_input_columns(valid_stages_1) + assert {"col1", "col2", "col3", "col4"}.issubset(required) + + pruned = pipeline.prune_unused_input_columns(example_dataframe, valid_stages_1) + + assert pruned.columns == ["col1", "col2", "col3", "col4"] + assert pruned.exceptAll( + example_dataframe.select("col1", "col2", "col3", "col4") + ).isEmpty() + + def test_collect_required_input_columns_includes_aux_columns(self): + """ + Aux columns read at fit time via params other than inputCol(s) - here + maskCols and relevanceCol on ConditionalStandardScaleEstimator, and + queryIdCol on a listwise transformer - must be reported by the collector so + pruning does not drop them. No Spark session needed. + """ + stages = [ + ConditionalStandardScaleEstimator( + inputCol="x", + outputCol="x_scaled", + maskCols=["m"], + relevanceCol="r", + ), + ListMeanTransformer( + inputCol="p", + outputCol="p_list_mean", + queryIdCol="q", + ), + ] + + required = KamaeSparkPipeline.collect_required_input_columns(stages) + + assert {"x", "m", "r", "p", "q"} <= required + + def test_collect_required_input_columns_plain_estimator(self): + """ + A stage with no aux column params must still report its inputCol and must + not gain spurious columns - confirms the aux sweep does not regress the + simple case. + """ + stages = [StandardScaleEstimator(inputCol="x", outputCol="x_scaled")] + + required = KamaeSparkPipeline.collect_required_input_columns(stages) + + assert "x" in required + + def test_spark_pipeline_prune_input_columns_is_opt_in( + self, valid_stages_1, example_dataframe + ): + """ + Pruning must only happen when pruneInputColumns is True. With the default + (False) the input DataFrame is not projected during fit. + """ + with patch.object( + KamaeSparkPipeline, + "prune_unused_input_columns", + autospec=True, + side_effect=KamaeSparkPipeline.prune_unused_input_columns, + ) as mock_prune: + KamaeSparkPipeline(stages=valid_stages_1).fit(example_dataframe) + assert mock_prune.call_count == 0 + + mock_prune.reset_mock() + KamaeSparkPipeline(stages=valid_stages_1, pruneInputColumns=True).fit( + example_dataframe + ) + assert mock_prune.call_count == 1 + + def test_spark_pipeline_prune_keeps_aux_fit_columns(self, spark_session): + """ + Regression: pruning must not drop columns an estimator reads at fit time + through params other than inputCol (here maskCols). The fit must not raise, + the genuinely-unused column must be pruned, and the fitted moments must be + identical to a prune-disabled baseline (numerically transparent). + """ + df = spark_session.createDataFrame( + [(1.0, 1, 3.0, 99.0), (2.0, 0, 1.0, 99.0), (3.0, 1, 2.0, 99.0)], + ["x", "m", "r", "junk"], + ) + + def build_pipeline(prune): + return KamaeSparkPipeline( + stages=[ + ConditionalStandardScaleEstimator( + inputCol="x", + outputCol="x_scaled", + maskCols=["m"], + maskOperators=["eq"], + maskValues=[1.0], + relevanceCol="r", + ), + ], + pruneInputColumns=prune, + ) + + pruned_pipeline = build_pipeline(prune=True) + + # Aux fit columns kept, genuinely-unused column dropped. + required = pruned_pipeline.collect_required_input_columns( + pruned_pipeline.getStages() + ) + assert {"x", "m", "r"} <= required + assert "junk" not in required + + # Must NOT raise UNRESOLVED_COLUMN / "Mask column m not found". + pruned_model = pruned_pipeline.fit(df) + baseline_model = build_pipeline(prune=False).fit(df) + + pruned_scaler = pruned_model.stages[-1] + baseline_scaler = baseline_model.stages[-1] + + assert pruned_scaler.getMean() == baseline_scaler.getMean() + assert pruned_scaler.getStddev() == baseline_scaler.getStddev() + + def test_spark_pipeline_prune_is_transparent_to_fit( + self, valid_stages_1, example_dataframe + ): + """ + Pruning drops only columns no stage reads, so a fitted model - and its + transform output - must be identical whether or not the input carries an + extra unused column when pruneInputColumns is enabled. + """ + with_extra = example_dataframe.withColumn( + "unused", example_dataframe["col1"] * 100.0 + ) + + baseline_out = ( + KamaeSparkPipeline(stages=valid_stages_1, pruneInputColumns=True) + .fit(example_dataframe) + .transform(example_dataframe) + ) + with_extra_out = ( + KamaeSparkPipeline(stages=valid_stages_1, pruneInputColumns=True) + .fit(with_extra) + .transform(example_dataframe) + ) + + assert baseline_out.schema == with_extra_out.schema + assert baseline_out.exceptAll(with_extra_out).isEmpty() + assert with_extra_out.exceptAll(baseline_out).isEmpty() + + def test_spark_pipeline_cache_estimator_input_is_transparent_to_fit( + self, spark_session + ): + """ + cacheEstimatorInput projects to still-needed columns and persists that + narrow frame once; independent sibling estimators must fit to byte-identical + params whether it is on or off, with the genuinely-unused column dropped. + """ + df = spark_session.createDataFrame( + [ + (1.0, 2.0, 3.0, 99.0), + (2.0, 4.0, 6.0, 99.0), + (3.0, 6.0, 9.0, 99.0), + (4.0, 8.0, 12.0, 99.0), + ], + ["x1", "x2", "x3", "junk"], + ) + + def build_pipeline(cache): + return KamaeSparkPipeline( + stages=[ + StandardScaleEstimator(inputCol="x1", outputCol="x1_scaled"), + StandardScaleEstimator(inputCol="x2", outputCol="x2_scaled"), + StandardScaleEstimator(inputCol="x3", outputCol="x3_scaled"), + ], + cacheEstimatorInput=cache, + ) + + cached_model = build_pipeline(cache=True).fit(df) + baseline_model = build_pipeline(cache=False).fit(df) + + for cached_scaler, baseline_scaler in zip( + cached_model.stages, baseline_model.stages + ): + assert cached_scaler.getMean() == baseline_scaler.getMean() + assert cached_scaler.getStddev() == baseline_scaler.getStddev() + + def test_spark_pipeline_cache_estimator_input_is_opt_in(self, spark_session): + """ + The narrow-cache projection (which computes the live keep-set via + collect_required_input_columns) must only run when cacheEstimatorInput is + True. Pruning is left off so the collector is not called for that reason. + """ + df = spark_session.createDataFrame( + [(1.0, 2.0, 99.0), (2.0, 4.0, 99.0), (3.0, 6.0, 99.0)], + ["x1", "x2", "junk"], + ) + stages = [ + StandardScaleEstimator(inputCol="x1", outputCol="x1_scaled"), + StandardScaleEstimator(inputCol="x2", outputCol="x2_scaled"), + ] + + with patch.object( + KamaeSparkPipeline, + "collect_required_input_columns", + wraps=KamaeSparkPipeline.collect_required_input_columns, + ) as mock_collect: + KamaeSparkPipeline(stages=stages).fit(df) + assert mock_collect.call_count == 0 + + mock_collect.reset_mock() + KamaeSparkPipeline(stages=stages, cacheEstimatorInput=True).fit(df) + assert mock_collect.call_count == 1 + + def test_spark_pipeline_cache_estimator_input_mutually_exclusive( + self, valid_stages_1, example_dataframe + ): + """ + cacheEstimatorInput and cacheIntermediateData are competing strategies; + enabling both warns, prefers cacheEstimatorInput, and still fits. + """ + pipeline = KamaeSparkPipeline( + stages=valid_stages_1, + cacheIntermediateData=True, + cacheEstimatorInput=True, + ) + with pytest.warns(UserWarning, match="takes precedence"): + pipeline_model = pipeline.fit(example_dataframe) + assert pipeline_model is not None + @pytest.mark.parametrize( "stages, input_col, original_dtype", [ diff --git a/uv.lock b/uv.lock index 68ac512a..ab5c5c8d 100644 --- a/uv.lock +++ b/uv.lock @@ -873,6 +873,7 @@ dependencies = [ { name = "numpy", version = "1.26.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, { name = "pandas", version = "1.5.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.12'" }, { name = "pandas", version = "2.3.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.12'" }, + { name = "pyarrow" }, { name = "pyfarmhash" }, { name = "pyspark" }, { name = "tensorflow" }, @@ -922,6 +923,7 @@ requires-dist = [ { name = "networkx", specifier = ">=2.6.3,<3.0.0" }, { name = "numpy", specifier = ">=1.22.0,<2.0.0" }, { name = "pandas", specifier = ">=1.3.4,<3.0.0" }, + { name = "pyarrow", specifier = ">=4.0.0" }, { name = "pyfarmhash", specifier = ">=0.3.2,<0.4.0" }, { name = "pyspark", specifier = ">=3.4.0,<4.0.0" }, { name = "tensorflow", specifier = ">=2.16.0,<3.0.0" }, @@ -1959,6 +1961,35 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/10/30/a58b32568f1623aaad7db22aa9eafc4c6c194b429ff35bdc55ca2726da47/py4j-0.10.9.7-py2.py3-none-any.whl", hash = "sha256:85defdfd2b2376eb3abf5ca6474b51ab7e0de341c75a02f46dc9b5976f5a5c1b", size = 200481 }, ] +[[package]] +name = "pyarrow" +version = "25.0.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/3d/e3/27f57f80141379d60defe6703eb50a707325706f07fedfd1312c7a751995/pyarrow-25.0.1.tar.gz", hash = "sha256:9150a83248bfed9813ea3c3af74c3856c1984d444aa28e58bf7733b9750ddf6a", size = 1201653 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/0a/3e/5cd70becb51e1d044c54ba5e627424a6e87df5b98008cbd22cc6abd409ca/pyarrow-25.0.1-cp310-cp310-macosx_12_0_arm64.whl", hash = "sha256:0b1edbb2f385a6a65e9711b62ba86ac54a7816a3f8d17bb3e8a5929d65fb2485", size = 35954271 }, + { url = "https://files.pythonhosted.org/packages/64/be/17599e086df264ea7dc221d1101e3131e181e00da428a2f9bd0358f0d06b/pyarrow-25.0.1-cp310-cp310-macosx_12_0_x86_64.whl", hash = "sha256:a4dd8bf99a8fac133efc0ed6a92f5fddbe2adba0d0f6dd720e39ba9855cea85c", size = 37647543 }, + { url = "https://files.pythonhosted.org/packages/42/34/e138b451fd3970a6eda4599f68ae3b2b32b661bc958de3239d54a0bf6575/pyarrow-25.0.1-cp310-cp310-manylinux_2_28_aarch64.whl", hash = "sha256:bddd0c4f7630c2a3ddf6347c1bdaa79d97bcf6bd445f9e60c816b7d77c85a5ae", size = 46837120 }, + { url = "https://files.pythonhosted.org/packages/57/5c/f8fc0eb2de03464a557d5a4d0c15e972d73362414696618833b771f7eddd/pyarrow-25.0.1-cp310-cp310-manylinux_2_28_x86_64.whl", hash = "sha256:a4d6d5e9a3d1879a97c08ded0c797579b7965eafd0f0c26c30b45ccc06db939b", size = 50066460 }, + { url = "https://files.pythonhosted.org/packages/3f/d1/0dd64fd06de0333b808a02f60981635f067b71aad3a30698a9a104fae778/pyarrow-25.0.1-cp310-cp310-musllinux_1_2_aarch64.whl", hash = "sha256:514ddb60285631af068875550c90eddc181db3e8e63a032b1559be189e82f056", size = 49937892 }, + { url = "https://files.pythonhosted.org/packages/cb/3c/f89d1bd76d5f3284c2a44d7d7ebbd8204535e5ae2b41f4077069b4ff2ec6/pyarrow-25.0.1-cp310-cp310-musllinux_1_2_x86_64.whl", hash = "sha256:cab40b1edfef0262e0e5251aa2c58d75630f24d06dd7794480243acc001a1d7d", size = 53107240 }, + { url = "https://files.pythonhosted.org/packages/67/67/b554a8e09f3f3decccf405eb8fbe86696321cbcb5b62d18b4a5057a4c113/pyarrow-25.0.1-cp310-cp310-win_amd64.whl", hash = "sha256:60e89d8f13861a1f7f8d950fa54aebb8023b30734d0ac51ffa80beabe2df4bba", size = 27848683 }, + { url = "https://files.pythonhosted.org/packages/ee/8b/0d23b47702fcfe8b3618d5292035099675c5a1c48258932350c08020f7b5/pyarrow-25.0.1-cp311-cp311-macosx_12_0_arm64.whl", hash = "sha256:51093dd9e10325fbdb3c10a2ae7c4806e5c822d94e74ae4938b26524a3323fee", size = 35946180 }, + { url = "https://files.pythonhosted.org/packages/d8/17/707d17a5476c55a9541fde0db8213ac30979a792864d72415f176ba50c45/pyarrow-25.0.1-cp311-cp311-macosx_12_0_x86_64.whl", hash = "sha256:eb6203482ff3746a5632303a7279ae0b5a304c46985b49ed1378cb350ea6728d", size = 37644787 }, + { url = "https://files.pythonhosted.org/packages/c1/b2/cdc98ecf1a6408280bc3a6a07054cdd99a3f4670acc0545d383ce113e87d/pyarrow-25.0.1-cp311-cp311-manylinux_2_28_aarch64.whl", hash = "sha256:880523be3d29efcf83d3998835d206118ccf35e3871dbd2fb60408cf6b007a80", size = 46834633 }, + { url = "https://files.pythonhosted.org/packages/c8/6e/d3fafc41f378b2c65be43b827798c0fae42049a641c8526633ed3eb573e2/pyarrow-25.0.1-cp311-cp311-manylinux_2_28_x86_64.whl", hash = "sha256:25f8720bf6387d5dc2ebd2622112de630760419e4b66134405dd24110d15f37e", size = 50065507 }, + { url = "https://files.pythonhosted.org/packages/d5/12/8d0698954b8c3001844a898e0a6900bebe83d7ee40c11195174c5122f324/pyarrow-25.0.1-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:4facd65742a024a4a366328a1d2292062d72d6e023c1b7dda8d4c37544933a25", size = 49955690 }, + { url = "https://files.pythonhosted.org/packages/d3/0b/1ecb936ac6409e90a34d58eea1c7cec09a9ae6d2141b9e49ad01a2b1ea47/pyarrow-25.0.1-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:aa0559502e1cd6254d6814614085dd9c5a3dd0419362978a936a3f68a9e5c3df", size = 53128198 }, + { url = "https://files.pythonhosted.org/packages/8e/1c/5236033550633c9b7377b2a53660b2bbb06cb06dc09c4356332d67643ca1/pyarrow-25.0.1-cp311-cp311-win_amd64.whl", hash = "sha256:62cd0d785b8aa6675ee355f9fc02252a340f4441257c42674937826fd7594325", size = 27857263 }, + { url = "https://files.pythonhosted.org/packages/a6/e2/9ab15b88cbfac28e16419ce5439ec29234c5172cb8259301b4ba639bdec0/pyarrow-25.0.1-cp312-cp312-macosx_12_0_arm64.whl", hash = "sha256:df961f2e7ae9cf496459259d798652c70625f6c080650d6952f8c04053c58ee9", size = 35861559 }, + { url = "https://files.pythonhosted.org/packages/58/79/a0036dbe1eabe1f73127427342f1d99982584c4a2cde2651d6c93499c6f6/pyarrow-25.0.1-cp312-cp312-macosx_12_0_x86_64.whl", hash = "sha256:cc4aa407fde9fc660be3939e49ea31f50f3e9fec17c0ec63159f7711edd3efc9", size = 37628383 }, + { url = "https://files.pythonhosted.org/packages/13/49/d93a57d375f4bf0cf82913dd6bb54acafde83dd993be2282c81ac5616cad/pyarrow-25.0.1-cp312-cp312-manylinux_2_28_aarch64.whl", hash = "sha256:4340f0ba6c1d2e13f21658de1d7c662ca2545018568d0030a1e9afca159d87e3", size = 46820190 }, + { url = "https://files.pythonhosted.org/packages/60/c9/711ca85d79f1ec98f29a5eae2b051e25b4ecec5de3e3c0e2d5c5dcb15664/pyarrow-25.0.1-cp312-cp312-manylinux_2_28_x86_64.whl", hash = "sha256:5389cdf79447ed1515c9e31620e6e1e2302249564d603f2ad727d4f6d313e4c3", size = 50102437 }, + { url = "https://files.pythonhosted.org/packages/80/53/8fb8359ff17cfb6263a1cf3ebf7caec9fe197de118719e84fcb1d0618026/pyarrow-25.0.1-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:d51592cb7561e87877c506113e7adbf1342ab579e6c21f0ef44b8ba41cb74c80", size = 49942424 }, + { url = "https://files.pythonhosted.org/packages/e8/83/4e5ae02a9341571b18a6fca380ac7a58ce6ddae7ab3c060208c0a1e79f02/pyarrow-25.0.1-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:6109c94d8b9f3b17a041daca16cacb2f651ad8f1ef70a4232c2c0f37a23da2a8", size = 53144206 }, + { url = "https://files.pythonhosted.org/packages/65/ee/197cbf47e49f83e6ebeb946a5259a48a638dea27ac774db42fe78022179d/pyarrow-25.0.1-cp312-cp312-win_amd64.whl", hash = "sha256:8858d7bfc22e3f51529aeaa4077225029724623e4595dc9eff8c793935c34140", size = 27953934 }, +] + [[package]] name = "pycodestyle" version = "2.12.1"