Platform For AI:OSSの使用

マウント方法

デフォルト設定の制限事項

詳細設定 パラメータを未設定のままにした場合、デフォルト設定が適用されます。デフォルト設定には次の制限があります：

• OSS ファイルの読み取りを高速化するため、OSS のマウント時にメタデータ (ディレクトリとファイルリスト) がキャッシュされます。

  分散トレーニングにおいて、複数のノードが同じディレクトリを作成しようとする場合、メタデータキャッシュにより各ノードが作成を試みる可能性があります。1 つのノードのみが成功し、他のノードはエラーを返します。

• デフォルトでは、OSS MultiPart API を使用してファイルが作成されます。オブジェクトは、書き込み操作が完了した後にのみ OSS 上で可視になります。

• 同じファイルに対する同時読み書き操作はサポートされていません。

• ファイルに対するランダム書き込み操作はサポートされていません。

一般的な JindoFuse 設定

ユースケースに応じて、詳細設定で JindoFuse パラメータをカスタマイズすることもできます。

このトピックでは、いくつかの一般的なシナリオに対する JindoFuse 設定の推奨事項を提供します。これらの設定は、すべてのワークロードに対して最適なパフォーマンスを提供するとは限りません。より柔軟な設定オプションについては、「JindoFuse ユーザーガイド」をご参照ください。

• クイック読み取り /書き込み: クイック読み取りと書き込みを保証します。 ただし、同時読み取りまたは書き込み中にデータの不整合が発生する可能性があります。 トレーニングデータとモデルをこのモードのマウントパスにマウントできます。 このモードのマウントパスを作業ディレクトリとして使用しないことを推奨します。

  {
    "fs.oss.download.thread.concurrency": "Twice the number of CPU cores",
    "fs.oss.upload.thread.concurrency": "Twice the number of CPU cores",
    "fs.jindo.args": "-oattr_timeout=3 -oentry_timeout=0 -onegative_timeout=0 -oauto_cache -ono_symlink"
  }
  

• 増分読み取り /書き込み: 増分書き込み中のデータの一貫性を保証します。 元のデータを上書きすると、データの不整合が発生する可能性があります。 読み取り速度はやや遅いです。 このモードを使用して、モデル重みファイルをトレーニングデータ用に保存できます。

  {
    "fs.oss.upload.thread.concurrency": "Twice the number of CPU cores",
    "fs.jindo.args": "-oattr_timeout=3 -oentry_timeout=0 -onegative_timeout=0 -oauto_cache -ono_symlink"
  }
  

• 一貫性のある読み取り /書き込み: 同時読み取りまたは同時書き込み中のデータの一貫性を確保し、高いデータの一貫性が必要で、迅速な読み取りを必要としないシナリオに適しています。 このモードを使用して、プロジェクトのコードを保存できます。

  {
    "fs.jindo.args": "-oattr_timeout=0 -oentry_timeout=0 -onegative_timeout=0 -oauto_cache -ono_symlink"
  }
  

• 読み取り専用: 読み取りのみを許可します。 このモードを使用して、パブリックデータセットをマウントできます。

  {
    "fs.oss.download.thread.concurrency": "Twice the number of CPU cores",
    "fs.jindo.args": "-oro -oattr_timeout=7200 -oentry_timeout=7200 -onegative_timeout=7200 -okernel_cache -ono_symlink"
  }

その他の一般的な設定操作には次のものがあります：

マウント方法

一般的な ossfs 設定

[詳細設定] では、fs.ossfs.args を使用して詳細パラメーターを設定できます。複数のパラメーターはカンマ (,) で区切ります。これらのパラメーターの詳細については、「ossfs 2.0」をご参照ください。以下に、いくつかの一般的なユースケースを示します。

• ジョブ中の静的データソース：ジョブ中に変更されないファイルについては、キャッシュのタイムアウトを長く設定してメタデータリクエストを削減します。一般的なユースケースは、既存ファイルの一括読み取りと、処理後の新しいファイルセットの生成です。

  {
      "mountType":"ossfs",
      "fs.ossfs.args": "-oattr_timeout=7200" 
  }

• 高速読み書き：メタデータキャッシュのタイムアウトを短くして、キャッシュ効率とデータの鮮度のバランスを取ります。

  {
      "mountType":"ossfs",
      "fs.ossfs.args": "-oattr_timeout=3, -onegative_timeout=0"
  }

• 分散ジョブのための一貫性のある読み書き：デフォルトでは、ossfs はメタデータキャッシュに基づいてファイルデータを更新します。複数のノード間で一貫したビューを確保するには、次の設定を使用します。

  {   
      "mountType":"ossfs",
      "fs.ossfs.args": "-onegative_timeout=0, -oclose_to_open"
  }

• DLC または DSW でのファイル開きすぎによるメモリ不足 (OOM)：DLC と DSW でジョブの並行性が高いと、多数のファイルが同時に開かれ、メモリ不足 (OOM) の問題が発生する可能性があります。次の設定は、メモリプレッシャーを軽減するのに役立ちます。

  {
      "mountType":"ossfs",
      "fs.ossfs.args": "-oreaddirplus=false, -oinode_cache_eviction_threshold=300000"
  }

• 大きなファイルの書き込み失敗：-oupload_buffer_size は、マルチパートアップロードのバッファサイズ (バイト単位) を設定します。このパラメーターは、書き込み可能な最大ファイルサイズを決定し、upload_buffer_size * 10000 として計算されます。

  デフォルトでは、ossfs 2.0 は 8 MiB のパートサイズを使用し、書き込み可能な最大ファイルサイズは 78.125 GiB に制限されます。この制限を超えるファイルを書き込むと、操作は失敗します。サポートされる最大ファイルサイズを増やすには、-oupload_buffer_size オプションを設定してパートサイズを大きくします。たとえば、パートサイズを 32 MiB (33,554,432 バイト) に設定すると、最大ファイルサイズは 312.5 GiB になります。-oupload_buffer_size を大きくすると、より多くのメモリが消費されることに注意してください。-total_mem_limit を設定することで、メモリ使用量を制御できます。詳細については、「マウントオプション」をご参照ください。

  {
      "mountType":"ossfs",
      "fs.ossfs.args": "-oupload_buffer_size=33554432"
  }
  

制限事項

• 公式イメージ：OSS Connector for AI/ML は、PyTorch 2.0 以降の公式イメージを使用する DLC ジョブおよび DSW インスタンスでのみ利用可能です。

• カスタムイメージ：PyTorch 2.0 以降のみがサポートされています。サポートされているバージョンを使用するカスタムイメージの場合は、次のコマンドを実行して OSS Connector for AI/ML をインストールしてください：

  pip install -i http://yum.tbsite.net/aliyun-pypi/simple/ --extra-index-url http://yum.tbsite.net/pypi/simple/ --trusted-host=yum.tbsite.net osstorchconnector

• Python バージョン：Python 3.8–3.12 のみがサポートされています。

前提条件

1. 認証情報ファイルを設定してください。

   認証情報は、次のいずれかの方法で設定できます：

   • DLC ジョブの OSS へのパスワードなしアクセスを設定してください。詳細については、「DLC RAM ロールの設定」をご参照ください。この方法では、DLC ジョブは STS から一時的な認証情報を取得して OSS や他のクラウドリソースに安全にアクセスするため、明示的な認証情報が不要になり、アクセスキーの漏洩リスクを軽減します。

   • コードプロジェクトに認証情報ファイルを設定して、認証情報を管理します。次のコードは設定例です：

     説明

     アクセスキー情報を平文で保存すると、セキュリティリスクが生じます。RAM ロールを使用して、DLC インスタンス内で認証情報を自動的に設定することを推奨します。詳細については、「DLC RAM ロールの設定」をご参照ください。

     OSS Connector for AI/ML を使用する際、認証情報ファイルのパスを指定して、OSS データリクエストの署名に必要な認証情報を自動的に取得できます。

     {
       "AccessKeyId": "<Access-key-id>",
       "AccessKeySecret": "<Access-key-secret>",
       "SecurityToken": "<Security-Token>",
       "Expiration": "2024-08-20T00:00:00Z"
     }

     次の表で各フィールドについて説明します。

     パラメータ	必須	説明	例
     AccessKeyId	はい	Alibaba Cloud アカウントまたは RAM ユーザーのアクセスキー ID とアクセスキーシークレット。 説明 STS からの一時的な認証情報を使用して OSS にアクセスする場合、これらのパラメータを一時的なアクセスキー ID とアクセスキーシークレットに設定してください。	NTS****
     AccessKeySecret	はい		7NR2****
     SecurityToken	いいえ	STS からのセキュリティトークン。このパラメータは、STS からの一時的な認証情報を使用して OSS にアクセスする場合にのみ必要です。	STS.6MC2****
     Expiration	いいえ	認証情報の有効期限。このフィールドが空の場合、認証情報は期限切れになりません。認証情報が期限切れになると、OSS Connector for AI/ML は認証情報ファイルを再読み込みします。	2024-08-20T00:00:00Z

2. config.json ファイルを設定してください。次のコードは設定例です：

   このファイルは、同時実行レベル、プリフェッチパラメータ、ログファイルの場所など、プロジェクトの設定を管理するために使用します。OSS Connector for AI/ML を使用する際に、この config.json ファイルのパスを指定すると、コネクタは設定を自動的に適用し、OSS データリクエストに関連するログを指定のログファイルに書き込みます。

   {
       "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
       }
   }

   次の表で各フィールドについて説明します。

   パラメータ		必須	説明	例
   logLevel		はい	ログレベル。デフォルト値は 1 (INFO) です。有効な値： 0: DEBUG 1: INFO 2: WARN 3: ERROR	1
   logPath		はい	コネクタのログパス。デフォルトパスは /var/log/oss-connector/connector.log です。	/var/log/oss-connector/connector.log
   auditPath		はい	コネクタ I/O の監査ログパス。監査ログは、100 ミリ秒を超えるレイテンシーの読み書きリクエストを記録します。デフォルトパスは /var/log/oss-connector/audit.log です。	/var/log/oss-connector/audit.log
   DatasetConfig	prefetchConcurrency	はい	データセットを使用する際に、OSS からデータをプリフェッチするための同時タスク数。デフォルト値は 24 です。	24
   	prefetchWorker	はい	データセットを使用する際に、OSS からデータをプリフェッチするために利用可能な vCPU 数。デフォルト値は 2 です。	2
   checkpointConfig	prefetchConcurrency	はい	チェックポイント読み取り中に OSS からデータをプリフェッチするための同時タスク数。デフォルト値は 24 です。	24
   	prefetchWorker	はい	チェックポイント読み取り中に OSS からデータをプリフェッチするために利用可能な vCPU 数。デフォルト値は 4 です。	4
   	uploadConcurrency	はい	チェックポイント書き込み中にデータをアップロードするための同時タスク数。デフォルト値は 64 です。	64

使用方法

OSS Connector for AI/ML は、PyTorch の Dataset および IterableDataset インターフェースをそれぞれ拡張した、OssMapDataset と OssIterableDataset という 2 つのデータセットアクセスインターフェースを提供します。OssIterableDataset は、より高いトレーニング効率のためにプリフェッチで最適化されています。OssMapDataset のデータ読み取り順序は DataLoader によって決定され、シャッフル操作をサポートします。次の推奨事項に基づいて、データセットアクセスインターフェースを選択できます：

• メモリが限られており、データセットが大規模で、並列度が低いシーケンシャル読み取りのみが必要な場合は、OssIterableDataset を使用します。

• メモリが十分にあり、データセットが比較的小さく、ランダムアクセスと並列処理が必要な場合は、OssMapDataset を使用します。

OSS Connector for AI/ML は、モデルの読み込みと保存のための OssCheckpoint インターフェースも提供します。現在、OssCheckpoint 機能は汎用コンピューティングリソース環境でのみ利用可能です。

以下のセクションでは、これら 3 つのインターフェースの使用方法について説明します。

checkpoint = OssCheckpoint(endpoint="{oss_endpoint}", cred_path=cred_path, config_path=config_path)

checkpoint_read_uri = "{checkpoint_path}"
checkpoint_write_uri = "{checkpoint_path}"
with checkpoint.reader(checkpoint_read_uri) as reader:
    state_dict = torch.load(reader)
    model.load_state_dict(state_dict)
with checkpoint.writer(checkpoint_write_uri) as writer:
    torch.save(model.state_dict(), writer)

コード例

次のコードは、OSS Connector for AI/ML を使用して OSS データにアクセスする方法の完全な例です：

from osstorchconnector import OssMapDataset, OssCheckpoint
import torchvision.transforms as transforms
import accimage
import torchvision.models as models
import torch

# DLC ジョブまたは DSW インスタンスに RAM ロールを設定した後のデフォルトの認証情報パス。
cred_path = "/mnt/.alibabacloud/credentials"  
config_path = "config.json"
checkpoint = OssCheckpoint(endpoint="{oss_endpoint}", cred_path=cred_path, config_path=config_path)
model = models.__dict__["resnet18"]()

epochs = 100  # エポック数を指定します。
checkpoint_read_uri = "{checkpoint_path}"
checkpoint_write_uri = "{checkpoint_path}"
with checkpoint.reader(checkpoint_read_uri) as reader:
    state_dict = torch.load(reader)
    model.load_state_dict(state_dict)


def read_and_transform(data):
    normalize = transforms.Normalize(mean=[0.485, 0.456, 0.406],
                                     std=[0.229, 0.224, 0.225])
    transform = transforms.Compose([
        transforms.RandomResizedCrop(224),
        transforms.RandomHorizontalFlip(),
        transforms.ToTensor(),
        normalize,
    ])

    try:
        img = accimage.Image((data.read()))
        val = transform(img)
        label = data.label # ファイル名
    except Exception as e:
        print("read failed", e)
        return None, 0
    return val, label
dataset = OssMapDataset.from_prefix("{oss_data_folder_uri}", endpoint="{oss_endpoint}", transform=read_and_transform, cred_path=cred_path, config_path=config_path)
data_loader = torch.utils.data.DataLoader(
    dataset, batch_size="{batch_size}", num_workers="{num_workers}", pin_memory=True)

for epoch in range(epochs):
    for step, (images, target) in enumerate(data_loader):
        # バッチ処理
        # モデルトレーニング
    # モデルの保存
    with checkpoint.writer(checkpoint_write_uri) as writer:
        torch.save(model.state_dict(), writer)

この例の要点：

• OssMapDataset を使用して、指定された OSS URI から直接データセットを構築し、標準的な PyTorch の DataLoader パターンに従います。

• このデータセットを使用して標準的な PyTorch の DataLoader を作成し、通常のトレーニングループを実行して各バッチを処理し、モデルをトレーニングし、チェックポイントを保存します。

• これによりオンデマンド読み込みが可能になり、コンテナ環境にデータセットをマウントしたり、事前にローカルストレージにダウンロードしたりする必要がなくなります。

OSS Python SDK

OSS Python SDK を使用して、OSS からデータを読み取り、OSS にデータを書き込みます。

1. OSS Python SDK をインストールします。詳細については、「インストール (Python SDK V1)」をご参照ください。

2. OSS Python SDK のアクセス認証情報を設定します。詳細については、「アクセス認証情報の設定 (Python SDK V1)」をご参照ください。

3. OSS のデータを読み書きします。

   # -*- coding: utf-8 -*-
   import oss2
   from oss2.credentials import EnvironmentVariableCredentialsProvider
   
   # 環境変数から RAM ユーザーのアクセスキーを使用してアクセス認証情報を設定します。
   auth = oss2.ProviderAuth(EnvironmentVariableCredentialsProvider())
   bucket = oss2.Bucket(auth, '<Endpoint>', '<your_bucket_name>')
   # ファイル全体を読み取ります。
   result = bucket.get_object('<your_file_path/your_file>')
   print(result.read())
   # 範囲を指定してデータを読み取ります。
   result = bucket.get_object('<your_file_path/your_file>', byte_range=(0, 99))
   # OSS にデータを書き込みます。
   bucket.put_object('<your_file_path/your_file>', '<your_object_content>')
   # 追加可能ファイルにデータを追加します。
   result = bucket.append_object('<your_file_path/your_file>', 0, '<your_object_content>')
   result = bucket.append_object('<your_file_path/your_file>', result.next_position, '<your_object_content>')
   

   必要に応じて次のパラメーターを変更します。

   パラメーター	説明
   <Endpoint>	バケットのリージョンのエンドポイント。たとえば、China (Hangzhou) リージョンの場合、エンドポイントを https://oss-cn-hangzhou.aliyuncs.com に設定します。エンドポイントの取得方法の詳細については、「リージョンとエンドポイント」をご参照ください。
   <your_bucket_name>	バケットの名前。
   <your_file_path/your_file>	読み取りまたは書き込むファイルのパス。オブジェクトの完全なパスを指定しますが、バケット名は含めません。たとえば、testfolder/exampleobject.txt のように指定します。
   <your_object_content>	書き込みまたは追加するコンテンツ。必要に応じて置き換えてください。

OSS Python API

OSS Python API を使用して、トレーニングデータとモデルを OSS に簡単に保存できます。開始する前に、OSS Python SDK がインストールされており、アクセス認証情報が設定されていることを確認してください。詳細については、「インストール (Python SDK V1)」および「アクセス認証情報の設定 (Python SDK V1)」をご参照ください。

• トレーニングデータの読み込み

  データを OSS バケットに保存し、データパスと対応するラベルを同じバケット内のインデックスファイルに保存できます。カスタムデータセットを作成することで、PyTorch の DataLoader API を使用して、複数のプロセスで並列にデータを読み取ることができます。以下にコード例を示します。

  import io
  import oss2
  from oss2.credentials import EnvironmentVariableCredentialsProvider
  from PIL import Image
  import torch
  
  class OSSDataset(torch.utils.data.dataset.Dataset):
      def __init__(self, endpoint, bucket, auth, index_file):
          self._bucket = oss2.Bucket(auth, endpoint, bucket)
          self._indices = self._bucket.get_object(index_file).read().split(',')
  
      def __len__(self):
          return len(self._indices)
  
      def __getitem__(self, index):
          img_path, label = self._indices[index].strip().split(':')
          img_str = self._bucket.get_object(img_path)
          img_buf = io.BytesIO()
          img_buf.write(img_str.read())
          img_buf.seek(0)
          img = Image.open(img_buf).convert('RGB')
          img_buf.close()
          return img, label
  
  
  # 環境変数からアクセス認証情報を取得します。このコードサンプルを実行する前に、
  # OSS_ACCESS_KEY_ID および OSS_ACCESS_KEY_SECRET 環境変数が設定されていることを確認してください。
  auth = oss2.ProviderAuth(EnvironmentVariableCredentialsProvider())
  dataset = OSSDataset(endpoint, bucket, auth, index_file)
  data_loader = torch.utils.data.DataLoader(
      dataset,
      batch_size=batch_size,
      num_workers=num_loaders,
      pin_memory=True)
  

  次の表で、主要なパラメーターについて説明します。

  パラメーター	説明
  endpoint	バケットのリージョンのエンドポイント。たとえば、China (Hangzhou) リージョンの場合、エンドポイントを https://oss-cn-hangzhou.aliyuncs.com に設定します。エンドポイントの取得方法の詳細については、「リージョンとエンドポイント」をご参照ください。
  bucket	バケットの名前。
  index_file	インデックスファイルのパス。 説明 この例のインデックスファイルでは、カンマ (,) で各サンプルを区切り、コロン (:) でサンプルパスとラベルを区切ります。

• モデルの保存または読み込み

  OSS Python API を使用して、PyTorch モデルを保存または読み込みます。PyTorch でモデルを保存および読み込む方法の詳細については、「PyTorch」をご参照ください。以下にコード例を示します。

  • モデルの保存

    from io import BytesIO
    import torch
    import oss2
    from oss2.credentials import EnvironmentVariableCredentialsProvider
    
    auth = oss2.ProviderAuth(EnvironmentVariableCredentialsProvider())
    # bucket_name
    bucket_name = "<your_bucket_name>"
    bucket = oss2.Bucket(auth, endpoint, bucket_name)
    buffer = BytesIO()
    torch.save(model.state_dict(), buffer)
    bucket.put_object("<your_model_path>", buffer.getvalue())
    

    各項目の説明：

    • endpoint は、バケットのリージョンのエンドポイントです。たとえば、China (Hangzhou) リージョンの場合、エンドポイントを https://oss-cn-hangzhou.aliyuncs.com に設定します。

    • <your_bucket_name> は、oss:// プレフィックスを除いた OSS バケットの名前です。

    • <your_model_path> は、モデルのパスです。必要に応じて置き換えてください。

  • モデルの読み込み

    from io import BytesIO
    import torch
    import oss2
    from oss2.credentials import EnvironmentVariableCredentialsProvider
    
    auth = oss2.ProviderAuth(EnvironmentVariableCredentialsProvider())
    bucket_name = "<your_bucket_name>"
    bucket = oss2.Bucket(auth, endpoint, bucket_name)
    buffer = BytesIO(bucket.get_object("<your_model_path>").read())
    model.load_state_dict(torch.load(buffer))

    各項目の説明：

    • endpoint は、バケットのリージョンのエンドポイントです。たとえば、China (Hangzhou) リージョンの場合、エンドポイントを https://oss-cn-hangzhou.aliyuncs.com に設定します。

    • <your_bucket_name> は、oss:// プレフィックスを除いた OSS バケットの名前です。

    • <your_model_path> は、モデルのパスです。必要に応じて置き換えてください。