[Data] Add serialization framework for preprocessors (ray-project#58321)

## Description
This commit introduces a new serialization system for Ray Data
preprocessors that improves maintainability, extensibility, and backward
compatibility.

Key changes:

1. New serialization infrastructure:
- Add serialization_handlers.py with factory pattern for format handling
   - Implement CloudPickleSerializationHandler (primary format)
- Support legacy PickleSerializationHandler for backward compatibility
   - Add format auto-detection via magic bytes (CPKL:)

2. New preprocessor base class:
   - Add SerializablePreprocessorBase abstract class
   - Define serialization interface via abstract methods:
     * _get_serializable_fields() / _set_serializable_fields()
     * _get_stats() / _set_stats()
   - Mark serialize() and deserialize() as @Final to prevent overrides

3. Preprocessor registration system:
   - Add version_support.py with @SerializablePreprocessor decorator
   - Enable versioned serialization with stable identifiers
   - Support class registration and lookup
   - Add UnknownPreprocessorError for missing types

4. Migrate preprocessors to new framework:
   - SimpleImputer
   - OrdinalEncoder
   - OneHotEncoder
   - MultiHotEncoder
   - LabelEncoder
   - Categorizer
   - StandardScaler
   - MinMaxScaler
   - MaxAbsScaler
    - RobustScaler

5. Enhanced Preprocessor base class:
- Add get_input_columns() and get_output_columns() methods (for future
use)
   - Add has_stats() (for future use)
   - Add type hints to __getstate__() and __setstate__()

6. Backward compatibility improvements to Concatenator for existing
functionality:
   - Add __setstate__ override in Concatenator for flatten field
   - Handle missing fields gracefully during deserialization

The new architecture makes it easier to:
- Add new serialization formats without modifying core logic
- Maintain backward compatibility with existing serialized data
- Handle version migrations for preprocessor schemas
- Register new preprocessors with stable identifiers

---------

Signed-off-by: cem <cem@anyscale.com>

Author: cem-anyscale

doc/source/conf.py

@@ -135,6 +135,9 @@
     ("py:class", ".*"),
     # Workaround for https://github.com/sphinx-doc/sphinx/issues/10974
     ("py:obj", "ray\\.data\\.datasource\\.datasink\\.WriteReturnType"),
+    # UnknownPreprocessorError is an internal exception not exported in public API
+    ("py:exc", "UnknownPreprocessorError"),
+    ("py:exc", "ray\\.data\\.preprocessors\\.version_support\\.UnknownPreprocessorError"),
 ]
 
 # Cache notebook outputs in _build/.jupyter_cache

doc/source/train/user-guides/data-loading-preprocessing.rst

@@ -502,6 +502,7 @@ You can use this with Ray Train Trainers by applying them on the dataset before
 
 .. testcode::
 
+    import base64
     import numpy as np
     from tempfile import TemporaryDirectory
 
@@ -542,16 +543,22 @@ You can use this with Ray Train Trainers by applying them on the dataset before
                 checkpoint=Checkpoint.from_directory(temp_dir),
             )
 
+    # Serialize the preprocessor. Since serialize() returns bytes,
+    # convert to base64 string for JSON compatibility.
+    serialized_preprocessor = base64.b64encode(scaler.serialize()).decode("ascii")
+
     my_trainer = TorchTrainer(
         train_loop_per_worker,
         scaling_config=ScalingConfig(num_workers=2),
         datasets={"train": dataset},
-        metadata={"preprocessor_pkl": scaler.serialize()},
+        metadata={"preprocessor_pkl": serialized_preprocessor},
     )
 
     # Get the fitted preprocessor back from the result metadata.
     metadata = my_trainer.fit().checkpoint.get_metadata()
-    print(StandardScaler.deserialize(metadata["preprocessor_pkl"]))
+    # Decode from base64 before deserializing
+    serialized_data = base64.b64decode(metadata["preprocessor_pkl"])
+    print(StandardScaler.deserialize(serialized_data))
 
 
 This example persists the fitted preprocessor using the ``Trainer(metadata={...})`` constructor argument. This arg specifies a dict that is available from ``TrainContext.get_metadata()`` and ``checkpoint.get_metadata()`` for checkpoints that the Trainer saves. This design enables the recreation of the fitted preprocessor for inference.