-
Notifications
You must be signed in to change notification settings - Fork 5.3k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[train] Hard-deprecate MosaicTrainer
and remove SklearnTrainer
#42814
Merged
justinvyu
merged 14 commits into
ray-project:master
from
justinvyu:deprecate_sklearn_trainer
Jan 31, 2024
Merged
Changes from all commits
Commits
Show all changes
14 commits
Select commit
Hold shift + click to select a range
124d099
remove sklearn trainer
justinvyu 9920d4d
remove unused utils
justinvyu 5c3a86c
add message
justinvyu e35a6aa
Remove sklearn trainer test
justinvyu 38aec83
fix lint
justinvyu 16d6c99
remove mosaic trainer
justinvyu 3b850ea
add todo
justinvyu fd823a3
add gh issue
justinvyu 77ba966
remove mosaic trainer test
justinvyu e96f703
Merge branch 'master' of https://github.com/ray-project/ray into depr…
justinvyu 4e9cdf0
Merge branch 'master' of https://github.com/ray-project/ray into depr…
justinvyu b72ffc9
remove test
justinvyu 3ac6b40
Merge branch 'master' of https://github.com/ray-project/ray into depr…
justinvyu 1a748c8
update issue
justinvyu File filter
Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,255 +1,30 @@ | ||
import inspect | ||
import warnings | ||
from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, Type | ||
from ray.util.annotations import Deprecated | ||
|
||
from composer.loggers.logger_destination import LoggerDestination | ||
from composer.trainer import Trainer | ||
_DEPRECATION_MESSAGE = ( | ||
"`ray.train.mosaic.MosaicTrainer` is deprecated. " | ||
"Use `ray.train.torch.TorchTrainer` instead. " | ||
"See this issue for more information: " | ||
"https://github.com/ray-project/ray/issues/42893" | ||
) | ||
|
||
from ray.train import Checkpoint, DataConfig, RunConfig, ScalingConfig | ||
from ray.train.mosaic._mosaic_utils import RayLogger | ||
from ray.train.torch import TorchConfig, TorchTrainer | ||
from ray.train.trainer import GenDataset | ||
from ray.util import PublicAPI | ||
|
||
if TYPE_CHECKING: | ||
from ray.data.preprocessor import Preprocessor | ||
|
||
|
||
@PublicAPI(stability="alpha") | ||
class MosaicTrainer(TorchTrainer): | ||
"""A Trainer for data parallel Mosaic Composers on PyTorch training. | ||
|
||
This Trainer runs the ``composer.trainer.Trainer.fit()`` method on multiple | ||
Ray Actors. The training is carried out in a distributed fashion through PyTorch | ||
DDP. These actors already have the necessary torch process group already | ||
configured for distributed PyTorch training. | ||
|
||
The training function ran on every Actor will first run the | ||
specified ``trainer_init_per_worker`` function to obtain an instantiated | ||
``composer.Trainer`` object. The ``trainer_init_per_worker`` function | ||
will have access to preprocessed train and evaluation datasets. | ||
|
||
Example: | ||
|
||
.. | ||
TODO(yunxuan): Enable the test after we resolve the mosaicml dependency issue | ||
|
||
.. testcode:: | ||
:skipif: True | ||
|
||
import torch.utils.data | ||
import torchvision | ||
from torchvision import transforms, datasets | ||
|
||
from composer.models.tasks import ComposerClassifier | ||
import composer.optim | ||
from composer.algorithms import LabelSmoothing | ||
|
||
import ray | ||
import ray.train as train | ||
from ray.train import ScalingConfig | ||
from ray.train.mosaic import MosaicTrainer | ||
|
||
def trainer_init_per_worker(config): | ||
# prepare the model for distributed training and wrap with | ||
# ComposerClassifier for Composer Trainer compatibility | ||
model = torchvision.models.resnet18(num_classes=10) | ||
model = ComposerClassifier(ray.train.torch.prepare_model(model)) | ||
|
||
# prepare train/test dataset | ||
mean = (0.507, 0.487, 0.441) | ||
std = (0.267, 0.256, 0.276) | ||
cifar10_transforms = transforms.Compose( | ||
[transforms.ToTensor(), transforms.Normalize(mean, std)] | ||
) | ||
data_directory = "~/data" | ||
train_dataset = datasets.CIFAR10( | ||
data_directory, | ||
train=True, | ||
download=True, | ||
transform=cifar10_transforms | ||
) | ||
|
||
# prepare train dataloader | ||
batch_size_per_worker = BATCH_SIZE // session.get_world_size() | ||
train_dataloader = torch.utils.data.DataLoader( | ||
train_dataset, | ||
batch_size=batch_size_per_worker | ||
) | ||
train_dataloader = ray.train.torch.prepare_data_loader(train_dataloader) | ||
|
||
# prepare optimizer | ||
optimizer = composer.optim.DecoupledSGDW( | ||
model.parameters(), | ||
lr=0.05, | ||
momentum=0.9, | ||
weight_decay=2.0e-3, | ||
) | ||
|
||
return composer.trainer.Trainer( | ||
model=model, | ||
train_dataloader=train_dataloader, | ||
optimizers=optimizer, | ||
**config | ||
) | ||
|
||
scaling_config = ScalingConfig(num_workers=2, use_gpu=True) | ||
trainer_init_config = { | ||
"max_duration": "1ba", | ||
"algorithms": [LabelSmoothing()], | ||
} | ||
|
||
trainer = MosaicTrainer( | ||
trainer_init_per_worker=trainer_init_per_worker, | ||
trainer_init_config=trainer_init_config, | ||
scaling_config=scaling_config, | ||
) | ||
|
||
trainer.fit() | ||
|
||
.. testoutput:: | ||
:hide: | ||
|
||
... | ||
|
||
Args: | ||
trainer_init_per_worker: The function that returns an instantiated | ||
``composer.Trainer`` object and takes in configuration | ||
dictionary (``config``) as an argument. This dictionary is based on | ||
``trainer_init_config`` and is modified for Ray - Composer integration. | ||
datasets: Any Datasets to use for training. At the moment, we do not support | ||
passing datasets to the trainer and using the dataset shards in the trainer | ||
loop. Instead, configure and load the datasets inside | ||
``trainer_init_per_worker`` function | ||
trainer_init_config: Configurations to pass into ``trainer_init_per_worker`` as | ||
kwargs. Although the kwargs can be hard-coded in the | ||
``trainer_init_per_worker``, using the config allows the flexibility of | ||
reusing the same worker init function while changing the trainer arguments. | ||
For example, when hyperparameter tuning you can reuse the | ||
same ``trainer_init_per_worker`` function with different hyperparameter | ||
values rather than having multiple ``trainer_init_per_worker`` functions | ||
with different hard-coded hyperparameter values. | ||
torch_config: Configuration for setting up the PyTorch backend. If set to | ||
None, use the default configuration. This replaces the ``backend_config`` | ||
arg of ``DataParallelTrainer``. Same as in ``TorchTrainer``. | ||
scaling_config: Configuration for how to scale data parallel training. | ||
dataset_config: Configuration for dataset ingest. | ||
run_config: Configuration for the execution of the training run. | ||
resume_from_checkpoint: A ``ray.train.Checkpoint`` to resume training from. | ||
# TODO(justinvyu): [code_removal] Delete in Ray 2.11. | ||
@Deprecated | ||
class MosaicTrainer: | ||
"""Deprecated. See this issue for more information: | ||
https://github.com/ray-project/ray/issues/42893 | ||
""" | ||
|
||
def __init__( | ||
self, | ||
trainer_init_per_worker: Callable[[Optional[Dict]], Trainer], | ||
*, | ||
datasets: Optional[Dict[str, GenDataset]] = None, | ||
trainer_init_config: Optional[Dict] = None, | ||
torch_config: Optional[TorchConfig] = None, | ||
scaling_config: Optional[ScalingConfig] = None, | ||
dataset_config: Optional[DataConfig] = None, | ||
run_config: Optional[RunConfig] = None, | ||
preprocessor: Optional["Preprocessor"] = None, | ||
resume_from_checkpoint: Optional[Checkpoint] = None, | ||
): | ||
|
||
warnings.warn( | ||
"This MosaicTrainer will be deprecated in Ray 2.8. " | ||
"It is recommended to use the TorchTrainer instead.", | ||
DeprecationWarning, | ||
) | ||
|
||
self._validate_trainer_init_per_worker( | ||
trainer_init_per_worker, "trainer_init_per_worker" | ||
) | ||
def __new__(cls, *args, **kwargs): | ||
raise DeprecationWarning(_DEPRECATION_MESSAGE) | ||
|
||
self._validate_datasets(datasets) | ||
self._validate_trainer_init_config(trainer_init_config) | ||
|
||
if resume_from_checkpoint: | ||
# TODO(ml-team): Reenable after Mosaic checkpointing is supported | ||
raise NotImplementedError | ||
|
||
super().__init__( | ||
train_loop_per_worker=_mosaic_train_loop_per_worker, | ||
train_loop_config=self._create_trainer_init_config( | ||
trainer_init_per_worker, trainer_init_config | ||
), | ||
torch_config=torch_config, | ||
scaling_config=scaling_config, | ||
dataset_config=dataset_config, | ||
run_config=run_config, | ||
datasets=datasets, | ||
preprocessor=preprocessor, | ||
resume_from_checkpoint=resume_from_checkpoint, | ||
) | ||
def __init__(self, *args, **kwargs): | ||
raise DeprecationWarning(_DEPRECATION_MESSAGE) | ||
|
||
@classmethod | ||
def _create_trainer_init_config( | ||
cls, | ||
trainer_init_per_worker: Callable[[Optional[Dict]], Trainer], | ||
trainer_init_config: Optional[Dict[str, Any]], | ||
) -> Dict[str, Any]: | ||
trainer_init_config = trainer_init_config.copy() if trainer_init_config else {} | ||
if "_trainer_init_per_worker" in trainer_init_config: | ||
raise ValueError( | ||
"'_trainer_init_per_worker' is a reserved key in `trainer_init_config`." | ||
) | ||
trainer_init_config["_trainer_init_per_worker"] = trainer_init_per_worker | ||
return trainer_init_config | ||
def restore(cls, *args, **kwargs): | ||
raise DeprecationWarning(_DEPRECATION_MESSAGE) | ||
|
||
@classmethod | ||
def restore(cls: Type["MosaicTrainer"], **kwargs) -> "MosaicTrainer": | ||
# TODO(ml-team): Reenable after Mosaic checkpointing is supported | ||
raise NotImplementedError | ||
|
||
def _validate_trainer_init_per_worker( | ||
self, trainer_init_per_worker: Callable, fn_name: str | ||
) -> None: | ||
num_params = len(inspect.signature(trainer_init_per_worker).parameters) | ||
if num_params != 1: | ||
raise ValueError( | ||
f"{fn_name} should take in at most 1 argument (`config`), " | ||
f"but it accepts {num_params} arguments instead." | ||
) | ||
|
||
def _validate_datasets(self, datasets) -> None: | ||
if not (datasets is None or len(datasets) == 0): | ||
raise ValueError( | ||
"MosaicTrainer does not support providing dataset shards \ | ||
to `trainer_init_per_worker`. Instead of passing in the dataset into \ | ||
MosaicTrainer, define a dataloader and use `prepare_dataloader` \ | ||
inside the `trainer_init_per_worker`." | ||
) | ||
|
||
def _validate_trainer_init_config(self, config) -> None: | ||
if config is not None and "loggers" in config: | ||
warnings.warn( | ||
"Composer's Loggers (any subclass of LoggerDestination) are \ | ||
not supported for MosaicComposer. Use Ray provided loggers instead" | ||
) | ||
|
||
|
||
def _mosaic_train_loop_per_worker(config): | ||
"""Per-worker training loop for Mosaic Composers.""" | ||
trainer_init_per_worker = config.pop("_trainer_init_per_worker") | ||
|
||
# Replace Composer's Loggers with RayLogger | ||
ray_logger = RayLogger(keys=config.pop("log_keys", [])) | ||
|
||
# initialize Composer trainer | ||
trainer: Trainer = trainer_init_per_worker(config) | ||
|
||
# Remove Composer's Loggers if there are any added in the trainer_init_per_worker | ||
# this removes the logging part of the loggers | ||
filtered_callbacks = list() | ||
for callback in trainer.state.callbacks: | ||
if not isinstance(callback, LoggerDestination): | ||
filtered_callbacks.append(callback) | ||
filtered_callbacks.append(ray_logger) | ||
trainer.state.callbacks = filtered_callbacks | ||
|
||
# this prevents data to be routed to all the Composer Loggers | ||
trainer.logger.destinations = (ray_logger,) | ||
|
||
# call the trainer | ||
trainer.fit() | ||
def can_restore(cls, *args, **kwargs): | ||
raise DeprecationWarning(_DEPRECATION_MESSAGE) |
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Do we still need SklearnPredictor?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We still have all the predictors around but not shown in docs. Maybe we can remove them all at some point at once?