数据并行
eg1. nn.parallel
# Replicate module to devices in device_ids
replicas = nn.parallel.replicate(module, device_ids)
# Distribute input to devices in device_ids
inputs = nn.parallel.scatter(input, device_ids)
# Apply the models to corresponding inputs
outputs = nn.parallel.parallel_apply(replicas, inputs)
# Gather result from all devices to output_device
result = nn.parallel.gather(outputs, output_device)

eg2. torch.nnDataParallel(module,device_ids=None,ouput_device=None,dim=0)
# device_ids指定在哪些GPU上进行优化，output_devices指定输出到哪个GPU上,如果不设定要使用的device_ids,程序会自动找可用的所有显卡用于训练， os.environ['CUDA_VISIBLE_DEVICES'] == ’‘限定GPU
new_net= nn.DataParallel(net, device_ids=[0, 1])
output= new_net(input)

并行数据加载
设置'num_workers>0'使用多个子进程进行数据加载

import torch
import torch.nn as nn
import argparse
from torch.autograd import Variable
from torch.utils.data import Dataset, DataLoader
import os
from torch.utils.data.distributed import DistributedSampler
 
# 1)添加参数  --local_rank
'''
每个进程分配一个 local_rank 参数, 表示当前进程在当前主机上的编号。这个参数是torch.distributed.launch传递过来的，我们设置位置参数来接受，local_rank代表当前程序进程使用的GPU标号
'''
parser = argparse.ArgumentParser()
parser.add_argument('--local_rank', default=-1, type=int,
                    help='node rank for distributed training')
args = parser.parse_args()
 
# 通过args接收 local_rank
local_rank = args.local_rank
 
# 通过 get_rank() 得到 local_rank，最好放在初始化之后使用
local_rank = torch.distributed.get_rank()
 
# 2)********************************************************
# 初始化使用nccl后端
torch.distributed.init_process_group(backend="nccl")
 
# 3)********************************************************
# 获取 local_rank
local_rank = torch.distributed.get_rank()
#配置每个进程的 GPU， 根据local_rank来设定当前使用哪块GPU
torch.cuda.set_device(local_rank)
device = torch.device("cuda", local_rank)
 
# 4)********************************************************
# 自己的数据获取
dataset = MyDataset(input_size, data_size)
 
# 使用 DistributedSampler
'''
使用 DistributedSampler 对数据集进行划分。它能帮助我们将每个 batch 划分成几个 partition，在当前进程中只需要获取和 rank 对应的那个 partition 进行训练
#注意 testset不用sampler
'''
train_sampler = torch.utils.data.distributed.DistributedSampler(dataset)
 
trainloader = DataLoader(dataset=dataset,
                         pin_memory=true,
                         shuffle=(train_sampler is None),   # 使用分布式训练 shuffle 应该设置为 False
                         batch_size=args.batch_size,
                         num_workers=args.workers,
                         sampler=train_sampler)
 
 
 
# 5)********************************************************
model = Model()
# 把模型移到对应的gpu
# 定义并把模型放置到单独的GPU上，需要在调用`model=DDP(model)`前做
model.to(device)
 
# 引入SyncBN，这句代码，会将普通BN替换成SyncBN。
model = torch.nn.SyncBatchNorm.convert_sync_batchnorm(model)
 
# GPU 数目大于 1 才有必要分布式训练
if torch.cuda.device_count() > 1:
    model = torch.nn.parallel.DistributedDataParallel(model,
                                                      device_ids=[local_rank],
                                                      output_device=local_rank)
 
 
# 6)********************************************************
 
for epoch in range(num_epochs):
    # 设置sampler的epoch，DistributedSampler需要这个来维持各个进程之间的相同随机数种子
    trainloader.sampler.set_epoch(epoch)
    # 后面这部分，则与原来完全一致了。
    for data, label in trainloader:
        prediction = model(data)
        loss = loss_fn(prediction, label)
        loss.backward()
        optimizer = optim.SGD(ddp_model.parameters(), lr=0.001)
        optimizer.step()

eg
CUDA_VISIBLE_DEVICES=0,1,2,3 python -m torch.distributed.launch --nproc_per_node=4 main.py --{args}
# --nproc_per_node一般设定为当前主机可用的GPU数目

Eg
from transformers import BertForSequenceClassification, AdamW, get_scheduler
from datasets import load_dataset
from accelerate import Accelerator

# 初始化 Accelerator
accelerator = Accelerator()

# 加载数据集
dataset = load_dataset("glue", "mrpc")
train_dataset = dataset["train"]

# 定义模型和优化器
model = BertForSequenceClassification.from_pretrained("bert-base-uncased")
optimizer = AdamW(model.parameters(), lr=5e-5)

# 准备数据加载器
train_dataloader = torch.utils.data.DataLoader(train_dataset, batch_size=8)

# 准备模型、数据和优化器
model, optimizer, train_dataloader = accelerator.prepare(model, optimizer, train_dataloader)

# 训练循环
for epoch in range(3):
    model.train()
    for batch in train_dataloader:
        outputs = model(**batch)
        loss = outputs.loss
        accelerator.backward(loss)
        optimizer.step()
        optimizer.zero_grad()
    print(f'Epoch {epoch+1}, Loss: {loss.item()}')