Object Storage Service:OSS Connector for AI/ML と OSS Accelerator を使用した小さいファイルの読み取りの高速化

加速の仕組み

OSS Connector for AI/ML と OSS Accelerator は、それぞれアプリケーション層とストレージ層という異なるレイヤーで最適化を行い、組み合わせることで相乗効果を生み出します。

• OSS Connector for AI/ML (アプリケーション層): 非同期 I/O とマルチスレッドプリフェッチを使用して、シリアルファイルリクエストを高度に同時並行な並列ストリームに変換します。次のバッチのデータはバックグラウンドでプリフェッチされ、I/O 待ちによる CPU/GPU のアイドル時間を排除します。

• OSS Accelerator (ストレージ層): コールド/ウォームキャッシュモデルを使用して、パフォーマンス専有型ストレージメディアにホットデータをキャッシュします。最初の読み取り時 (コールドキャッシュ) には、データは OSS オリジンからキャッシュにロードされるため、パフォーマンスは標準の OSS アクセスと同等です。その後の読み取り時 (ウォームキャッシュ) には、データはキャッシュから直接提供されます。これは OSS オリジンと比較して P50 レイテンシーが約 2.8 倍低くなります (約 12 ミリ秒 vs. 35 ミリ秒)。同じデータセットが複数のエポックにわたって読み取られるモデルトレーニングでは、2 番目のエポックから加速が有効になります。

• 組み合わせた効果: Connector からの大量の同時リクエストは、ミリ秒単位の速度で Accelerator によって吸収され、OSS オリジンレイテンシーがボトルネックとなることを排除します。トレーニング開始時やエポックトランジション中のトラフィックバーストは、キャッシュによって平滑化され、一貫して高いスループットを確保します。

前提条件

• 訓練データセットがアップロードされた OSS バケット。

• AccessKey ID および AccessKey シークレット。詳細については、「AccessKey ペアの作成」をご参照ください。

• ECS インスタンス。コンピューティング最適化またはネットワーク最適化インスタンス (ecs.g7.32xlarge など) を推奨します。Alibaba Cloud Linux 3/4 が推奨されるオペレーティングシステムです。詳細については、「ECS インスタンスのカスタム起動」をご参照ください。

バケットの OSS Accelerator を有効にする

1. OSS コンソールにログインし、ターゲットバケット名をクリックします。

2. 左側のナビゲーションウィンドウで、[バケット設定] > [OSS Accelerator] を選択します。

3. [OSS Accelerator の作成] をクリックし、次のパラメーターを構成します。

   パラメーター	説明
   ゾーン	ECS インスタンスと同じゾーンを選択します。この例では、中国 (北京) ゾーン H が使用されます。
   容量	データセットの合計サイズ以上である必要があります。この例では、20 TB が使用されます。
   アクセラレーションポリシー	[指定パスの高速化] を選択してデータセットプレフィックスを入力するか、[バケット全体の高速化] を選択します。

   重要

   ECS インスタンスと OSS Accelerator は同じゾーンにある必要があります。クロスゾーンアクセスは追加のネットワーク遅延を発生させ、アクセラレーションパフォーマンスを低下させます。

4. 作成後、OSS Connector for AI/ML で使用するエンドポイントは、oss-cache-<zone>.aliyuncs.com の形式に従います。たとえば、ゾーンが cn-beijing-h の場合、対応するエンドポイントは oss-cache-cn-beijing-h.aliyuncs.com です。

OSS Connector for AI/ML のインストールと構成

1. ECS インスタンスにログインし、OSS Connector for AI/ML (PyTorch エディション) をインストールします。

   pip install osstorchconnector

2. アクセス認証情報を構成します。<yourAccessKeyId> と <yourAccessKeySecret> を実際の AccessKey 情報に置き換えます。

   mkdir -p /root/.alibabacloud
   cat > /root/.alibabacloud/credentials << 'EOF'
   {
       "AccessKeyId": "<yourAccessKeyId>",
       "AccessKeySecret": "<yourAccessKeySecret>"
   }
   EOF

   認証情報ファイルは JSON 形式を使用する必要があります。その他の構成オプションについては、「OSS コネクタの AI/ML 向け構成」をご参照ください。

3. OSS Connector の構成ファイルを作成し、小さいファイルの読み取りのプリフェッチパラメーターを調整します。

   mkdir -p /etc/oss-connector/
   cat > /etc/oss-connector/config.json << 'EOF'
   {
       "logLevel": 1,
       "logPath": "/var/log/oss-connector/connector.log",
       "auditPath": "/var/log/oss-connector/audit.log",
       "datasetConfig": {
           "prefetchMB": 1024,
           "prefetchConcurrency": 16,
           "prefetchWorker": 2,
           "prefetchUnitMB": 1,
           "timeoutMs": 10000
       },
       "checkpointConfig": {
           "prefetchConcurrency": 24,
           "prefetchWorker": 4,
           "uploadConcurrency": 64
       }
   }
   EOF

   主要パラメーター：

   パラメーター	推奨値	説明
   prefetchMB	1024	プリフェッチバッファーサイズ (MB 単位)。1 GB のバッファーは、それぞれ 115 KB のファイルを約 8,700 個保持できます。大きいファイルの場合はこの値を増やします。
   prefetchConcurrency	16	同時プリフェッチ操作の数。高帯域幅インスタンスを最大限に活用します。
   prefetchUnitMB	1	各プリフェッチユニットのサイズ (MB 単位)。小さいファイルの場合は 1 MB に設定します。大きいファイルの場合はファイルサイズに合わせて増やします。

パフォーマンス比較テストの実行

OSS 内部エンドポイントと OSS Accelerator エンドポイントの両方を使用してデータセットを読み取り、結果を比較する test_accelerator.py という名前のテストスクリプトを作成します。

from osstorchconnector import OssMapDataset
import torch
from torch.utils.data import DataLoader
import time
import numpy as np

# === Configuration ===
# First run: use OSS internal endpoint. Second run: switch to OSS Accelerator endpoint.
ENDPOINT = "http://oss-cn-beijing-internal.aliyuncs.com"
# ENDPOINT = "http://oss-cache-cn-beijing-h.aliyuncs.com"

CONFIG_PATH = "/etc/oss-connector/config.json"
CRED_PATH = "/root/.alibabacloud/credentials"
OSS_URI = "oss://<yourBucketName>/<yourDatasetPrefix>/"
REGION = "cn-beijing"

def collate_fn(batch):
    results = []
    for item in batch:
        start = time.perf_counter()
        content = item.read()
        read_time = time.perf_counter() - start
        results.append({'size': item.size, 'read_time': read_time})
    return results

dataset = OssMapDataset.from_prefix(
    OSS_URI,
    endpoint=ENDPOINT,
    cred_path=CRED_PATH,
    config_path=CONFIG_PATH,
    region=REGION
)

NUM_WORKERS = 8
BATCH_SIZE = 256
dataloader = DataLoader(
    dataset,
    batch_size=BATCH_SIZE,
    num_workers=NUM_WORKERS,
    collate_fn=collate_fn,
    pin_memory=False,
    shuffle=True
)

# === Run test ===
all_batch_durations = []
total_size = 0
file_count = 0
start_time = time.perf_counter()
last_receive_time = start_time

print(f"Starting test: {NUM_WORKERS} workers, batch_size={BATCH_SIZE}")
print(f"Endpoint: {ENDPOINT}\n")

try:
    for batch in dataloader:
        current_receive_time = time.perf_counter()
        batch_duration = current_receive_time - last_receive_time
        all_batch_durations.append(batch_duration)
        last_receive_time = current_receive_time

        batch_total_size = sum(data['size'] for data in batch)
        total_size += batch_total_size
        file_count += len(batch)

        if file_count % 10000 == 0:
            elapsed = current_receive_time - start_time
            throughput = total_size / elapsed / (1024 ** 2)
            print(f"  {file_count:,} files processed | Throughput: {throughput:.2f} MB/s")

except KeyboardInterrupt:
    print("\nTest interrupted")

# === Results ===
end_time = time.perf_counter()
total_elapsed = end_time - start_time
num_batches = len(all_batch_durations)

if num_batches > 0:
    durations_ms = np.array(all_batch_durations) * 1000
    print(f"\n{'='*50}")
    print(f"Test Results")
    print(f"{'='*50}")
    print(f"Total files:      {file_count:,}")
    print(f"Total data:       {total_size / (1024**2):,.2f} MB")
    print(f"Total time:       {total_elapsed:.2f} s")
    print(f"Avg throughput:   {total_size / total_elapsed / (1024**2):,.2f} MB/s")
    print(f"\nBatch latency (ms): Avg={np.mean(durations_ms):.2f}  P50={np.percentile(durations_ms, 50):.2f}  P95={np.percentile(durations_ms, 95):.2f}")
else:
    print("No batches received")

スクリプト内の OSS_URI と REGION を実際の値に置き換えてから、以下の手順に従います。

1. OSS 内部エンドポイントで実行 (ベースライン)

スクリプト内の ENDPOINT が OSS 内部アドレスに設定されていることを確認し、テストを実行して合計時間を記録します。

python test_accelerator.py

2. OSS Accelerator エンドポイントに切り替える

ENDPOINT をアクセラレータアドレスに変更します。

ENDPOINT = "http://oss-cache-cn-beijing-h.aliyuncs.com"

アクセラレータを介した最初の読み取りでは、OSS オリジンからキャッシュにデータがロードされるため (コールドキャッシュ)、パフォーマンスは直接 OSS アクセスと同等になります。テストを 2 回実行する必要があります。最初の実行でキャッシュをウォームアップし、2 回目の実行で実際の加速が示されます。

# First run: warm up the cache
python test_accelerator.py

# Second run: cache hits, acceleration takes effect
python test_accelerator.py

ベースライン実行の合計時間と 2 回目のアクセラレータ実行を比較して、小さいファイルワークロードの高速化を確認します。

パフォーマンス比較結果

以下の結果は、ImageNet データセット (128 万ファイル、合計 137 GB、平均ファイルサイズ約 115 KB) を使用した ecs.g7.32xlarge インスタンスで測定されました。

実際のパフォーマンスは、インスタンスタイプ、データセットサイズ、ファイルサイズによって異なります。正確なベンチマークを得るには、独自のテストを実行することを推奨します。