Ray Train & Ray Tune: Pelatihan Terdistribusi dan Penyetelan Hiperparameter

Sebagian besar proyek machine learning dimulai pada satu mesin, dan pada titik tertentu loop pelatihan kehabisan memori, terlalu lama pada satu GPU, atau membutuhkan ratusan percobaan hiperparameter yang tidak mungkin diselesaikan satu proses dalam waktu wajar. Ray Train dan Ray Tune menjawab kedua masalah itu secara langsung: yang pertama menskalakan loop pelatihan yang sudah ada ke beberapa worker dan GPU, dan yang kedua menjalankan banyak proses pelatihan secara paralel untuk menelusuri ruang hiperparameter. Tutorial ini membahas kedua pustaka tersebut dengan satu contoh PyTorch yang konsisten, lalu menunjukkan cara menggabungkannya sehingga Anda dapat menyetel hiperparameter sebuah job pelatihan terdistribusi tanpa menulis ulang kode.

Pengantar Singkat Ray dan Pustaka AI-nya

Ray adalah framework untuk Python terdistribusi. Pada intinya, Ray menyediakan task dan actor yang berjalan di seluruh cluster, ditambah scheduler yang menempatkan pekerjaan pada CPU dan GPU yang tersedia. Di atas inti itu, Ray AI Libraries menyediakan blok bangunan tingkat lebih tinggi untuk kebutuhan machine learning yang umum:

• Ray Data untuk pemuatan dan prapemrosesan data terdistribusi.
• Ray Train untuk pelatihan model terdistribusi.
• Ray Tune untuk pencarian hiperparameter dan orkestrasi eksperimen.
• Ray Serve untuk penyajian model online (dibahas di tutorial terpisah, jadi di sini hanya disebut sebagai konteks).

Sifat penting di sini adalah kode yang sama berjalan di laptop Anda maupun di cluster multi-node. Anda mengembangkan secara lokal, lalu mengarahkan skrip yang sama ke cluster ketika membutuhkan komputasi lebih besar. Tutorial ini berfokus pada Ray Train dan Ray Tune, yang bersama-sama mencakup tahap pelatihan dan penyetelan dalam alur kerja umum.

Instalasi

Instal Ray dengan ekstra Train dan Tune, ditambah PyTorch:

pip install "ray[train,tune]" torch torchvision

Verifikasi instalasi:

python -c "import ray; from ray import train, tune; print(ray.version)"

Contoh di bawah ditulis untuk Ray 2.x. API Ray Train ditata ulang pada seri 2.x, jadi pastikan Anda tidak memakai rilis di bawah 2.7 yang impornya berbeda.

Ray Train: Dari Satu Node ke Terdistribusi

Titik awal: loop PyTorch biasa

Perhatikan loop pelatihan satu proses yang umum untuk klasifikasi gambar pada FashionMNIST. Belum ada apa pun yang terdistribusi di dalamnya.

import torch
import torch.nn as nn
from torch.utils.data import DataLoader
from torchvision import datasets, transforms


def buildmodel():

    return nn.Sequential(
        nn.Flatten(),
        nn.Linear(28  28, 256),

        nn.ReLU(),
        nn.Linear(256, 128),
        nn.ReLU(),
        nn.Linear(128, 10),
    )


def trainsinglenode(epochs=5, lr=1e-3, batchsize=64):
    transform = transforms.ToTensor()
    traindata = datasets.FashionMNIST(

        root="./data", train=True, download=True, transform=transform
    )
    loader = DataLoader(traindata, batchsize=batchsize, shuffle=True)

    device = "cuda" if torch.cuda.isavailable() else "cpu"

    model = buildmodel().to(device)
    optimizer = torch.optim.Adam(model.parameters(), lr=lr)
    lossfn = nn.CrossEntropyLoss()


    for epoch in range(epochs):
        model.train()
        for images, labels in loader:
            images, labels = images.to(device), labels.to(device)
            optimizer.zerograd()
            loss = lossfn(model(images), labels)

            loss.backward()
            optimizer.step()
        print(f"epoch {epoch}: last batch loss {loss.item():.4f}")

Kode ini bekerja, tetapi memakai satu proses dan satu device. Untuk melatih lintas beberapa GPU atau beberapa node dengan data-parallelism, kita menyesuaikan loop ini untuk Ray Train.

Mengonversi loop dengan TorchTrainer

Ray Train membungkus fungsi pelatihan Anda dan menjalankan satu salinannya per worker. Setiap worker menangani satu shard data dan menyinkronkan gradien secara otomatis. Perubahannya kecil dan terlokalisasi:

Bungkus model dengan ray.train.torch.preparemodel, yang memindahkannya ke device yang benar dan membungkusnya dalam DistributedDataParallel.

Bungkus data loader dengan ray.train.torch.preparedataloader, yang menyisipkan DistributedSampler sehingga tiap worker melihat shard berbeda.

Laporkan metrik dan checkpoint dengan ray.train.report alih-alih print biasa.

import os
import tempfile

import torch
import torch.nn as nn
from torch.utils.data import DataLoader
from torchvision import datasets, transforms

import ray.train.torch
from ray.train import Checkpoint, ScalingConfig
from ray.train.torch import TorchTrainer


def buildmodel():

    return nn.Sequential(
        nn.Flatten(),
        nn.Linear(28  28, 256),
        nn.ReLU(),
        nn.Linear(256, 128),
        nn.ReLU(),
        nn.Linear(128, 10),
    )


def trainfunc(config):
    lr = config["lr"]
    epochs = config["epochs"]
    batchsize = config["batchsize"]

    transform = transforms.ToTensor()
    traindata = datasets.FashionMNIST(

        root="./data", train=True, download=True, transform=transform
    )
    loader = DataLoader(traindata, batchsize=batchsize, shuffle=True)
    # Menyisipkan DistributedSampler dan memindahkan batch ke device worker.
    loader = ray.train.torch.preparedataloader(loader)

    model = buildmodel()

    # Memindahkan model ke device dan membungkusnya dalam DistributedDataParallel.
    model = ray.train.torch.preparemodel(model)

    optimizer = torch.optim.Adam(model.parameters(), lr=lr)
    lossfn = nn.CrossEntropyLoss()


    for epoch in range(epochs):
        # Dengan DistributedSampler, epoch harus diset agar pengacakan berbeda.
        if ray.train.getcontext().getworldsize() > 1:
            loader.sampler.setepoch(epoch)


        model.train()
        totalloss, nbatches = 0.0, 0

        for images, labels in loader:
            optimizer.zerograd()
            loss = lossfn(model(images), labels)

            loss.backward()
            optimizer.step()
            totalloss += loss.item()
            nbatches += 1


        avgloss = totalloss / max(nbatches, 1)

        # Simpan checkpoint dan laporkan metrik kembali ke Ray Train.
        with tempfile.TemporaryDirectory() as tmpdir:

            torch.save(
                model.module.statedict(),
                os.path.join(tmpdir, "model.pt"),

            )
            checkpoint = Checkpoint.fromdirectory(tmpdir)

            ray.train.report(
                {"loss": avgloss, "epoch": epoch},
                checkpoint=checkpoint,
            )

Ada beberapa detail yang perlu dicatat. Kita memanggil model.module.statedict() karena preparemodel membungkus model dalam DistributedDataParallel, dan bobot sebenarnya berada di bawah .module. Pelaporan dan checkpoint terjadi di setiap worker, tetapi Ray Train mengoordinasikannya sehingga secara default hanya satu rank yang menyimpan checkpoint.

Mengatur skala dengan ScalingConfig

ScalingConfig mendeklarasikan berapa banyak worker yang dijalankan dan apakah setiap worker membutuhkan GPU. Fungsi pelatihan tidak berubah saat Anda menaikkan skala; hanya konfigurasi ini yang berubah.

scalingconfig = ScalingConfig(
    numworkers=4,     # empat worker data-parallel
    usegpu=True,      # satu GPU per worker

    # resourcesperworker={"CPU": 2},  # opsional: tambahan CPU per worker

)

trainer = TorchTrainer(
    trainloopperworker=trainfunc,

    trainloopconfig={"lr": 1e-3, "epochs": 5, "batchsize": 64},
    scalingconfig=scalingconfig,
)

result = trainer.fit()

Untuk berjalan hanya pada CPU (untuk pengembangan lokal), set usegpu=False dan pilih numworkers kecil seperti 2. Ukuran batch efektif adalah batchsize * numworkers karena tiap worker memproses shard-nya sendiri, jadi Anda mungkin perlu menyesuaikan learning rate.

Mengakses hasil dan checkpoint terbaik

trainer.fit() mengembalikan objek Result yang menyimpan metrik akhir dan riwayat checkpoint.

print("Metrik akhir yang dilaporkan:", result.metrics)

Checkpoint terbaru.
bestcheckpoint = result.checkpoint


Atau pilih berdasarkan metrik di seluruh checkpoint yang dilaporkan.
bestcheckpoint = result.getbestcheckpoint(metric="loss", mode="min")

Muat kembali bobot ke model baru.
with bestcheckpoint.asdirectory() as ckptdir:

    statedict = torch.load(os.path.join(ckptdir, "model.pt"))

    model = buildmodel()
    model.loadstatedict(statedict)

Pola yang sama untuk framework lain

Pola data-parallel ini tidak khusus untuk PyTorch murni. Ray Train menyediakan trainer yang mengikuti struktur sama untuk ekosistem lain, sehingga ScalingConfig, pelaporan, dan penanganan hasil di sekitarnya tetap identik:

• PyTorch Lightning melalui utilitas ray.train.lightning yang terintegrasi dengan Trainer.
• Hugging Face Transformers dengan membungkus transformers.Trainer menggunakan callback pelaporan Ray Train.
• XGBoost dan LightGBM melalui XGBoostTrainer dan LightGBMTrainer, yang menjalankan gradient boosting terdistribusi.

Dalam setiap kasus, Anda tetap memakai kode pelatihan milik framework tersebut dan membiarkan Ray Train mengurus orkestrasi worker serta penempatan resource.

Ray Tune: Pencarian Hiperparameter yang Dapat Diskalakan

Ray Tune menjalankan banyak konfigurasi pelatihan secara paralel dan melacak hasilnya. Satu konfigurasi disebut trial. Anda memberi Tune sebuah ruang pencarian, sebuah trainable, dan sebuah anggaran, lalu Tune mengelola penjadwalan, penghentian dini, dan pencatatan.

Mendefinisikan ruang pencarian

Ruang pencarian adalah kamus berisi distribusi dari ray.tune:

from ray import tune

searchspace = {
    "lr": tune.loguniform(1e-4, 1e-1),
    "batchsize": tune.choice([32, 64, 128]),

    "epochs": 5,  # nilai tetap juga boleh; tidak semua harus dicari
}

Fungsi ruang yang umum mencakup tune.uniform, tune.loguniform, tune.choice, tune.randint, dan tune.gridsearch([...]) untuk grid menyeluruh.

Trainable sederhana

Trainable adalah fungsi yang menerima config, melatih, lalu melaporkan metrik. Untuk eksperimen Tune saja (tanpa pelatihan terdistribusi), fungsi tersebut melaporkan dengan ray.train.report seperti sebelumnya:

import ray.train


def tunetrainable(config):

    # ... bangun model, loader, optimizer memakai config["lr"], dst. ...
    for epoch in range(config["epochs"]):
        avgloss = trainoneepoch(config)  # loop Anda
        accuracy = evaluate(config)
        ray.train.report({"loss": avgloss, "accuracy": accuracy})

Menjalankan Tuner

tune.Tuner menyatukan trainable, ruang pencarian, dan sebuah TuneConfig. TuneConfig mengatur jumlah trial (numsamples), metric target, dan arah optimasi (mode).

from ray import tune
from ray.tune import TuneConfig

tuner = tune.Tuner(
    tunetrainable,

    paramspace=searchspace,

    tuneconfig=TuneConfig(
        numsamples=20,        # dua puluh trial

        metric="accuracy",
        mode="max",
    ),
)

results = tuner.fit()

Algoritma pencarian

Secara default Tune mengambil sampel dari ruang secara acak (atau menjalankan grid jika Anda memakai tune.gridsearch). Untuk pengambilan sampel yang lebih cerdas, Anda dapat menyambungkan algoritma pencarian yang belajar dari trial yang selesai. Dua pilihan yang umum:

from ray.tune.search.optuna import OptunaSearch
from ray.tune.search.hyperopt import HyperOptSearch

Tree-structured Parzen estimator melalui Optuna.
optunasearch = OptunaSearch(metric="accuracy", mode="max")


Atau implementasi TPE dari HyperOpt.
hyperoptsearch = HyperOptSearch(metric="accuracy", mode="max")

tuner = tune.Tuner(
    tunetrainable,

    paramspace=searchspace,

    tuneconfig=TuneConfig(
        numsamples=20,

        searchalg=optunasearch,

    ),
)

Perhatikan bagaimana Ray Tune berhubungan dengan Optuna: Optuna menyediakan sampler-nya, sedangkan Tune menyediakan eksekusi paralel, penjadwalan, dan manajemen resource lintas cluster. Jika Anda hanya membutuhkan penyetelan satu proses, Optuna polos lebih sederhana; Tune memberi nilai tambah begitu Anda ingin trial berjalan paralel.

Scheduler dan penghentian dini

Scheduler menentukan trial mana yang dilanjutkan, dijeda, atau dihentikan. Yang paling banyak dipakai adalah ASHA (Asynchronous Successive Halving), yang menghentikan trial berkinerja buruk lebih awal sehingga anggaran terfokus pada konfigurasi yang menjanjikan.

from ray.tune.schedulers import ASHAScheduler

scheduler = ASHAScheduler(
    metric="accuracy",
    mode="max",
    maxt=10,            # epoch maksimum (atau iterasi pelatihan) per trial
    graceperiod=2,      # jalankan minimal dua iterasi sebelum dihentikan

    reductionfactor=2,
)

tuner = tune.Tuner(
    tunetrainable,

    paramspace=searchspace,

    tuneconfig=TuneConfig(
        numsamples=40,

        scheduler=scheduler,
        metric="accuracy",
        mode="max",
    ),
)

Population Based Training (PopulationBasedTraining) adalah scheduler lain. Alih-alih hanya menghentikan trial, ia secara berkala menyalin bobot dan hiperparameter trial yang kuat ke trial yang lemah lalu memberi perturbasi, sehingga hiperparameter dapat berubah selama satu proses berjalan. Cocok untuk skema seperti peluruhan learning rate, tetapi lebih berat disiapkan, jadi gunakan ASHA lebih dulu kecuali Anda memang butuh hiperparameter yang berevolusi.

Menganalisis hasil

tuner.fit() mengembalikan ResultGrid. Anda dapat mengambil trial terbaik dan memeriksa setiap trial sebagai dataframe.

results = tuner.fit()

best = results.getbestresult(metric="accuracy", mode="max")

print("Config terbaik:", best.config)
print("Akurasi terbaik:", best.metrics["accuracy"])

Checkpoint dari trial terbaik, jika trainable menyimpannya.
bestcheckpoint = best.checkpoint

Semua trial sebagai DataFrame pandas untuk plotting atau ekspor.
df = results.getdataframe()

print(df[["accuracy", "loss", "config/lr", "config/batchsize"]].head())

Menggabungkan Ray Train dan Ray Tune

Kedua pustaka ini bersatu dengan rapi: Anda dapat menyetel hiperparameter sebuah TorchTrainer terdistribusi dengan membungkusnya dalam Tuner. Setiap trial kemudian meluncurkan kelompok worker data-parallel-nya sendiri. Kuncinya adalah meletakkan ruang pencarian di bawah kunci trainloopconfig agar Tune tahu bagian trainer mana yang divariasikan.

from ray import tune
from ray.tune import TuneConfig, Tuner
from ray.tune.schedulers import ASHAScheduler
from ray.train import ScalingConfig
from ray.train.torch import TorchTrainer

Gunakan kembali trainfunc terdistribusi dari bagian Ray Train.

trainer = TorchTrainer(
    trainloopperworker=trainfunc,

    scalingconfig=ScalingConfig(numworkers=2, usegpu=True),
)

paramspace = {

    "trainloopconfig": {

        "lr": tune.loguniform(1e-4, 1e-1),
        "batchsize": tune.choice([32, 64, 128]),
        "epochs": 5,
    }
}

tuner = Tuner(
    trainer,
    paramspace=paramspace,
    tuneconfig=TuneConfig(

        numsamples=16,
        metric="loss",
        mode="min",
        scheduler=ASHAScheduler(maxt=5, graceperiod=1),
    ),
)

results = tuner.fit()
best = results.getbestresult(metric="loss", mode="min")
print("Config terdistribusi terbaik:", best.config["trainloopconfig"])

Perhatikan total kebutuhan resource. Dengan numsamples=16, numworkers=2, dan satu GPU per worker, proses yang sepenuhnya paralel membutuhkan 32 GPU. Pada cluster yang lebih kecil, Tune sekadar mengantrekan trial sampai resource tersedia, tetapi Anda sebaiknya menyesuaikan numsamples dan numworkers dengan perangkat keras yang benar-benar Anda miliki.

Mesin Lokal vs Cluster Ray

Semua di atas berjalan tanpa perubahan pada satu mesin. Ray memulai instans lokal secara otomatis saat pertama kali Anda memanggil trainer.fit() atau tuner.fit(). Untuk memakai lebih dari satu node, jalankan cluster dan arahkan driver Anda ke sana; kode pelatihan dan penyetelan tidak berubah.

Memulai cluster secara manual:

# Pada node head.
ray start --head --port=6379

Pada setiap node worker, mengarah ke alamat head.
ray start --address='HEADNODEIP:6379'

Lalu dalam skrip Anda, sambungkan ke cluster yang berjalan alih-alih memulai yang lokal:

import ray

ray.init(address="auto")  # bergabung ke cluster yang dimulai di atas

Untuk lingkungan cloud, peluncur cluster Ray membaca file YAML yang menjelaskan tipe node head dan worker serta batas autoscaling, lalu menyiapkan cluster untuk Anda:

ray up cluster.yaml
ray submit cluster.yaml trainscript.py

Intinya, ScalingConfig dan fungsi pelatihan Anda tetap sama. Anda berpindah dari laptop ke cluster 50 node dengan mengubah ke mana skrip menyambung dan berapa banyak worker yang Anda minta, bukan dengan menulis ulang loop-nya.

Integrasi Pelacakan Eksperimen

Ray Tune terintegrasi dengan tracker eksperimen melalui callback, sehingga metrik trial mengalir ke dashboard yang biasa Anda pakai. Dua opsi umum:

from ray.air.integrations.wandb import WandbLoggerCallback
from ray.air.integrations.mlflow import MLflowLoggerCallback
from ray.tune import RunConfig

runconfig = RunConfig(
    callbacks=[
        WandbLoggerCallback(project="ray-train-demo"),
        # atau:
        MLflowLoggerCallback(
            trackinguri="http://localhost:5000",

            experimentname="ray-train-demo",
        ),
    ]
)

tuner = tune.Tuner(
    tunetrainable,

    paramspace=searchspace,

    tuneconfig=TuneConfig(numsamples=20, metric="accuracy", mode="max"),

    runconfig=runconfig,

)

Setiap nilai yang Anda kirim ke ray.train.report dicatat otomatis, dan tiap trial muncul sebagai run terpisah di Weights & Biases atau MLflow.

Toleransi Kegagalan dan Melanjutkan Eksperimen

Job panjang gagal karena alasan biasa: instans spot ditarik kembali, sebuah node kehabisan memori, atau driver terganggu. Ray Train dan Ray Tune menyimpan checkpoint dan state eksperimen sehingga Anda dapat melanjutkan, bukan memulai dari awal.

Atur tempat penyimpanan state dan berapa kali percobaan ulang melalui RunConfig dan FailureConfig:

from ray.train import RunConfig, FailureConfig

runconfig = RunConfig(
    storagepath="s3://my-bucket/ray-experiments",  # atau path lokal bersama

    name="fashionmnist-tune",
    failureconfig=FailureConfig(maxfailures=2),   # ulangi trial gagal dua kali

)

Jika seluruh eksperimen terganggu, pulihkan dari path penyimpanan yang sama dan lanjutkan:

from ray.tune import Tuner

tuner = Tuner.restore(
    path="s3://my-bucket/ray-experiments/fashionmnist-tune",
    trainable=tunetrainable,
    resumeerrored=True,

)
results = tuner.fit()

Agar ini bekerja, path penyimpanan harus dapat dijangkau dari setiap node, itulah sebabnya filesystem bersama atau object store disarankan pada cluster multi-node, bukan direktori lokal di node head.

Praktik Terbaik

• Kembangkan di CPU dulu. Set usegpu=False dan numworkers=2 untuk memvalidasi loop dengan cepat, lalu naikkan skala. Jalur kode terdistribusi berjalan sama pada skala kecil.
• Laporkan pada irama yang wajar. Panggil ray.train.report sekali per epoch atau setiap N langkah, bukan tiap batch, agar overhead koordinasi tetap rendah.
• Perhitungkan ukuran batch efektif. Pelatihan data-parallel mengalikan ukuran batch per worker dengan jumlah worker. Sesuaikan learning rate atau warmup.
• Biarkan ASHA memangkas anggaran. Penghentian dini dengan ASHAScheduler biasanya menemukan konfigurasi bagus dengan komputasi jauh lebih sedikit dibanding menjalankan setiap trial sampai selesai.
• Cari hal yang tepat. Tetapkan parameter yang tidak penting dan cari hanya yang penting; ruang yang lebih kecil konvergen lebih cepat.
• Gunakan path penyimpanan bersama. Pada cluster, arahkan storagepath ke object store atau filesystem bersama agar checkpoint dan pemulihan bekerja lintas node.
• Sesuaikan jumlah trial dengan perangkat keras. Kalikan numsamples dengan worker per trial untuk memperkirakan puncak kebutuhan resource sebelum meluncurkan.

Kesimpulan dan Poin Penting

Ray Train dan Ray Tune mencakup dua masalah skala yang pada akhirnya ditemui pelatihan satu mesin. Ray Train mengubah loop PyTorch standar menjadi pelatihan data-parallel terdistribusi dengan tiga suntingan kecil (preparemodel, preparedataloader, dan ray.train.report), sementara ScalingConfig mengontrol jumlah worker dan GPU tanpa menyentuh loop. Ray Tune menjalankan banyak konfigurasi secara paralel, mendukung pencarian lebih cerdas melalui OptunaSearch atau HyperOptSearch, dan menghemat komputasi dengan scheduler seperti ASHA. Karena keduanya dapat disatukan, Anda dapat menyetel trainer terdistribusi cukup dengan membungkusnya dalam Tuner.

Poin penting:

• Konversi loop pelatihan dengan preparemodel, preparedataloader, dan ray.train.report; skalakan dengan ScalingConfig.
• Gunakan hasil TorchTrainer.fit() untuk mengambil metrik dan checkpoint terbaik.
• Definisikan ruang pencarian, jalankan tune.Tuner dengan TuneConfig, dan analisis ResultGrid dengan getbestresult.
• Tambahkan OptunaSearch/HyperOptSearch untuk efisiensi sampel dan ASHAScheduler untuk penghentian dini.
• Bungkus TorchTrainer dalam Tuner untuk menyetel pelatihan terdistribusi; kode yang sama berjalan lokal maupun di cluster.
• Atur storagepath dan FailureConfig agar proses panjang dapat dilanjutkan setelah kegagalan.