&`ijddlZddlZddlZddlZddlZddlZddlmZddlm Z m Z m Z m Z m Z mZmZddlmZddlmZmZe rddlZddlZddlmZddlmZejeZed Gd d e Z!ed Gd d ej"Z#eGdde#ej"Z$e$j%Z%dS)N)Enum) TYPE_CHECKINGAnyDictListOptionalUnionfinal) BatchFormat) DeveloperAPI PublicAPI) DataBatchType)Datasetbeta) stabilityceZdZdZdS)PreprocessorNotFittedExceptionzddlm}||_i|_dSNrStatComputationPlan)ray.data.preprocessors.utilsr stat_computation_planstats_)selfr s r__init__zPreprocessor.__init__6s0DDDDDD%8%8%:%:" rc"eZdZdZdZdZdZdZdS)Preprocessor.FitStatuszThe fit status of preprocessor. NOT_FITTABLE NOT_FITTEDPARTIALLY_FITTEDFITTEDN)rrrrr(r)r*r+rrr FitStatusr'<s+--% ! .rr,TcXfdtD}t|S)aGChecks if the Preprocessor has fitted state. This is also used as an indiciation if the Preprocessor has been fit, following convention from Ray versions prior to 2.6. This allows preprocessors that have been fit in older versions of Ray to be used to transform data in newer versions. c^g|])}|dt|'|*S)_)endswithgetattr).0vr$s r z8Preprocessor._check_has_fitted_state..Us8UUUQ 3UGDRSDTDTUqUUUr)varsbool)r$ fitted_varss` r_check_has_fitted_statez$Preprocessor._check_has_fitted_stateLs3VUUU$t**UUU K   rreturnr'c|jstjjSt |dr|js|rtjjStjjS)N_fitted) _is_fittablerr,r(hasattrr;r8r+r)r$s r fit_statuszPreprocessor.fit_statusXsc  5)6 6 D) $ $ 5)- 5  ) ) + + 5 )0 0)4 4rdsrc\|}|tjjkr|S|tjjtjjfvrt jd|j | | |}d|_ |S)aFit this Preprocessor to the Dataset. Fitted state attributes will be directly set in the Preprocessor. Calling it more than once will overwrite all previously fitted state: ``preprocessor.fit(A).fit(B)`` is equivalent to ``preprocessor.fit(B)``. Args: ds: Input dataset. Returns: Preprocessor: The fitted Preprocessor with state attributes. z`fit` has already been called on the preprocessor (or at least one contained preprocessors if this is a chain). All previously fitted state will be overwritten!T) r?rr,r(r+r*warningswarnr"reset_fit _fit_executer;)r$r@r? fitted_dss rfitzPreprocessor.fitbs__&& /< < <K   " )  " 3    MC    "((***IIbMM..r22  rdatasetcV|xj|j|zc_|S)N)r#r"compute)r$rIs rrFzPreprocessor._fit_executes( t199'BBB  rcRt|dot|jdkS)Nr#r)r=lenr#r>s r has_statszPreprocessor.has_statss&tX&&?3t{+;+;a+??rN)transform_num_cpustransform_memorytransform_batch_sizetransform_concurrencyrOrPrQrRc`||||||||S)a$Fit this Preprocessor to the Dataset and then transform the Dataset. Calling it more than once will overwrite all previously fitted state: ``preprocessor.fit_transform(A).fit_transform(B)`` is equivalent to ``preprocessor.fit_transform(B)``. Args: ds: Input Dataset. transform_num_cpus: [experimental] The number of CPUs to reserve for each parallel map worker. transform_memory: [experimental] The heap memory in bytes to reserve for each parallel map worker. transform_batch_size: [experimental] The maximum number of rows to return. transform_concurrency: [experimental] The maximum number of Ray workers to use concurrently. Returns: ray.data.Dataset: The transformed Dataset. )num_cpusmemory batch_size concurrency)rH transform)r$r@rOrPrQrRs r fit_transformzPreprocessor.fit_transforms=2  ~~ '#+-    rrVrTrUrWrVrTrUrWc|}|tjjtjjfvrt d||||||}|S)aTransform the given dataset. Args: ds: Input Dataset. batch_size: [experimental] Advanced configuration for adjusting input size for each worker. num_cpus: [experimental] The number of CPUs to reserve for each parallel map worker. memory: [experimental] The heap memory in bytes to reserve for each parallel map worker. concurrency: [experimental] The maximum number of Ray workers to use concurrently. Returns: ray.data.Dataset: The transformed Dataset. Raises: PreprocessorNotFittedException: if ``fit`` is not called yet. zX`fit` must be called before `transform`, or simply use fit_transform() to run both stepsrZ)r?rr,r*r)r _transform)r$r@rVrTrUrWr?transformed_dss rrXzPreprocessor.transforms0__&&   " 3  " -   1B  !# )  rdatarc|}|tjjtjjfvrt d||S)aTransform a single batch of data. The data will be converted to the format supported by the Preprocessor, based on which ``_transform_*`` methods are implemented. Args: data: Input data batch. Returns: DataBatchType: The transformed data batch. This may differ from the input type depending on which ``_transform_*`` methods are implemented. z.`fit` must be called before `transform_batch`.)r?rr,r*r)r_transform_batch)r$r^r?s rtransform_batchzPreprocessor.transform_batchsd__&&   " 3  " -   1@ $$T***rct)z2Sub-classes should override this instead of fit().NotImplementedError)r$r@s rrEzPreprocessor._fit"###rc|jjtjk}|jjtjk}|r|r|S|r t jS|r t jStd)afDetermine which batch format to use based on Preprocessor implementation. * If only `_transform_pandas` is implemented, then use ``pandas`` batch format. * If only `_transform_numpy` is implemented, then use ``numpy`` batch format. * If both are implemented, then use the Preprocessor defined preferred batch format. zNone of `_transform_numpy` or `_transform_pandas` are implemented. At least one of these transform functions must be implemented for Preprocessor transforms.) __class___transform_pandasr_transform_numpypreferred_batch_formatr NUMPYPANDASrd)r$has_transform_pandashas_transform_numpys r_determine_transform_to_usez(Preprocessor._determine_transform_to_uses N , 0N N  N +|/L L   #7 ..00 0 $ $ ! % %%/ rcj|}|}|||d<|||d<|||d<|||d<|tjkr|j|jfdtji|S|tjkr|j|jfdtji|Std|)NrTrUrVrW batch_formatziInvalid transform type returned from _determine_transform_to_use; "pandas" and "numpy" allowed, but got: ) ro_get_transform_configr rl map_batchesrhrkri ValueError)r$r@rVrTrUrWtransform_typekwargss rr\zPreprocessor._transforms99;;++--  !)F:   %F8   !#-F<  "$/F= ! [/ / /!2>&5@5GKQ {0 0 0!2>%4?4EIO K:HKK rciS)zReturns kwargs to be passed to :meth:`ray.data.Dataset.map_batches`. This can be implemented by subclassing preprocessors. rr>s rrrz"Preprocessor._get_transform_config6s  rcddl}ddl}ddlm}m} ddl}n#t $rd}YnwxYwt||j|j tj j |j fs tdt|d|}|t"jkr|||S|t"jkr|||SdS)Nr)_convert_batch_type_to_numpy_convert_batch_type_to_pandasz`transform_batch` is currently only implemented for Pandas DataFrames, pyarrow Tables, NumPy ndarray and dictionary of ndarray. Got .)numpypandas"ray.air.util.data_batch_conversionryrzpyarrow ImportError isinstance DataFrameTable collectionsabcMappingndarrayrttyperor rlrhrkri)r$r^nppdryrzrrus rr`zPreprocessor._transform_batch=sK          NNNN   GGG  2< 0GT   . $T ...  99;; [/ / /))*G*G*M*MNN N {0 0 0(()E)Ed)K)KLL L1 0s  &&columnsoutput_columnscl|r/t|t|krtd|p|S)a(Returns the output columns after validation. Checks if the columns are explicitly set, otherwise defaulting to the input columns. Raises: ValueError: If the length of the output columns does not match the length of the input columns. zuInvalid output_columns: Got len(columns) != len(output_columns). The length of columns and output_columns must match.)rMrt)clsrrs r#_derive_and_validate_output_columnsz0Preprocessor._derive_and_validate_output_columns\sI  c'llc..A.AAAG ((rdf pd.DataFramect)zDRun the transformation on a data batch in a Pandas DataFrame format.rc)r$rs rrhzPreprocessor._transform_pandasqrernp_dataz np.ndarrayct)zARun the transformation on a data batch in a NumPy ndarray format.rc)r$rs rrizPreprocessor._transform_numpyvs "###rctjS)aPBatch format hint for upstream producers to try yielding best block format. The preferred batch format to use if both `_transform_pandas` and `_transform_numpy` are implemented. Defaults to Pandas. Can be overriden by Preprocessor classes depending on which transform path is the most optimal. )r rlrs rrjz#Preprocessor.preferred_batch_format}s !!rc$t|dgS)Nrr1r>s rget_input_columnszPreprocessor.get_input_columnsstY+++rc$t|dgS)Nrrr>s rget_output_columnszPreprocessor.get_output_columnsst-r222rcd|j}|dd|S)Nr")__dict__copypop)r$states r __getstate__zPreprocessor.__getstate__s/ ""$$ )4000 rrcdddlm}|j|||_dSr)r!r rupdater")r$rr s r __setstate__zPreprocessor.__setstate__sADDDDDD U###%8%8%:%:"""rcttjtj|dS)zReturn this preprocessor serialized as a string. Note: This is not a stable serialization format as it uses `pickle`. ascii)base64 b64encodepickledumpsdecoder>s r serializezPreprocessor.serializes- T 2 233::7CCCr serializedcNtjtj|S)zALoad the original preprocessor serialized via `self.serialize()`.)rloadsr b64decode)rs r deserializezPreprocessor.deserializes|F,Z88999r)r9r')r@rr9r)rIr)r^rr9r)NNN)rrr9r),rrrrr%strrr,r<r8r?rHrFr6rNrfloatintrYrXrar rEr ror\rrrrr` classmethodrrrhr rirjrrrrr staticmethodrrrrrrs*     C   L ! ! !5555    D@4@@@@/3,0.2/3      %UO  #5/  'sm   (}        L%)$("&%)((( (SM ( 5/ (  (c]( ((((T++++2$$$\$[B%)"&%) "" "SM"5/ "  " c] " """"HtCH~MMMM>)3i)19$s)1D) c)))[)($$$\$$\4\0A+BBC$ |T#|"344 5$$$\$  "{ " " "\[ ",49,,,,3DI3333d38n ;$sCx.;;;; D3DDD\D:::::\\:::rrceZdZdZeGddeZdZdZe j de e e ffdZe j de e e fd efd Zde e e ffd Zd e e e ffd Zede fdZede ddfdZedefdZed eddfdZeedee effdZeeedee efddfdZdS)SerializablePreprocessorBaseaAbstract base class for serializable preprocessors. This class defines the serialization interface that all preprocessors must implement to support saving and loading their state. The serialization system uses CloudPickle as the primary format. **Architecture Overview:** The serialization system is built around two types of methods: 1. **Final Methods (DO NOT OVERRIDE):** - ``serialize()``: Orchestrates the serialization process - ``deserialize()``: Orchestrates the deserialization process These methods are marked as ``@final`` and should never be overridden by subclasses. They handle format detection, factory coordination, and error handling. 2. **Abstract Methods (MUST IMPLEMENT):** - ``_get_serializable_fields()``: Extract instance fields for serialization - ``_set_serializable_fields()``: Restore instance fields from deserialization - ``_get_stats()``: Extract computed statistics for serialization - ``_set_stats()``: Restore computed statistics from deserialization These methods must be implemented by each preprocessor subclass to define their specific serialization behavior. **Format Support:** - **CloudPickle** (default): - **Pickle** (legacy): Backward compatibility for existing serialized data **Important Notes:** - Never override ``serialize()`` or ``deserialize()`` in subclasses - Always call ``super().__init__()`` in subclass constructors - Use ``_fitted`` attribute to track fitting state - Store computed statistics in ``stats_`` dictionary - Handle version migration and backwards compatibility in ``_set_serializable_fields()`` if needed ceZdZdZdZdS)0SerializablePreprocessorBase.SerializationFormat cloudpicklerN)rrr CLOUDPICKLEPICKLErrrSerializationFormatrs# rrsCPKL:r9cdS)azExtract instance fields that should be serialized. This method should return a dictionary containing all instance attributes that are necessary to restore the preprocessor's configuration state. This typically includes constructor parameters and internal state flags. Returns: Dictionary mapping field names to their values Nrr>s r_get_serializable_fieldsz5SerializablePreprocessorBase._get_serializable_fieldss  rfieldsversioncdS)aRestore instance fields from deserialized data. This method should restore the preprocessor's configuration state from the provided fields' dictionary. It's called during deserialization to recreate the instance state. **Version Migration:** If the serialized version differs from the current ``VERSION``, implement migration logic to handle schema changes: .. testcode:: def _set_serializable_fields(self, fields: Dict[str, Any], version: int): # Handle version migration if version == 1 and self.VERSION == 2: # Migrate from version 1 to 2 if "old_field" in fields: fields["new_field"] = migrate_old_field(fields.pop("old_field")) # Set all fields for key, value in fields.items(): setattr(self, key, value) # Reinitialize derived state self.stat_computation_plan = StatComputationPlan() Args: fields: Dictionary of field names to values version: Version of the serialized data Nr)r$rrs r_set_serializable_fieldsz5SerializablePreprocessorBase._set_serializable_fieldss B rc$t|diS)aExtract computed statistics that should be serialized. This method should return the computed statistics that were generated during the ``fit()`` process. These statistics are typically stored in the ``stats_`` attribute and contain the learned parameters needed for transformation. Returns: Dictionary containing computed statistics r#rr>s r _get_statsz'SerializablePreprocessorBase._get_statsstXr***rstatsc||_dS)avRestore computed statistics from deserialized data. This method should restore the preprocessor's computed statistics from the provided stats dictionary. These statistics are typically stored in the ``stats_`` attribute and contain learned parameters from fitting. Args: stats: Dictionary containing computed statistics N)r#)r$rs r _set_statsz'SerializablePreprocessorBase._set_statss rc|jS)zGet the preprocessor class identifier for this preprocessor class. Returns: The preprocessor class identifier string used to identify this preprocessor type in serialized data. 4_SerializablePreprocessorBase__PREPROCESSOR_CLASS_IDrs rget_preprocessor_class_idz6SerializablePreprocessorBase.get_preprocessor_class_id's **r identifierNc||_dS)zSet the preprocessor class identifier for this preprocessor class. Args: identifier: The preprocessor class identifier string to use. Nr)rrs rset_preprocessor_class_idz6SerializablePreprocessorBase.set_preprocessor_class_id1s'1###rc|jS)zGet the version number for this preprocessor class. Returns: The version number for this preprocessor's serialization format. &_SerializablePreprocessorBase__VERSIONrs r get_versionz(SerializablePreprocessorBase.get_version:s }rc||_dS)zSet the version number for this preprocessor class. Args: version: The version number for this preprocessor's serialization format. Nr)rrs r set_versionz(SerializablePreprocessorBase.set_versionCs  rcddlm}m}|||||jd}||j  |S)u[Serialize this preprocessor to a string or bytes. **⚠️ DO NOT OVERRIDE THIS METHOD IN SUBCLASSES ⚠️** This method is marked as ``@final`` in the concrete implementation and handles the complete serialization orchestration. Subclasses should implement the abstract methods instead: ``_get_serializable_fields()`` and ``_get_stats()``. **Serialization Process:** 1. Extracts fields via ``_get_serializable_fields()`` 2. Extracts statistics via ``_get_stats()`` 3. Packages data with metadata (type, version, format) 4. Delegates to ``SerializationHandlerFactory`` for format-specific handling 5. Returns serialized data with magic bytes for format identification **Supported Formats:** - **CloudPickle** (default): - **Pickle** (legacy): Backward compatibility for existing serialized data Returns: Serialized preprocessor data (bytes for CloudPickle, str for legacy Pickle) Raises: ValueError: If the serialization format is invalid or unsupported r)HandlerFormatNameSerializationHandlerFactory)rrrrserializer_format_version)format_identifier) -ray.data.preprocessors.serialization_handlersrrrrrrSERIALIZER_FORMAT_VERSION get_handlerrr)r$rrr^s rrz&SerializablePreprocessorBase.serializeLs<        2244''))3355__&&*.)G  +66/;7  )D// rrrcNddlm}m}ddlm}m} ||}||}t||r|S||d}|d|j krtd|d| |}ddl m } | |_||d |d  ||d  |S#|$rt"$r!} td|ddd| d} ~ wwxYw)u%Deserialize a preprocessor from serialized data. **⚠️ DO NOT OVERRIDE THIS METHOD IN SUBCLASSES ⚠️** This method is marked as ``@final`` in the concrete implementation and handles the complete deserialization orchestration. Subclasses should implement the abstract methods instead: ``_set_serializable_fields()`` and ``_set_stats()``. **Deserialization Process:** 1. Detects format from magic bytes in serialized data 2. Delegates to ``SerializationHandlerFactory`` for format-specific parsing 3. Extracts metadata (type, version, fields, stats) 4. Looks up preprocessor class from registry 5. Creates new instance and restores state via abstract methods 6. Returns fully reconstructed preprocessor instance **Format Detection:** The method automatically detects the serialization format: - ``CPKL:`` → CloudPickle format - Base64 string → Legacy Pickle format **Error Handling:** Provides comprehensive error handling for: - Unknown serialization formats - Corrupted or invalid data - Missing preprocessor types - Version compatibility issues Args: serialized: Serialized preprocessor data (bytes or str) Returns: Reconstructed preprocessor instance Raises: ValueError: If the serialized data is corrupted or format is unrecognized UnknownPreprocessorError: If the preprocessor type is not registered r)PickleSerializationHandlerr)UnknownPreprocessorError _lookup_class)r^rrz'Unsupported serializer format version: rrr)rrr)rz2Failed to deserialize preprocessor. Data preview: N2z...)rrr&ray.data.preprocessors.version_supportrrrrrrrt__new__r!r r"rr Exception) rrrrrhandlermetarobjr es rrz(SerializablePreprocessorBase.deserialize~sZ                $ 1==:=NNG&&z22D'#=>>   -V --C/0C4QQQ adC^>_aa++c""C I H H H H H(;(;(=(=C %  ( (XY ( X X X NNgN / / /J'       YZPSQSPS_YYY  srs  IIIIIIIIIIIIIIIIII::::::88888888)555555((((((  8 $ $ V     \    VI:I:I:I:I:37I:I:I:X kkkkk<kkk\ 3Fr