|
| 1 | +# Copyright Contributors to the Pyro project. |
| 2 | +# SPDX-License-Identifier: Apache-2.0 |
| 3 | + |
| 4 | +# Distributed training via Pytorch Lightning. |
| 5 | +# |
| 6 | +# This tutorial demonstrates how to distribute SVI training across multiple |
| 7 | +# machines (or multiple GPUs on one or more machines) using the PyTorch Lightning |
| 8 | +# library. PyTorch Lightning enables data-parallel training by aggregating stochastic |
| 9 | +# gradients at each step of training. We focus on integration between PyTorch Lightning and Pyro. |
| 10 | +# For further details on distributed computing with PyTorch Lightning, see |
| 11 | +# https://lightning.ai/docs/pytorch/latest |
| 12 | +# |
| 13 | +# This assumes you have installed pytorch lightning, e.g. via |
| 14 | +# pip install pyro-ppl[lightning] |
| 15 | + |
| 16 | +import argparse |
| 17 | + |
| 18 | +import pytorch_lightning as pl |
| 19 | +import torch |
| 20 | + |
| 21 | +import pyro |
| 22 | +import pyro.distributions as dist |
| 23 | +from pyro.infer import Trace_ELBO |
| 24 | +from pyro.infer.autoguide import AutoNormal |
| 25 | +from pyro.nn import PyroModule |
| 26 | + |
| 27 | + |
| 28 | +# We define a model as usual, with no reference to Pytorch Lightning. |
| 29 | +# This model is data parallel and supports subsampling. |
| 30 | +class Model(PyroModule): |
| 31 | + def __init__(self, size): |
| 32 | + super().__init__() |
| 33 | + self.size = size |
| 34 | + |
| 35 | + def forward(self, covariates, data=None): |
| 36 | + coeff = pyro.sample("coeff", dist.Normal(0, 1)) |
| 37 | + bias = pyro.sample("bias", dist.Normal(0, 1)) |
| 38 | + scale = pyro.sample("scale", dist.LogNormal(0, 1)) |
| 39 | + |
| 40 | + # Since we'll use a distributed dataloader during training, we need to |
| 41 | + # manually pass minibatches of (covariates,data) that are smaller than |
| 42 | + # the full self.size. In particular we cannot rely on pyro.plate to |
| 43 | + # automatically subsample, since that would lead to all workers drawing |
| 44 | + # identical subsamples. |
| 45 | + with pyro.plate("data", self.size, len(covariates)): |
| 46 | + loc = bias + coeff * covariates |
| 47 | + return pyro.sample("obs", dist.Normal(loc, scale), obs=data) |
| 48 | + |
| 49 | + |
| 50 | +# We define an ELBO loss, a PyTorch optimizer, and a training step in our PyroLightningModule. |
| 51 | +# Note that we are using a PyTorch optimizer instead of a Pyro optimizer and |
| 52 | +# we are using ``training_step`` instead of Pyro's SVI machinery. |
| 53 | +class PyroLightningModule(pl.LightningModule): |
| 54 | + def __init__(self, loss_fn: pyro.infer.elbo.ELBOModule, lr: float): |
| 55 | + super().__init__() |
| 56 | + self.loss_fn = loss_fn |
| 57 | + self.model = loss_fn.model |
| 58 | + self.guide = loss_fn.guide |
| 59 | + self.lr = lr |
| 60 | + self.predictive = pyro.infer.Predictive( |
| 61 | + self.model, guide=self.guide, num_samples=1 |
| 62 | + ) |
| 63 | + |
| 64 | + def forward(self, *args): |
| 65 | + return self.predictive(*args) |
| 66 | + |
| 67 | + def training_step(self, batch, batch_idx): |
| 68 | + """Training step for Pyro training.""" |
| 69 | + loss = self.loss_fn(*batch) |
| 70 | + # Logging to TensorBoard by default |
| 71 | + self.log("train_loss", loss) |
| 72 | + return loss |
| 73 | + |
| 74 | + def configure_optimizers(self): |
| 75 | + """Configure an optimizer.""" |
| 76 | + return torch.optim.Adam(self.loss_fn.parameters(), lr=self.lr) |
| 77 | + |
| 78 | + |
| 79 | +def main(args): |
| 80 | + # Create a model, synthetic data, a guide, and a lightning module. |
| 81 | + pyro.set_rng_seed(args.seed) |
| 82 | + pyro.settings.set(module_local_params=True) |
| 83 | + model = Model(args.size) |
| 84 | + covariates = torch.randn(args.size) |
| 85 | + data = model(covariates) |
| 86 | + guide = AutoNormal(model) |
| 87 | + loss_fn = Trace_ELBO()(model, guide) |
| 88 | + training_plan = PyroLightningModule(loss_fn, args.learning_rate) |
| 89 | + |
| 90 | + # Create a dataloader. |
| 91 | + dataset = torch.utils.data.TensorDataset(covariates, data) |
| 92 | + dataloader = torch.utils.data.DataLoader(dataset, batch_size=args.batch_size) |
| 93 | + |
| 94 | + # All relevant parameters need to be initialized before ``configure_optimizer`` is called. |
| 95 | + # Since we used AutoNormal guide our parameters have not be initialized yet. |
| 96 | + # Therefore we initialize the model and guide by running one mini-batch through the loss. |
| 97 | + mini_batch = dataset[: args.batch_size] |
| 98 | + loss_fn(*mini_batch) |
| 99 | + |
| 100 | + # Run stochastic variational inference using PyTorch Lightning Trainer. |
| 101 | + trainer = pl.Trainer.from_argparse_args(args) |
| 102 | + trainer.fit(training_plan, train_dataloaders=dataloader) |
| 103 | + |
| 104 | + |
| 105 | +if __name__ == "__main__": |
| 106 | + assert pyro.__version__.startswith("1.8.4") |
| 107 | + parser = argparse.ArgumentParser( |
| 108 | + description="Distributed training via PyTorch Lightning" |
| 109 | + ) |
| 110 | + parser.add_argument("--size", default=1000000, type=int) |
| 111 | + parser.add_argument("--batch_size", default=100, type=int) |
| 112 | + parser.add_argument("--learning_rate", default=0.01, type=float) |
| 113 | + parser.add_argument("--seed", default=20200723, type=int) |
| 114 | + parser = pl.Trainer.add_argparse_args(parser) |
| 115 | + args = parser.parse_args() |
| 116 | + main(args) |
0 commit comments