Object Storage Service:OSS で 100 Gbps の内部帯域幅を実現

ネットワーク受信能力

ECS インスタンスに OSS クライアントをインストールした場合、データダウンロード速度は ECS インスタンスのネットワーク速度に制限されます。ネットワーク能力が最も高い Alibaba Cloud モデルでは、160 Gbit/s の帯域幅を提供できます。クラスター内で複数のインスタンスをデプロイした場合、クライアントへの同時アクセスによって合計帯域幅が 100 Gbit/s に達します。単一インスタンスのみをデプロイする場合は、ネットワーク拡張型インスタンスファミリーまたは高クロックスピードのインスタンスファミリーを使用することを推奨します。後者は多数のデータパケットを受信する際に優れたパフォーマンスを発揮します。

ディスク I/O

データをローカルマシンにダウンロードした後、ディスク上に保存する場合は、ダウンロード速度がディスク性能に制限されます。ossutil や ossfs などのツールはデフォルトでディスク上にデータを保存します。このようなケースでは、下図に示すように、高性能ディスクまたはメモリディスクを使用してダウンロード速度を向上させることができます。

VPC の使用

Alibaba Cloud の内部ネットワークは、ネットワークリクエスト向けに最適化されています。VPC エンドポイントを使用すると、インターネットよりも安定したネットワーク接続が得られます。最大ダウンロード帯域幅を達成するには、VPC を使用する必要があります。

ご利用の ECS インスタンスは VPC 内で動作します。OSS は統一された内部ドメイン名を提供しており、どのお客様も VPC 経由でアクセスできます。例：oss-cn-beijing-internal.aliyuncs.com。ECS と OSS 間のデータフローは Server Load Balancer (SLB) を経由し、リクエストはバックエンドの分散クラスターに送信されます。これにより、クラスター全体にデータリクエストが均等に分散され、OSS は強力な高同時処理能力を実現します。

同時ダウンロード

OSS は HTTP プロトコルを使用してデータを転送します。単一の HTTP リクエストにはパフォーマンス上の制限があるため、データダウンロードを高速化するために同時ダウンロードが使用されます。たとえば、オブジェクトを複数の範囲に分割し、各リクエストが 1 つの範囲のみにアクセスすることで、最大ダウンロード帯域幅を達成できます。課金は API オペレーションの回数に基づいて行われます。範囲数を増やすと API オペレーション数も増加し、必ずしもシングルストリームデータダウンロード時のピークダウンロード速度が保証されるわけではありません。一般的なツールを用いた同時ダウンロードの詳細については、「一般的なツールを用いた同時ダウンロードの最適化」をご参照ください。

テスト環境

テスト手順

1. 環境変数を設定します。

   export OSS_ACCESS_KEY_ID=<ALIBABA_CLOUD_ACCESS_KEY_ID>
   export OSS_ACCESS_KEY_SECRET=<ALIBABA_CLOUD_ACCESS_KEY_SECRET>

2. サンプルコード：

   package main
   
   import (
       "context"
       "flag"
       "fmt"
       "io"
       "log"
       "time"
       "github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
       "github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
   )
   
   // コマンドライン引数を格納するグローバル変数を定義します。
   var (
       region      string // バケットが配置されているリージョン。
       endpoint    string // OSS にアクセスするためのエンドポイント。
       bucketName  string // バケット名。
       objectName  string // オブジェクト名。
       chunkSize   int64  // チャンクサイズ（バイト単位）。
       prefetchNum int    // プリフェッチするチャンク数。
   )
   
   // コマンドライン引数を解析する関数を初期化します。
   func init() {
       flag.StringVar(&region, "region", "", "バケットが配置されているリージョン。")
       flag.StringVar(&endpoint, "endpoint", "", "OSS にアクセスするためのエンドポイント。")
       flag.StringVar(&bucketName, "bucket", "", "バケット名。")
       flag.StringVar(&objectName, "object", "", "オブジェクト名。")
       flag.Int64Var(&chunkSize, "chunk-size", 0, "チャンクサイズ（バイト単位）。")
       flag.IntVar(&prefetchNum, "prefetch-num", 0, "プリフェッチするチャンク数。")
   }
   
   func main() {
       // コマンドライン引数を解析します。
       flag.Parse()
   
       // 必須パラメーターが指定されているか確認します。
       if len(bucketName) == 0 {
           flag.PrintDefaults()
           log.Fatalf("無効なパラメーターです。バケット名が必要です。")
       }
   
       if len(region) == 0 {
           flag.PrintDefaults()
           log.Fatalf("無効なパラメーターです。リージョンが必要です。")
       }
   
       // OSS クライアントを構成します。
       cfg := oss.LoadDefaultConfig().
           WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()). // 環境変数から認証情報を使用します。
           WithRegion(region) // リージョンを設定します。
   
       // カスタムエンドポイントが指定されている場合、それを設定します。
       if len(endpoint) > 0 {
           cfg.WithEndpoint(endpoint)
       }
   
       // OSS クライアントを作成します。
       client := oss.NewClient(cfg)
   
       // OSS オブジェクトを開きます。
       f, err := client.OpenFile(context.TODO(), bucketName, objectName, func(oo *oss.OpenOptions) {
           oo.EnablePrefetch = true      // プリフェッチを有効にします。
           oo.ChunkSize = chunkSize      // チャンクサイズを設定します。
           oo.PrefetchNum = prefetchNum  // プリフェッチするチャンク数を設定します。
           oo.PrefetchThreshold = int64(0) // プリフェッチしきい値を設定します。
       })
   
       if err != nil {
           log.Fatalf("オープンに失敗しました。err:%v", err)
       }
   
       // 開始時刻を記録します。
       startTick := time.Now().UnixNano() / 1000 / 1000
   
       // ファイル内容を読み取り、速度テストのために破棄します。
       written, err := io.Copy(io.Discard, f)
   
       // 終了時刻を記録します。
       endTick := time.Now().UnixNano() / 1000 / 1000
   
       if err != nil {
           log.Fatalf("コピーに失敗しました。err:%v", err)
       }
   
       // MiB/s 単位で読み取り速度を計算します。
       speed := float64(written/1024/1024) / (float64(endTick-startTick) / 1000)
   
       // 平均読み取り速度を出力します。
       fmt.Printf("平均速度:%.2f(MiB/s)\n", speed)
   }

3. テストプログラムを開始します。

   go run down_object.go -bucket yourbucket -endpoint oss-cn-hangzhou-internal.aliyuncs.com  -object 100GB.file -region cn-hangzhou -chunk-size 419430400 -prefetch-num 256

テスト結果

上記の手順において、同時実行数およびチャンクサイズを調整して、ダウンロード所要時間およびピークダウンロード帯域幅の変化を観測しました。一般的に、同時実行数を CPU コア数の 1～4 倍に設定し、チャンクサイズを FileSize/Concurrency に設定することを推奨します。ただし、チャンクサイズは 2 MB 未満であってはなりません。このパラメーター構成により、最短のダウンロード所要時間を達成でき、ピークダウンロード帯域幅が 100 Gbit/s に到達します。

ossutil

• パラメーターの説明

  パラメーター	説明
  --bigfile-threshold	再開可能なダウンロードを使用するオブジェクトサイズのしきい値。デフォルト値：104857600（100 MB）。有効値：0～9223372036854775807。単位：バイト。
  --range	ダウンロードするオブジェクトのバイト範囲。バイト番号は 0 から始まります。 範囲を指定できます。例：3-9 はバイト 3 からバイト 9 までの範囲（バイト 3 およびバイト 9 を含む）を示します。 ダウンロード開始位置からの範囲を指定できます。例：3- はバイト 3 からオブジェクト末尾までの範囲（バイト 3 を含む）を示します。 ダウンロード終了位置までの範囲を指定できます。例：-9 はバイト 0 からバイト 9 までの範囲（バイト 9 を含む）を示します。
  --parallel	単一オブジェクトに対して実行する同時操作数。有効値：1～10000。デフォルトでは、ossutil が操作タイプおよびオブジェクトサイズに基づいて自動的にこのオプションの値を設定します。
  --part-size	パートサイズ。単位：バイト。有効値：2097152～16777216（2～16 MB）。多くの場合、CPU コア数が多い場合は小さいパートサイズを設定し、CPU コア数が少ない場合はパートサイズを適宜大きくすることができます。

• 例

  次のコマンドは、同時実行数 256、パートサイズ 468,435,456 バイトで 100GB.file オブジェクトを /dev/shm にダウンロードします。

  time ossutil --parallel=256 --part-size=468435456 --endpoint=oss-cn-hangzhou-internal.aliyuncs.com cp oss://cache-test/EcsTest/100GB.file /dev/shm/100GB.file

• パフォーマンスの説明

  平均 E2E 速度は 2.94 GB/s（約 24 Gbit/s）です。

  

ossfs

• パラメーターの説明

  パラメーター	説明
  parallel_count	フラグメントアップロードを使用してラージオブジェクトをアップロードする際の同時ダウンロード可能なパート数。デフォルト値：5。
  multipart_size	フラグメントアップロードを使用してデータをアップロードする際のパートサイズ（MB 単位）。デフォルト値：10。このパラメーターはアップロード可能なオブジェクトの最大サイズを制限します。フラグメントアップロードでは、オブジェクトを最大 10,000 パートに分割できます。デフォルトでは、アップロード可能なオブジェクトの最大サイズは 100 GB です。このオプションの値を変更することで、より大きなオブジェクトをアップロードできます。
  direct_read	ダイレクトリードモードを有効にします。デフォルトでは、ossfs は一時データのアップロードまたはダウンロードにディスクストレージ容量を使用します。このオプションを指定すると、ローカルディスクではなく OSS から直接データを読み取ることができます。このオプションはデフォルトで無効になっています。-odirect_read を使用してダイレクトリードモードを有効にできます。 説明 読み取り中のオブジェクトに対して書き込み、名前変更、または切り詰め操作が実行されると、ダイレクトリードが中断されます。この場合、オブジェクトはダイレクトリードモードを終了し、再度オープンする必要があります。
  direct_read_prefetch_chunks	メモリにプリフェッチできるチャンク数。このオプションは、シーケンシャルリードパフォーマンスの最適化に使用できます。デフォルト値：32。 このオプションは、-odirect_read オプションが指定されている場合にのみ有効です。
  direct_read_chunk_size	1 回のリードリクエストで OSS から直接読み取れるデータ量。単位：MB。デフォルト値：4。有効値：1～32。 このオプションは、-odirect_read オプションが指定されている場合にのみ有効です。
  ensure_diskfree	他のアプリケーションがデータを書き込む際にディスク容量が完全に占有されないようにするための予約ディスク容量サイズ。デフォルトでは、ディスク容量は予約されません。単位：MB。 たとえば、ossfs が 1,024 MB の空きディスク領域を確保するようにするには、マウント時に -oensure_diskfree=1024 を追加します。
  free_space_ratio	予約したい最小残りディスク領域比率。たとえば、ディスク領域が 50 GB で -o free_space_ratio を 20 に設定した場合、10 GB（50 GB × 20% = 10 GB）が予約されます。
  max_stat_cache_size	メタデータキャッシュに格納できるファイルの最大数。デフォルトでは、最大 100,000 オブジェクトのメタデータをキャッシュできます。ディレクトリに多数のオブジェクトが含まれる場合、このオプションを変更することで ls コマンドのオブジェクト一覧表示パフォーマンスを向上させることができます。メタデータキャッシュを無効にするには、このオプションを 0 に設定します。
  stat_cache_expire	オブジェクトメタデータキャッシュの有効期間。単位：秒。デフォルト値：900。
  readdir_optimize	キャッシュ最適化を使用するかどうかを指定します。デフォルト値：false。 このマウントオプションを追加すると、ossfs は ls 操作中に gid や uid などのファイルアイテムメタデータを取得するために HeadObject リクエストを送信しません。HeadObject リクエストは、サイズが 0 のファイルにアクセスされた場合にのみ送信されます。ただし、権限チェックなどの理由により、一定数の HeadObject リクエストが引き続き生成される可能性があります。アプリケーションの特性に応じてこのオプションを有効にしてください。有効にするには、マウント時に -oreaddir_optimize を追加します。

• 例

  説明

  CPU 処理能力およびネットワーク帯域幅に基づいてパラメーターを調整することを推奨します。

  • デフォルトリードモード：ご利用のコンピューター上の /mnt/cache-test フォルダに cache-test という名前のバケットをマウントし、同時ダウンロード可能なパート数を 128 に設定し、各パートのサイズを 32 MB に設定します。

    ossfs cache-test /mnt/cache-test -ourl=http://oss-cn-hangzhou-internal.aliyuncs.com  -oparallel_count=128 -omultipart_size=32 

  • ダイレクトリードモード：ご利用のコンピューター上の /mnt/cache-test フォルダに cache-test という名前のバケットをマウントし、ダイレクトリードモードを有効にし、プリフェッチするチャンク数を 128 に設定し、各チャンクのサイズを 32 MB に設定します。

    ossfs cache-test /mnt/cache-test -ourl=http://oss-cn-hangzhou-internal.aliyuncs.com  -odirect_read -odirect_read_prefetch_chunks=128 -odirect_read_chunk_size=32

• パフォーマンスの説明

  デフォルトリードモードでは、ossfs はデータをローカルディスクにダウンロードします。ダイレクトリードモードでは、データはメモリに保存されるためアクセスが高速化されますが、より多くのメモリを消費します。

  説明

  ダイレクトリードモードでは、ossfs はダウンロードされたデータをチャンク単位で管理します。各チャンクのサイズはデフォルトで 4 MB ですが、direct_read_chunk_size パラメーターを使用して変更できます。メモリ内では、ossfs は [現在のチャンク - 1、現在のチャンク + direct_read_prefetch_chunks] の範囲内のデータを保持します。特にページキャッシュのサイズに基づいて、ダイレクトリードモードを使用するかどうかを判断してください。多くの場合、ダイレクトリードモードはページキャッシュ容量が不足しているシナリオに適しています。たとえば、ご利用のコンピューターの合計メモリが 16 GB で、ページキャッシュが 6 GB を消費できる場合、オブジェクトサイズが 6 GB を超えるときにダイレクトリードモードを使用できます。詳細については、「ダイレクトリードモード」をご参照ください。

  モード	並行性	チャンクサイズ (MB)	ピーク帯域幅 (Gbit/s)	E2E 帯域幅 (Gbit/s)	E2E 所要時間 (秒)
  デフォルトリードモード	128	32	24	11.3	72.01
  ダイレクトリードモード	128	32	24	16.1	50.9

  モデルオブジェクト読み取りの最適化：

  モデルオブジェクトサイズ (GB)	デフォルトリードモード (所要時間：秒；最大メモリ：6 GB)	ハイブリッドリードモード (所要時間：秒)	ハイブリッドリードモード (所要時間：秒；データ保持：[-32, +32])
  1	8.19	8.20	8.56
  2.4	24.5	20.43	20.02
  5	26.5	22.3	19.89
  5.5	22.8	23.1	22.98
  8.5	106.0	36.6	36.00
  12.6	154.6	42.1	41.9

Python SDK

デフォルトでは、Python SDK はストレージにシリアルアクセスします。Python の同時実行ライブラリを使用してマルチスレッドダウンロードを行うことで、帯域幅を大幅に向上させることができます。

• 前提条件

  モデルオブジェクトのサイズは約 5.6 GB で、テストマシンの仕様は ECS 48 vCPU、帯域幅 16 Gbit/s、メモリ 180 GB です。

• 例

  import oss2
  import time
  import os
  import threading
  from io import BytesIO
  
  # パラメーターを設定します。
  OSS_CONFIG = {
      "bucket_endpoint": os.environ.get('OSS_BUCKET_ENDPOINT', 'oss-cn-hangzhou-internal.aliyuncs.com'), # バケットが配置されているリージョンのデフォルトエンドポイント。
      "bucket_name": os.environ.get('OSS_BUCKET_NAME', 'bucket_name'), # バケット名。
      "access_key_id": os.environ['ACCESS_KEY_ID'], # RAM ユーザーの AccessKey ID。
      "access_key_secret": os.environ['ACCESS_KEY_SECRET'] # RAM ユーザーの AccessKey Secret。
  }
  
  # バケットを初期化します。
  def __bucket__():
      auth = oss2.Auth(OSS_CONFIG["access_key_id"], OSS_CONFIG["access_key_secret"])
      return oss2.Bucket(
          auth, 
          OSS_CONFIG["bucket_endpoint"], 
          OSS_CONFIG["bucket_name"], 
          enable_crc=False
      )
  
  # モデルオブジェクトのサイズを照会します。
  def __get_object_size(object_name):
      simplifiedmeta = __bucket__().get_object_meta(object_name)
      return int(simplifiedmeta.headers['Content-Length'])
  
  # モデルオブジェクトの最終更新時刻を照会します。
  def get_remote_model_mmtime(model_name):
      return __bucket__().head_object(model_name).last_modified
  
  # モデルオブジェクトを一覧表示します。
  def list_remote_models(ext_filter=('.ckpt',)): # デフォルトの拡張フィルターを指定します。
      dir_prefix = ""
      output = []
      
      for obj in oss2.ObjectIteratorV2(
          __bucket__(),
          prefix=dir_prefix,
          delimiter='/',
          start_after=dir_prefix,
          fetch_owner=False
      ):
          if not obj.is_prefix():
              _, ext = os.path.splitext(obj.key)
              if ext.lower() in ext_filter:
                  output.append(obj.key)
      return output
  
  # チャンク単位でマルチパートダウンロードスレッド関数を実行します。
  def __range_get(object_name, buffer, offset, start, end, read_chunk_size, progress_callback, total_bytes):
      chunk_size = int(read_chunk_size)
      with __bucket__().get_object(object_name, byte_range=(start, end)) as object_stream:
          s = start
          while True:
              chunk = object_stream.read(chunk_size)
              if not chunk:
                  break
              buffer.seek(s - offset)
              buffer.write(chunk)
              s += len(chunk)
              # ダウンロード済みバイト数を計算し、ダウンロード進捗コールバック関数を呼び出します。
              if progress_callback:
                  progress_callback(s - start, total_bytes)
  
  # モデルオブジェクトを読み取り、オプションで進捗コールバックパラメーターを指定します。
  def read_remote_model(
      checkpoint_file, 
      start=0, 
      size=-1, 
      read_chunk_size=2*1024*1024,  # 2MB
      part_size=256*1024*1024,      # 256MB
      progress_callback=None # ダウンロード進捗コールバック。
  ):
      time_start = time.time()
      buffer = BytesIO()
      obj_size = __get_object_size(checkpoint_file)
      
      end = (obj_size if size == -1 else start + size) - 1
      s = start
      tasks = []
  
      # 進捗を計算します。
      total_bytes = end - start + 1
      downloaded_bytes = 0
  
      while s <= end:
          current_end = min(s + part_size - 1, end)
          task = threading.Thread(
              target=__range_get,
              args=(checkpoint_file, buffer, start, s, current_end, read_chunk_size, progress_callback, total_bytes)
          )
          tasks.append(task)
          task.start()
          s += part_size
  
      for task in tasks:
          task.join()
  
      time_end = time.time()
      # オブジェクトの合計ダウンロード量を表示します。
      print(f"{checkpoint_file} を {time_end - time_start:.2f} 秒でダウンロードしました。")
  
       # ダウンロードされたオブジェクトのサイズを計算して表示します。単位：GB。
      file_size_gb = obj_size / (1024 * 1024 * 1024)
      print(f"合計ダウンロードファイルサイズ：{file_size_gb:.2f} GB")
  
      buffer.seek(0)
      return buffer
  
  # ダウンロード進捗コールバック関数を指定します。
  def show_progress(downloaded, total):
      progress = (downloaded / total) * 100
      print(f"進捗：{progress:.2f}%", end="\r")
  
  # ダウンロード進捗コールバック関数を呼び出します。
  if __name__ == "__main__":
      # list_remote_models メソッドを使用してモデルオブジェクトを一覧表示します。
      models = list_remote_models()
      print("リモートモデル：", models)
  
      if models:
          # 最初のモデルオブジェクトをダウンロードします。
          first_model = models[0]
          buffer = read_remote_model(first_model, progress_callback=show_progress)
          print(f"\n{first_model} をバッファーにダウンロードしました。")

• 結論

  バージョン	OSS 所要時間 (秒)	OSS 平均帯域幅 (Mbit/s)	OSS ピーク帯域幅 (Mbit/s)
  OSS SDK for Python (シリアルモード)	109	53	100
  OSS SDK for Python (同時ダウンロードモード)	11.1	516	600

  テスト結果から明らかにわかるように、OSS Python SDK のシリアルモードと比較して、同時ダウンロードモードでは所要時間が約 10.2 % に短縮され、平均帯域幅は約 9.7 倍、ピーク帯域幅は 6 倍に向上しています。したがって、AI モデルトレーニングで Python SDK を使用して同時ダウンロードを行うことで、効率および帯域幅が大幅に向上します。

• その他のソリューション

  Python 向け OSS SDK に加えて、Alibaba Cloud は osstorchconnector という名前の Python ライブラリを提供しています。このライブラリは、主に PyTorch トレーニングタスクで OSS データへの効率的なアクセスおよび保存に使用されます。このライブラリは、ユーザー向けの同時ダウンロードの二次カプセル化を完了しています。以下の表では、osstorchconnector を使用した AI モデルのロードに関するテスト結果について説明します。詳細については、「性能テスト」をご参照ください。

  項目	説明
  テストシナリオ	モデルロードおよびチャット Q&A
  モデル名	gpt3-finnish-3B
  モデルサイズ	11 GB
  シナリオ	チャット Q&A
  ハードウェア構成	高性能 ECS インスタンス：96 vCPU、メモリ 384 GiB、内部帯域幅 30 Gbit/s
  結論	平均帯域幅は約 10 Gbit/s です。OSS は 10 タスクが同時にモデルをロードすることをサポートできます。