class_docstr: Defines interface for distributed training on Ray.
  
  Note: The base ``BaseTrainer`` class cannot be instantiated directly. Only
  one of its subclasses can be used.
  
  Note to developers: If a new trainer is added, please update
  `air/_internal/usage.py`.
  
  **How does a trainer work?**
  
  - First, initialize the Trainer. The initialization runs locally,
    so heavyweight setup should not be done in ``__init__``.
  - Then, when you call ``trainer.fit()``, the Trainer is serialized
    and copied to a remote Ray actor. The following methods are then
    called in sequence on the remote actor.
  - ``trainer.setup()``: Any heavyweight Trainer setup should be
    specified here.
  - ``trainer.preprocess_datasets()``: The datasets passed to the Trainer will be
    setup here.
  - ``trainer.train_loop()``: Executes the main training logic.
  - Calling ``trainer.fit()`` will return a ``ray.result.Result``
    object where you can access metrics from your training run, as well
    as any checkpoints that may have been saved.
  
  **How do I create a new Trainer?**
  
  Subclass ``ray.train.trainer.BaseTrainer``, and override the ``training_loop``
  method, and optionally ``setup``.
  
  .. testcode::
  
      import torch
  
      from ray.train.trainer import BaseTrainer
      from ray import train, tune
  
  
      class MyPytorchTrainer(BaseTrainer):
          def setup(self):
              self.model = torch.nn.Linear(1, 1)
              self.optimizer = torch.optim.SGD(
                  self.model.parameters(), lr=0.1)
  
          def training_loop(self):
              # You can access any Trainer attributes directly in this method.
              # self.datasets["train"] has already been
              dataset = self.datasets["train"]
  
              torch_ds = dataset.iter_torch_batches(dtypes=torch.float)
              loss_fn = torch.nn.MSELoss()
  
              for epoch_idx in range(10):
                  loss = 0
                  num_batches = 0
                  torch_ds = dataset.iter_torch_batches(
                      dtypes=torch.float, batch_size=2
                  )
                  for batch in torch_ds:
                      X = torch.unsqueeze(batch["x"], 1)
                      y = torch.unsqueeze(batch["y"], 1)
                      # Compute prediction error
                      pred = self.model(X)
                      batch_loss = loss_fn(pred, y)
  
                      # Backpropagation
                      self.optimizer.zero_grad()
                      batch_loss.backward()
                      self.optimizer.step()
  
                      loss += batch_loss.item()
                      num_batches += 1
                  loss /= num_batches
  
                  # Use Tune functions to report intermediate
                  # results.
                  train.report({"loss": loss, "epoch": epoch_idx})
  
  
      # Initialize the Trainer, and call Trainer.fit()
      import ray
      train_dataset = ray.data.from_items(
          [{"x": i, "y": i} for i in range(10)])
      my_trainer = MyPytorchTrainer(datasets={"train": train_dataset})
      result = my_trainer.fit()
  
  .. testoutput::
          :hide:
  
          ...
  
  Args:
      scaling_config: Configuration for how to scale training.
      run_config: Configuration for the execution of the training run.
      datasets: Any Datasets to use for training. Use the key "train"
          to denote which dataset is the training dataset.
      metadata: Dict that should be made available via
          `train.get_context().get_metadata()` and in `checkpoint.get_metadata()`
          for checkpoints saved from this Trainer. Must be JSON-serializable.
      resume_from_checkpoint: A checkpoint to resume training from.
class_docstr_tok: ["Defines","interface","for","distributed","training","on","Ray",".","Note",":","The","base","``","BaseTrainer","``","class","can","not","be","instantiated","directly",".","Only","one","of","its","subclasses","can","be","used",".","Note","to","developers",":","If","a","new","trainer","is","added",",","please","update","`","air\/_internal\/usage.py","`",".","*","*","How","does","a","trainer","work","?","*","*","-","First",",","initialize","the","Trainer",".","The","initialization","runs","locally",",","so","heavyweight","setup","should","not","be","done","in","``","__init__","``",".","-","Then",",","when","you","call","``","trainer.fit","(",")","``",",","the","Trainer","is","serialized","and","copied","to","a","remote","Ray","actor",".","The","following","methods","are","then","called","in","sequence","on","the","remote","actor",".","-","``","trainer.setup","(",")","``",":","Any","heavyweight","Trainer","setup","should","be","specified","here",".","-","``","trainer.preprocess_datasets","(",")","``",":","The","datasets","passed","to","the","Trainer","will","be","setup","here",".","-","``","trainer.train_loop","(",")","``",":","Executes","the","main","training","logic",".","-","Calling","``","trainer.fit","(",")","``","will","return","a","``","ray.result.Result","``","object","where","you","can","access","metrics","from","your","training","run",",","as","well","as","any","checkpoints","that","may","have","been","saved",".","*","*","How","do","I","create","a","new","Trainer","?","*","*","Subclass","``","ray.train.trainer.BaseTrainer","``",",","and","override","the","``","training_loop","``","method",",","and","optionally","``","setup","``",".","..","testcode",":",":","import","torch","from","ray.train.trainer","import","BaseTrainer","from","ray","import","train",",","tune","class","MyPytorchTrainer","(","BaseTrainer",")",":","def","setup","(","self",")",":","self.model","=","torch.nn.Linear","(","1",",","1",")","self.optimizer","=","torch.optim.SGD","(","self.model.parameters","(",")",",","lr=0.1",")","def","training_loop","(","self",")",":","#","You","can","access","any","Trainer","attributes","directly","in","this","method",".","#","self.datasets","[","``","train","''","]","has","already","been","dataset","=","self.datasets","[","``","train","''","]","torch_ds","=","dataset.iter_torch_batches","(","dtypes=torch.float",")","loss_fn","=","torch.nn.MSELoss","(",")","for","epoch_idx","in","range","(","10",")",":","loss","=","0","num_batches","=","0","torch_ds","=","dataset.iter_torch_batches","(","dtypes=torch.float",",","batch_size=2",")","for","batch","in","torch_ds",":","X","=","torch.unsqueeze","(","batch","[","``","x","''","]",",","1",")","y","=","torch.unsqueeze","(","batch","[","``","y","''","]",",","1",")","#","Compute","prediction","error","pred","=","self.model","(","X",")","batch_loss","=","loss_fn","(","pred",",","y",")","#","Backpropagation","self.optimizer.zero_grad","(",")","batch_loss.backward","(",")","self.optimizer.step","(",")","loss","+=","batch_loss.item","(",")","num_batches","+=","1","loss","\/=","num_batches","#","Use","Tune","functions","to","report","intermediate","#","results",".","train.report","(","{","``","loss","''",":","loss",",","``","epoch","''",":","epoch_idx","}",")","#","Initialize","the","Trainer",",","and","call","Trainer.fit","(",")","import","ray","train_dataset","=","ray.data.from_items","(","[","{","``","x","''",":","i",",","``","y","''",":","i","}","for","i","in","range","(","10",")","]",")","my_trainer","=","MyPytorchTrainer","(","datasets=","{","``","train","''",":","train_dataset","}",")","result","=","my_trainer.fit","(",")","..","testoutput",":",":",":","hide",":","...","Args",":","scaling_config",":","Configuration","for","how","to","scale","training",".","run_config",":","Configuration","for","the","execution","of","the","training","run",".","datasets",":","Any","Datasets","to","use","for","training",".","Use","the","key","``","train","''","to","denote","which","dataset","is","the","training","dataset",".","metadata",":","Dict","that","should","be","made","available","via","`","train.get_context","(",")",".get_metadata","(",")","`","and","in","`","checkpoint.get_metadata","(",")","`","for","checkpoints","saved","from","this","Trainer",".","Must","be","JSON-serializable",".","resume_from_checkpoint",":","A","checkpoint","to","resume","training","from","."]
docstr: Called during fit() to preprocess dataset attributes with preprocessor.
  
  .. note:: This method is run on a remote process.
  
  This method is called prior to entering the training_loop.
  
  If the ``Trainer`` has both a datasets dict and
  a preprocessor, the datasets dict contains a training dataset (denoted by
  the "train" key), and the preprocessor has not yet
  been fit, then it will be fit on the train dataset.
  
  Then, all Trainer's datasets will be transformed by the preprocessor.
  
  The transformed datasets will be set back in the ``self.datasets`` attribute
  of the Trainer to be used when overriding ``training_loop``.
code: def preprocess_datasets(self) -> None:
      """Called during fit() to preprocess dataset attributes with preprocessor.
  
      .. note:: This method is run on a remote process.
  
      This method is called prior to entering the training_loop.
  
      If the ``Trainer`` has both a datasets dict and
      a preprocessor, the datasets dict contains a training dataset (denoted by
      the "train" key), and the preprocessor has not yet
      been fit, then it will be fit on the train dataset.
  
      Then, all Trainer's datasets will be transformed by the preprocessor.
  
      The transformed datasets will be set back in the ``self.datasets`` attribute
      of the Trainer to be used when overriding ``training_loop``.
      """
      # Evaluate all datasets.
      self.datasets = {
          k: d() if callable(d) else d for k, d in self.datasets.items()
      }
  
      if self.preprocessor:
          train_dataset = self.datasets.get(TRAIN_DATASET_KEY, None)
          if train_dataset and self.preprocessor.fit_status() in (
              ray.data.Preprocessor.FitStatus.NOT_FITTED,
              ray.data.Preprocessor.FitStatus.PARTIALLY_FITTED,
          ):
              self.preprocessor.fit(train_dataset)
  
          # Execute dataset transformations serially for now.
          # Cannot execute them in remote tasks due to dataset ownership model:
          # if datasets are created on a remote node, then if that node fails,
          # we cannot recover the dataset.
          new_datasets = {}
          for key, dataset in self.datasets.items():
              new_datasets[key] = self.preprocessor.transform(dataset)
  
          self.datasets = new_datasets
  
imports: import abc
  import copy
  import inspect
  import json
  import logging
  import os
  import warnings
  from pathlib import Path
  from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Type, Union
  import pyarrow.fs
  import ray
  import ray.cloudpickle
  from ray._private.dict import merge_dicts
  from ray.air._internal import usage
  from ray.air._internal.config import ensure_only_allowed_dataclass_keys_updated
  from ray.air._internal.usage import AirEntrypoint
  from ray.air.config import RunConfig, ScalingConfig
  from ray.air.result import Result
  from ray.train import Checkpoint
  from ray.train._internal.session import _get_session
  from ray.train._internal.storage import _exists_at_fs_path, get_fs_and_path
  from ray.train.constants import TRAIN_DATASET_KEY
  from ray.util import PublicAPI
  from ray.util.annotations import DeveloperAPI
label_desc: Use image node_id 9 for calling the BaseTrainer obj's underlying member method code with example usage: obj.preprocess_datasets() without return types