get_logging_handler()でDEイベントをログ出力

PyTorch Distributed Elastic: イベントと get_logging_handler()

その機能の一つがイベントです。 イベントは、訓練中に発生する重要な出来事を通知するメカニズムです。 例えば、訓練の開始、エポックの終了、チェックポイントの保存などがイベントとして発生します。

DE は、イベントを購読し、イベント発生時に処理を行うための API を提供しています。 その API の一つが torch.distributed.elastic.events.get_logging_handler() です。

get_logging_handler() は、イベントを購読し、ログ出力に記録するためのハンドラを取得します。 ハンドラは、イベント発生時に、イベントの内容をログファイルに書き込みます。

get_logging_handler() は、以下の引数を受け取ります。

• name: ハンドラの名前。 複数のハンドラを登録する場合、名前を指定することで特定のハンドラを取得できます。
• formatter: ログ出力のフォーマットを指定します。 デフォルトでは、logging.Formatter() が使用されます。
• stream: ログ出力を出力するストリームを指定します。 デフォルトでは、sys.stdout が使用されます。

以下のコードは、get_logging_handler() を使用して、イベントを購読し、ログ出力に記録する例です。

import torch.distributed.elastic.events as events

handler = events.get_logging_handler(
    name="my_handler",
    formatter=logging.Formatter("%(asctime)s %(message)s"),
    stream=sys.stdout,
)

events.subscribe(handler)

# イベント発生時に、ハンドラが呼び出され、ログ出力に記録される

イベントの種類

DE は、以下のイベントを発生します。

• TRAIN_STARTED: 訓練の開始時に発生します。
• EPOCH_STARTED: エポックの開始時に発生します。
• EPOCH_COMPLETED: エポックの終了時に発生します。
• CHECKPOINT_SAVED: チェックポイントが保存された時に発生します。
• TRAIN_COMPLETED: 訓練の終了時に発生します。

torch.distributed.elastic.events.get_logging_handler() は、DE のイベントを購読し、ログ出力に記録するためのハンドラを取得します。 ハンドラを使用することで、訓練中の重要な出来事を記録することができます。

PyTorch Distributed Elastic サンプルコード

DE の機能を理解し、使い始めるために、いくつかのサンプルコードを紹介します。

MNIST 分類

このサンプルコードは、MNIST データセットを使用して、DE でシンプルなニューラルネットワークを訓練する方法を示します。

import torch
import torch.distributed.elastic as de

# MNIST データセットの読み込み
train_dataset = torchvision.datasets.MNIST(
    root="./data",
    train=True,
    download=True,
    transform=torchvision.transforms.ToTensor(),
)

# DataLoader の作成
train_loader = torch.utils.data.DataLoader(
    train_dataset,
    batch_size=64,
    shuffle=True,
)

# モデルの定義
model = torch.nn.Sequential(
    torch.nn.Linear(784, 128),
    torch.nn.ReLU(),
    torch.nn.Linear(128, 10),
    torch.nn.LogSoftmax(dim=1),
)

# オプティマイザの定義
optimizer = torch.optim.SGD(model.parameters(), lr=0.01)

# DE の初期化
de.init_process_group(
    backend="nccl",
    init_method="env://",
    world_size=4,
    rank=0,
)

# 訓練ループ
for epoch in range(10):
    for batch_idx, (data, target) in enumerate(train_loader):
        # データを GPU に転送
        data = data.cuda()
        target = target.cuda()

        # 予測
        output = model(data)

        # 損失の計算
        loss = torch.nn.functional.nll_loss(output, target)

        # 勾配の計算
        optimizer.zero_grad()
        loss.backward()

        # パラメータの更新
        optimizer.step()

        # ログ出力
        if batch_idx % 100 == 0:
            print(f"Epoch: {epoch}, Batch: {batch_idx}, Loss: {loss.item()}")

# モデルの保存
torch.save(model.state_dict(), "./model.pth")

ImageNet 分類

このサンプルコードは、ImageNet データセットを使用して、DE で ResNet モデルを訓練する方法を示します。

import torch
import torch.distributed.elastic as de
import torchvision

# ImageNet データセットの読み込み
train_dataset = torchvision.datasets.ImageNet(
    root="./data",
    split="train",
    download=True,
    transform=torchvision.transforms.Compose(
        [
            torchvision.transforms.RandomResizedCrop(224),
            torchvision.transforms.RandomHorizontalFlip(),
            torchvision.transforms.ToTensor(),
            torchvision.transforms.Normalize(
                mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]
            ),
        ]
    ),
)

# DataLoader の作成
train_loader = torch.utils.data.DataLoader(
    train_dataset,
    batch_size=64,
    shuffle=True,
)

# モデルの定義
model = torchvision.models.resnet50(pretrained=False)

# オプティマイザの定義
optimizer = torch.optim.SGD(model.parameters(), lr=0.01)

# DE の初期化
de.init_process_group(
    backend="nccl",
    init_method="env://",
    world_size=4,
    rank=0,
)

# 訓練ループ
for epoch in range(100):
    for batch_idx, (data, target) in enumerate(train_loader):
        # データを GPU に転送
        data = data.cuda()
        target = target.cuda()

        # 

PyTorch Distributed Elastic サンプルコード：その他の方法

Horovod は、DE と統合できるオープンソースの分散訓練ライブラリです。 Horovod を使用すると、DE の機能をさらに拡張し、より高度な分散訓練を行うことができます。

import torch
import torch.distributed.elastic as de
import horovod.torch as hvd

# Horovod の初期化
hvd.init()

# DE の初期化
de.init_process_group(
    backend="nccl",
    init_method="env://",
    world_size=hvd.size(),
    rank=hvd.rank(),
)

# 訓練ループ
for epoch in range(100):
    for batch_idx, (data, target) in enumerate(train_loader):
        # データを GPU に転送
        data = data.cuda()
        target = target.cuda()

        # 予測
        output = model(data)

        # 損失の計算
        loss = torch.nn.functional.nll_loss(output, target)

        # 勾配の計算
        optimizer.zero_grad()
        loss.backward()

        # Horovod を使用して勾配を同期
        hvd.all_reduce(loss)

        # パラメータの更新
        optimizer.step()

        # ログ出力
        if batch_idx % 100 == 0:
            print(f"Epoch: {epoch}, Batch: {batch_idx}, Loss: {loss.item()}")

# モデルの保存
torch.save(model.state_dict(), "./model.pth")

PySpark は、Apache Spark 上で Python を実行するためのフレームワークです。 PySpark を使用して、DE と Spark の機能を組み合わせ、大規模なデータセットに対して分散訓練を行うことができます。

import torch
import torch.distributed.elastic as de
from pyspark.sql import SparkSession

# SparkSession の作成
spark = SparkSession.builder.appName("PySpark DE").getOrCreate()

# DE の初期化
de.init_process_group(
    backend="nccl",
    init_method="env://",
    world_size=spark.conf.get("spark.executor.instances"),
    rank=spark.sparkContext.getRank(),
)

# 訓練データの読み込み
rdd = spark.sparkContext.parallelize(train_dataset)

# 訓練ループ
for epoch in range(100):
    # データを RDD から取得
    data, target = rdd.take(batch_size)

    # データを GPU に転送
    data = data.cuda()
    target = target.cuda()

    # 予測
    output = model(data)

    # 損失の計算
    loss = torch.nn.functional.nll_loss(output, target)

    # 勾配の計算
    optimizer.zero_grad()
    loss.backward()

    # DE を使用して勾配を同期
    de.all_reduce(loss)

    # パラメータの更新
    optimizer.step()

    # ログ出力
    if batch_idx % 100 == 0:
        print(f"Epoch: {epoch}, Batch: {batch_idx}, Loss: {loss.item()}")

# モデルの保存
torch.save(model.state_dict(), "./model.pth")

Kubernetes は、コンテナ化されたアプリケーションをデプロイおよび管理するためのオープンソースのプラットフォームです。 Kubernetes を使用して、DE を複数のノードにスケールアウトし、大規模な分散訓練を行うことができます。

詳細は、PyTorch Distributed Elastic のドキュメントを参照してください。

DE は、PyTorch で大規模な分散訓練を行うための強力なツールです。 上記のサンプルコードは、DE を使用して分散訓練を行うための出発点となります。