このトピックでは、OSS Connector for AI/ML を使用してデータモデルを効率的に作成およびトレーニングする方法について説明します。
デプロイ環境
オペレーティングシステム： Linux x86-64
glibc： >=2.17
Python： 3.8-3.12
PyTorch： >=2.0
OssCheckpoint 機能を使用するには、Linux カーネルが userfaultfd をサポートしている必要があります。
説明たとえば、Ubuntu システムでは、
sudo grep CONFIG_USERFAULTFD /boot/config-$(uname -r)コマンドを実行して、Linux カーネルが userfaultfd をサポートしているかどうかを確認できます。コマンドがCONFIG_USERFAULTFD=yを返す場合、カーネルはサポートしています。コマンドがCONFIG_USERFAULTFD=nを返す場合、カーネルはサポートしていないため、OssCheckpoint 機能は使用できません。
クイックインストール
以下の例は、Python 3.12 を使用して AI/ML 用 OSS Connector をインストールする方法を示しています。
Linux オペレーティングシステム、または Linux ベースのイメージから作成されたコンテナで、
pip3.12 install osstorchconnectorコマンドを実行して OSS Connector for AI/ML をインストールします。pip3.12 install osstorchconnector-
pip3.12 show osstorchconnectorコマンドを実行して、インストールが成功したかどうかを確認できます。pip3.12 show osstorchconnectorosstorchconnector のバージョン情報が返された場合、インストールは成功しています。
Name: osstorchconnector Version: 1.0.0rc1 Summary: OSS connector for AI/ML Home-page: Author: Author-email: License: Location: /usr/local/lib/python3.12/dist-packages Requires: torch Required-by:
設定
アクセス認証情報の設定ファイルを作成します。
mkdir -p /root/.alibabacloud && touch /root/.alibabacloud/credentialsアクセス認証情報の設定を追加し、ファイルを保存します。
例の
<Access-key-id>と<Access-key-secret>を、RAM ユーザーの AccessKey ID と AccessKey シークレットに置き換えてください。AccessKey ID と AccessKey シークレットの作成方法の詳細については、「Create an AccessKey」をご参照ください。設定項目と一時的なアクセス認証情報の使用方法の詳細については、「Configure access credentials」をご参照ください。{ "AccessKeyId": "LTAI************************", "AccessKeySecret": "At32************************" }OSS Connector の設定ファイルを作成します。
mkdir -p /etc/oss-connector/ && touch /etc/oss-connector/config.jsonOSS Connector の設定を追加し、ファイルを保存します。設定項目の詳細については、「Configure 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" # リージョンの内部エンドポイント。このエンドポイントを使用するには、ECS インスタンスと OSS バケットが同じリージョンにある必要があります。
CONFIG_PATH = "/etc/oss-connector/config.json" # OSS Connector 設定ファイルへのパス。
CRED_PATH = "/root/.alibabacloud/credentials" # アクセス認証情報設定ファイルへのパス。
OSS_URI = "oss://you_bucketname/mnist/" # OSS バケット内のリソースへのパス。
# OssCheckpoint オブジェクトを作成し、トレーニング中にチェックポイントを OSS に保存およびロードします。
checkpoint = OssCheckpoint(endpoint=ENDPOINT, cred_path=CRED_PATH, config_path=CONFIG_PATH)
# シンプルな畳み込みニューラルネットワーク (CNN) を定義します。
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.key.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}")
# 最初からトレーニングするか、別のチェックポイントを使用できます。