Ray Distributed Computing Cheat Sheet

import rayray.init()  # or ray.init(address="auto") to join an existing cluster@ray.remotedef square(x):    return x * xfutures = [square.remote(i) for i in range(10)]results = ray.get(futures)@ray.remoteclass Counter:    def __init__(self):        self.n = 0    def incr(self):        self.n += 1        return self.ncounter = Counter.remote()ray.get([counter.incr.remote() for _ in range(5)])  # -> 5

from ray.train.torch import TorchTrainerfrom ray.train import ScalingConfigdef train_loop_per_worker(config):    model = build_model()    model = ray.train.torch.prepare_model(model)    for epoch in range(config["epochs"]):        loss = train_one_epoch(model)        ray.train.report({"loss": loss})trainer = TorchTrainer(    train_loop_per_worker,    train_loop_config={"epochs": 10},    scaling_config=ScalingConfig(num_workers=4, use_gpu=True),)result = trainer.fit()

from ray import tunefrom ray.tune.schedulers import ASHASchedulerdef objective(config):    for step in range(20):        acc = train_step(config["lr"], config["batch_size"])        tune.report({"accuracy": acc})tuner = tune.Tuner(    objective,    param_space={"lr": tune.loguniform(1e-4, 1e-1), "batch_size": tune.choice([16, 32, 64])},    tune_config=tune.TuneConfig(scheduler=ASHAScheduler(metric="accuracy", mode="max"), num_samples=50),)results = tuner.fit()print(results.get_best_result().config)

from ray import serve@serve.deployment(num_replicas=2, ray_actor_options={"num_gpus": 0.5})class Predictor:    def __init__(self):        self.model = load_model()    async def __call__(self, request):        data = await request.json()        return {"prediction": self.model.predict(data["input"])}serve.run(Predictor.bind(), route_prefix="/predict")