Object Storage Service:OSS Connector for AI/ML の使用開始

デプロイメント環境

• オペレーティングシステム: Linux x86-64

• glibc: >=2.17

• Python: 3.8-3.12

• PyTorch: >=2.0

• OSS チェックポイント機能を使用するには、Linux カーネルが userfaultfd をサポートしている必要があります。

  説明

  たとえば、Ubuntu システムでは、sudo grep CONFIG_USERFAULTFD /boot/config-$(uname -r) コマンドを実行して、Linux カーネルが userfaultfd をサポートしているかどうかを確認できます。コマンドが CONFIG_USERFAULTFD=y を返した場合、カーネルはそれをサポートしています。コマンドが CONFIG_USERFAULTFD=n を返した場合、カーネルはそれをサポートしておらず、OSS チェックポイント機能は使用できません。

クイックインストール

次の例では、Python 3.12 用に OSS Connector for AI/ML をインストールする方法について説明します。

1. Linux オペレーティングシステム、または Linux ベースのイメージから作成されたコンテナーで、pip3.12 install osstorchconnector コマンドを実行して OSS Connector for AI/ML をインストールできます。

   pip3.12 install osstorchconnector

2. pip3.12 show osstorchconnector コマンドを実行して、インストールが成功したかどうかを確認できます。

   pip3.12 show osstorchconnector

   osstorchconnector のバージョン情報が返された場合、インストールは成功です。

   

構成

1. アクセス資格情報の設定ファイルを作成できます。

   mkdir -p /root/.alibabacloud && touch /root/.alibabacloud/credentials

2. アクセス資格情報の設定を追加してファイルを保存できます。

   例の <Access-key-id> と <Access-key-secret> を、Resource Access Management (RAM) ユーザーの AccessKey ID と AccessKey シークレットに置き換えます。AccessKey ID と AccessKey シークレットの作成方法の詳細については、「AccessKey の作成」をご参照ください。設定項目と一時的なアクセス資格情報の使用方法の詳細については、「アクセス資格情報の設定」をご参照ください。

   {
     "AccessKeyId": "LTAI************************",
     "AccessKeySecret": "At32************************"
   }

3. OSS Connector の設定ファイルを作成できます。

   mkdir -p /etc/oss-connector/ && touch /etc/oss-connector/config.json

4. OSS Connector の設定を追加してファイルを保存できます。設定項目の詳細については、「OSS Connector の設定」をご参照ください。

   ほとんどの場合、次のデフォルト設定を使用できます。

   {
       "logLevel": 1,
       "logPath": "/var/log/oss-connector/connector.log",
       "auditPath": "/var/log/oss-connector/audit.log",
       "datasetConfig": {
           "prefetchConcurrency": 24,
           "prefetchWorker": 2
       },
       "checkpointConfig": {
           "prefetchConcurrency": 24,
           "prefetchWorker": 4,
           "uploadConcurrency": 64
       }
   }

例

次の例は、PyTorch を使用して手書き認識モデルを作成する方法を示しています。このモデルは、OssMapDataset で構築された MNIST データセットを使用し、OssCheckpoint を使用してモデルのチェックポイントを保存およびロードします。

import io
import torch
import torch.nn as nn
import torch.optim as optim
import torchvision.transforms as transforms
from PIL import Image
from torch.utils.data import DataLoader
from osstorchconnector import OssMapDataset
from osstorchconnector import OssCheckpoint

# ハイパーパラメーターを定義します。
EPOCHS = 1
BATCH_SIZE = 64
LEARNING_RATE = 0.001
CHECKPOINT_READ_URI = "oss://you_bucketname/epoch.ckpt"  # OSS からチェックポイントを読み取るための URI。
CHECKPOINT_WRITE_URI = "oss://you_bucketname/epoch.ckpt" # チェックポイントを OSS に保存するための URI。
ENDPOINT = "oss-cn-hangzhou-internal.aliyuncs.com"       # OSS にアクセスするための内部エンドポイント。このエンドポイントを使用するには、ECS インスタンスと OSS バケットが同じリージョンにある必要があります。
CONFIG_PATH = "/etc/oss-connector/config.json"           # OSS Connector 設定ファイルへのパス。
CRED_PATH = "/root/.alibabacloud/credentials"            # アクセス資格情報設定ファイルへのパス。
OSS_URI = "oss://you_bucketname/mnist/"                  # OSS 内のバケットリソースの URI。

# トレーニング中にチェックポイントを OSS に保存し、OSS から読み取るための OssCheckpoint オブジェクトを作成します。
checkpoint = OssCheckpoint(endpoint=ENDPOINT, cred_path=CRED_PATH, config_path=CONFIG_PATH)

# シンプルな畳み込みニューラルネットワークを定義します。
class SimpleCNN(nn.Module):
    def __init__(self):
        super(SimpleCNN, self).__init__()
        self.conv1 = nn.Conv2d(1, 32, kernel_size=3, stride=1, padding=1)
        self.conv2 = nn.Conv2d(32, 64, kernel_size=3, stride=1, padding=1)
        # アダプティブプーリングを使用してサイズ処理を簡素化します。
        self.adaptive_pool = nn.AdaptiveAvgPool2d((7, 7))
        self.fc1 = nn.Linear(64 * 7 * 7, 128)
        self.fc2 = nn.Linear(128, 10)

    def forward(self, x):
        x = nn.ReLU()(self.conv1(x)) 
        x = nn.MaxPool2d(2)(x)
        x = nn.ReLU()(self.conv2(x))
        x = nn.MaxPool2d(2)(x)
        x = self.adaptive_pool(x)
        x = x.view(x.size(0), -1)
        x = nn.ReLU()(self.fc1(x))
        x = self.fc2(x)
        return x

# データを前処理します。
trans = transforms.Compose([
    transforms.Resize(256),
    transforms.CenterCrop(224),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.5], std=[0.5])
])

def transform(object):
    try:
        img = Image.open(io.BytesIO(object.read())).convert('L')
        val = trans(img)
    except Exception as e:
        raise e
    
    # オブジェクトパスからラベルを抽出します。パスのフォーマットは oss://bucket/mnist/label/filename であると仮定します。
    # 実際のデータセットの構造に基づいてラベル抽出ロジックを調整します。
    try:
        label = int(object.name.split('/')[-2])  # 最後から 2 番目のパスセグメントをラベルとして抽出します。
    except (ValueError, IndexError):
        label = 0  # デフォルトのラベル。データセットの構造に基づいてこれを調整します。
    
    return val, torch.tensor(label)

# OssMapDataset データセットをロードします。
train_dataset = OssMapDataset.from_prefix(OSS_URI, endpoint=ENDPOINT, transform=transform, cred_path=CRED_PATH, config_path=CONFIG_PATH)
train_loader = DataLoader(train_dataset, batch_size=BATCH_SIZE, num_workers=32, prefetch_factor=2, shuffle=True)

# モデル、損失関数、オプティマイザーを初期化します。
model = SimpleCNN()
criterion = nn.CrossEntropyLoss()  
optimizer = optim.Adam(model.parameters(), lr=LEARNING_RATE)  

# モデルをトレーニングします。
for epoch in range(EPOCHS):
    for i, (images, labels) in enumerate(train_loader):
        optimizer.zero_grad()  
        outputs = model(images)  
        loss = criterion(outputs, labels)  
        loss.backward()  
        optimizer.step()  
        if (i + 1) % 100 == 0:
            print(f'Epoch [{epoch + 1}/{EPOCHS}], Step [{i + 1}/{len(train_loader)}], Loss: {loss.item():.4f}')
    
    # OssCheckpoint オブジェクトを使用してチェックポイントを保存します。
    with checkpoint.writer(CHECKPOINT_WRITE_URI) as writer:
        torch.save(model.state_dict(), writer)
        print("-------------------------")
        print("Checkpoint saved")
        print(model.state_dict())

# OssCheckpoint オブジェクトを使用してチェックポイントを読み取ります。
try:
    with checkpoint.reader(CHECKPOINT_READ_URI) as reader:
        state_dict = torch.load(reader)
    
    # モデルをロードします。
    model = SimpleCNN()
    model.load_state_dict(state_dict)
    model.eval()
    print("Checkpoint loaded successfully")
except Exception as e:
    print(f"Failed to load checkpoint: {e}")
    # 最初からトレーニングするか、別のチェックポイントを使用するかを選択できます。