o bi8@sddlZddlZddlZddlZddlZddlmZddlmZm Z m Z m Z m Z m Z ddlmZddlmZmZerNddlZddlZddlmZddlmZedd Gd d d eZedd Gd d d ejZdS)N)Enum) TYPE_CHECKINGAnyDictListOptionalUnion) BatchFormat) DeveloperAPI PublicAPI) DataBatchType)Datasetbeta) stabilityc@seZdZdZdS)PreprocessorNotFittedExceptionzZ#dS)D PreprocessoraImplements an ML preprocessing operation. Preprocessors are stateful objects that can be fitted against a Dataset and used to transform both local data batches and distributed data. For example, a Normalization preprocessor may calculate the mean and stdev of a field during fitting, and uses these attributes to implement its normalization transform. Preprocessors can also be stateless and transform data without needed to be fitted. For example, a preprocessor may simply remove a column, which does not require any state to be fitted. If you are implementing your own Preprocessor sub-class, you should override the following: * ``_fit`` if your preprocessor is stateful. Otherwise, set ``_is_fittable=False``. * ``_transform_pandas`` and/or ``_transform_numpy`` for best performance, implement both. Otherwise, the data will be converted to the match the implemented method. c@s eZdZdZdZdZdZdZdS)Preprocessor.FitStatuszThe fit status of preprocessor. NOT_FITTABLE NOT_FITTEDPARTIALLY_FITTEDFITTEDN)rrrrrrrrrrrr FitStatus2s rTcCsddt|D}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. cSsg|] }|dr|qS)_)endswith).0vrrr Ksz8Preprocessor._check_has_fitted_state..)varsbool)self fitted_varsrrr_check_has_fitted_stateBs z$Preprocessor._check_has_fitted_statereturnrcCs6|jstjjSt|dr|js|rtjjStjjS)N_fitted) _is_fittablerrrhasattrr)r'rrr%rrr fit_statusNszPreprocessor.fit_statusdsr cCsJ|}|tjjkr |S|tjjtjjfvrtd||}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-rrrrrwarningswarn_fitr))r%r.r- fitted_dsrrrfitXs  zPreprocessor.fitN)transform_num_cpustransform_memorytransform_batch_sizetransform_concurrencyr4r5r6r7cCs|||j|||||dS)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)r3 transform)r%r.r4r5r6r7rrr fit_transformys zPreprocessor.fit_transformr:r8r9r;r:r8r9r;cCs<|}|tjjtjjfvrtd|j|||||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 stepsr>)r-rrrrr _transform)r%r.r:r8r9r;r-transformed_dsrrrr<s zPreprocessor.transformdatar cCs.|}|tjjtjjfvrtd||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-rrrrr_transform_batch)r%rAr-rrrtransform_batchs zPreprocessor.transform_batchcCt)z2Sub-classes should override this instead of fit().NotImplementedError)r%r.rrrr1zPreprocessor._fitcCsH|jjtjk}|jjtjk}|r|r|S|rtjS|r tjStd)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 NUMPYPANDASrF)r%has_transform_pandashas_transform_numpyrrr_determine_transform_to_uses  z(Preprocessor._determine_transform_to_usecCs|}|}|dur||d<|dur||d<|dur ||d<|dur(||d<|tjkr:|j|jfdtji|S|tjkrL|j|jfdtji|Std|)Nr8r9r:r; batch_formatziInvalid transform type returned from _determine_transform_to_use; "pandas" and "numpy" allowed, but got: ) rP_get_transform_configr rM map_batchesrIrLrJ ValueError)r%r.r:r8r9r;transform_typekwargsrrrr?s>  zPreprocessor._transformcCsiS)zReturns kwargs to be passed to :meth:`ray.data.Dataset.map_batches`. This can be implemented by subclassing preprocessors. rr,rrrrR$sz"Preprocessor._get_transform_configcCsddl}ddl}ddlm}m}zddl}Wn ty!d}Ynwt||j|j t j j |j fs:tdt|d|}|tjkrJ|||S|tjkrV|||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_conversionrWrXpyarrow ImportError isinstance DataFrameTable collectionsabcMappingndarrayrTtyperPr rMrIrLrJ)r%rAnppdrWrXr]rUrrrrB+s,    zPreprocessor._transform_batchcolumnsoutput_columnscCs$|rt|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.)lenrT)clsrirjrrr#_derive_and_validate_output_columnsJs z0Preprocessor._derive_and_validate_output_columnsdf pd.DataFramecCrD)zDRun the transformation on a data batch in a Pandas DataFrame format.rE)r%rnrrrrI_rGzPreprocessor._transform_pandasnp_dataz np.ndarraycCrD)zARun the transformation on a data batch in a NumPy ndarray format.rE)r%rprrrrJdszPreprocessor._transform_numpycCstjS)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 rM)rlrrrrKks z#Preprocessor.preferred_batch_formatcCstt|dS)zReturn this preprocessor serialized as a string. Note: This is not a stable serialization format as it uses `pickle`. ascii)base64 b64encodepickledumpsdecoder,rrr serializexszPreprocessor.serialize serializedcCstt|S)zALoad the original preprocessor serialized via `self.serialize()`.)rtloadsrr b64decode)rxrrr deserializeszPreprocessor.deserialize)r(r)r.r r(r)rAr r(r )NNN)rnror(ro)$rrrrstrrrr*r'r-r3rfloatintr=r<rCr r1r rPr?rrrRrB classmethodrrmrIrrJrKrw staticmethodr{rrrrrs % & * ! $       r)rcrrrbrtr/enumrtypingrrrrrrr\r ray.util.annotationsr r rZrgr[rhray.air.data_batch_typer ray.datar RuntimeErrorrABCrrrrrs$