クラウドストレージ用の分散アクセラレータ(DACS)は、分散キャッシュを通じてAzure Blobデータを読み取るネイティブなPythonクライアントを提供します。
この記事を使う前に、 AKSのクラウドストレージ用分散アクセラレータを使い始め ましょう。
ネイティブのPythonクライアントをインストールしてください
PyPIのパッケージをワークロードイメージにインストールしてください:
python3 -m pip install "dacs-client==0.1.0"
既存のワークロードを更新する場合は、そのワークロードのイメージビルド定義のパッケージピンを更新し、イメージを再構築してプッシュし、更新されたイメージでワークロードポッドを再デプロイしてください。 パッケージをローカルにインストールしても、実行中のコンテナは更新されません。 イメージビルドやプッシュコマンドについては、「 クライアントをワークロードイメージに追加」をご覧ください。
パッケージにはPython 3.8以降と、glibc 2.35以降のLinux環境が必要です。 x86-64およびarm64用の自己完結型 manylinux_2_35 ホイールを提供し、ネイティブライブラリおよび共有ライブラリ依存関係も含まれます。
azure-identity依存関係は自動的にインストールされます。
クイックスタートでは、python:3.12-slim-bookworm 上で x86-64 ノードを使用して Python 3.12 を使用しています。
ワークロードコンテナ内のインストールを確認してください:
python3 -c "import importlib.metadata; import py_dacs_client; print(importlib.metadata.version('dacs-client'))"
コマンドは 0.1.0を印刷しなければなりません。 流通名はdacs-clientですが、Python輸入名はpy_dacs_clientです。
クライアントの構成
クラウドストレージに分散アクセラレータを使用するすべてのポッドで、両方のキャッシュディスカバリ変数を設定してください:
env:
- name: CACHE_DISCOVERY_URL
value: <discovery-service-hostname>
- name: CACHE_SERVER_PORT
value: "9065"
Important
両方の変数が必要です。 もしどちらかが欠けている場合、クライアントはCache not configured. Working in Remote Blob mode only.をログに記録し、Azure Blob Storage に直接アクセスします。 この操作はキャッシュを使わずに成功可能です。
例えば、キャッシュリソースおよび名前空間用のDiscoveryサービスのホスト名(例えば cache-sample-discovery.<namespace>.svc.cluster.local)を、ポート 9065を付けて使用します。
クライアントを作成する前にBlob認証を設定してください。 以下の例では、ノードのインスタンスメタデータサービス(IMDS)を通じてAKS kubeletマネージデンティティを用いており、Federated AKSワークロードIDではありません。
AZURE_CLIENT_ID を kubelet アイデンティティのクライアント ID に設定します:
env:
- name: AZURE_CLIENT_ID
value: <kubelet-identity-client-id>
- name: MSI_ENDPOINT
value: "http://169.254.169.254/metadata/identity/oauth2/token"
- name: MSI_SECRET
value: "required-nonempty-value"
ポッドはIMDSに到達できなければなりません。 このネイティブクライアント構成において、 MSI_SECRET は必須の空でない互換性マーカーであり、実際の秘密や認証情報ではありません。 その価値に資格をつけないでください。 これらの設定はこの統合に特化したものです。これらをRun:ai認証設定にコピーしないでください。
読み取り専用推論の場合、kubelet ID にモデル コンテナーをスコープとする Storage Blob Data Reader ロールを付与します。 モデル データへの読み取りアクセスの付与を参照してください。 アプリケーションがデータを書き込む必要がある場合にのみ Storage Blob Data Contributor を使用してください。 ランタイムの読み込みには別のアップローダーの書き込み権限は必要ありません。
この正規パス形式をご利用ください:
az://<account>.blob.core.windows.net/<container>/<path>
ネイティブレイヤーは、プロセス内で最初のクライアントが作成されるときに設定を読み込みます。 クライアントを作成する前に環境とクライアント設定を設定してください。 後から接続するクライアントが、すでに有効になっているネイティブ構成を変更することはありません。
キャッシュミスと同時読み込みを理解する
複数のクライアントが同時に同じ欠落データを要求した場合、各クライアントは独立してAzure Blob Storageからデータを取得できます。 このプレビューでは、協調コールドキャッシュの事前投入は利用できません。
キャッシュの生成は非同期です。 読み取りはキャッシュからデータが利用可能になる前に完了することがあるため、返還されたバイトが一致しても即時のリードが両方ともキャッシュミスになることがあります。
大規模同期スケールイベントの前に、選択されたモデルのブロブをDACSでウォームアップしてから推論レプリカを開始します。 デプロイ前にモデルデータをプリフェッチする に従って、専用のウォームアップ Pod、read コマンド、ロールアウトの検証を行ってください。
読み取り成功はAzure Blob Storageのフォールバックから得られます。 読み取りで DACS が使用されていることを確認するには:
- 読む前に、キャッシュ リソースのすべてのキャッシュサーバー Pod で合計した
cache_server_request_counter{request_type="Download",status="Success"}を記録してください。 ラベルの値は大文字と小文字を区別します。 - 同じブロブの範囲を繰り返し読み込み、読み取りの間に遅延があります。 クイックスタートは1秒間隔で5回のリードを使用します。このシーケンスはデータ投入が完了したことを保証するものではありません。 サーバーの正常に完了したダウンロード回数が増えたことと、クライアントがキャッシュヒットを報告していることを確認しましょう。 返されたバイト数とチェックサムを確認して、データの整合性を検証してください。応答時間が速いだけでは、キャッシュされている証拠にはなりません。
- クライアントとサーバーのメトリクスをAzure Blobのテレメトリと関連付けて、キャッシュトラフィックとオリジン読み取り、フォールバックを区別します。
クライアントのメトリックとサーバーのカウンターを収集するためのコマンドについては、DACSリクエストパスの検証を参照してください。
ネイティブ クライアントを使用する
ネイティブクライアントは非同期メソッドと同期対応のメソッドを提供します。 事前に割り当てられたネイティブバッファに読み込みます。
MODEL_PATHを既存のモデルシャードの標準的なaz://パスに設定し、次の例を実行する前に、この例では先頭の4 MiBまでを読み取ります。
import asyncio
import os
from py_dacs_client.dacs_client import DACSCacheClient
from py_dacs_client.dacs_utils import BoundedBufferManager
async def main():
path = os.environ["MODEL_PATH"]
client = DACSCacheClient()
buffers = BoundedBufferManager(max_memory_limit=64 * 1024 * 1024)
try:
size = min(await client.get_file_size(path), 4 * 1024 * 1024)
if size == 0:
print("read 0 bytes")
return
buffer = await buffers.acquire_buffer(size)
try:
bytes_read = await client.read_blob(path, buffer, size, 0)
print(f"read {bytes_read} bytes")
finally:
await buffers.release_buffer(buffer)
finally:
await client.shutdown()
asyncio.run(main())
一般的な読み取り方法には以下があります:
| 非同期メソッド | 同期メソッド | Description |
|---|---|---|
get_file_size(path) |
get_file_size_sync(path) |
ブロブのサイズをバイト単位で取得してください。 |
get_file_stat(path) |
get_file_stat_sync(path) |
モデルファイルのメタデータを取得する。 |
read_blob(path, buffer, size, offset) |
read_blob_sync(...) |
事前に割り当てられたバッファにバイト範囲を読み込みます。 |
list_dir(path) |
list_dir_sync(path) |
ディレクトリエントリを一覧表示します。 |
get_file_handle(path, flags) |
get_file_handle_sync(...) |
再利用可能なハンドラを開く。 |
release_file_handle(handle) |
release_file_handle_sync(handle) |
再利用可能なハンドルを閉じます。 |
shutdown() |
shutdown_sync() |
接続を解放し、最終的な指標を発行します。 |
非同期 list_dir(path) メソッドはイテレーターを返します。ディレクトリエントリを反復処理するには async for を使用します。
すべてのバイトを必要としないモデルフォーマットにはレンジリードを使いましょう。 クラウドストレージ用の分散アクセラレータは、要求された範囲をカバーするチャンクのみを保存します。
読み取り性能の最適化
アプリケーションの読み取り同時実行数と要求バイト範囲サイズを一緒に調整します。 最適な設定は、アクセスパターン、ポッドCPUとメモリの制限、クライアントとキャッシュノードのネットワーク帯域幅、そして読み取りがキャッシュにヒットするかBlob Storageにフォールバックするかによって異なります。
このセクションの値はワークロード ベンチマークの出発点として使用し、性能を保証する推奨値とはみなさないでください。 変更を導入する前に、自分の作業量でその効果を測定してください。
並行性を調整する
並行性とは、アプリケーションが同時進行中の読み取り要求数のことです。 各 read_blob 呼び出しを待って次の呼び出しを開始するループは、非同期APIを使っていてもシーケンシャルです。
- 独立したバイトレンジやブロブ読み取りを同時に発行します。 アクティブなリード数を、
asyncio.Semaphoreまたは固定数の作業タスクで制限します。 大規模なデータセットの各範囲ごとに一度にタスクを作成しないでください。 - まずは4本の同時読み取りなどの小さな制限から始め、それを8本と16本と比較してください。 スループットが向上している間に限り、許容できない遅延、メモリ使用、エラー、スロットリングが発生しない範囲でのみ並行性を増やしてください。
- ネイティブクライアントは
concurrent.futures.Executorコンストラクタパラメータを通じてexecutorを受け入れます。 Pythonスレッドプールを明示的に制御するには、例えばThreadPoolExecutorのように、DACSCacheClient(executor=ThreadPoolExecutor(max_workers=8))を渡します。 エグゼキュータへの参照を保持し、すべての読み取りが完了するまでそれを有効な状態に保ち、await client.shutdown()を呼び出した後でそれをシャットダウンします。 - スレッドプールが大きくても順次呼び出しは並行実行にはなりません。 アプリケーションタスクとexecutor ワーカーの両方をバインドし、各アクティブリードごとに別々のバッファを使用します。
すべてのワーカープロセスとポッド間の並行性を考慮してください。 スケールアウト中に各ポッドの同時進行量を増やすと、同じキャッシュノードが過負荷になったり、コールドキャッシュミス時に重複するBlobダウンロードが増加したりする可能性があります。
キャッシュチャンク サイズを設定
アプリケーションポッドで SI_streamingChunkSizeをバイト単位で設定してキャッシュチャンクサイズを設定します。 例えば、以下の設定では3MiBチャンクを使用します:
env:
- name: SI_streamingChunkSize
value: "3145728"
プロセスの最初のクライアントを作成する前にこの変数を設定してください。 それを変更した後にアプリケーションポッドを再起動してください。
デフォルトのキャッシュチャンクサイズは32 MiBです。
size への read_blob(path, buffer, size, offset) 引数は、キャッシュチャンクサイズではなく、アプリケーションが要求するバイト範囲を指定します。
| アクセス パターン | アプリケーションの読み取りガイダンス |
|---|---|
| 大規模逐次モデルの読み出し | より大きなキャッシュチャンク、要求された範囲サイズ、そして並列数の上限を比較してください。 一度に 1 つの設定を変更します。 |
| 小規模またはランダムな読み取り | アプリケーションが必要とするバイトだけを要求してください。 要求された範囲外で取得されるデータを減らすため、より小さいキャッシュチャンクを比較してください。 |
| 容量制約キャッシュ | アプリケーションの読み取りバッファサイズだけでキャッシュ使用量を推定するのではなく、キャッシュ占有率やエビクションの挙動を測定しましょう。 |
予算アプリケーションはキャッシュチャンクとは別にバッファを読み込みます。 例えば、8つの同時4MiB read_blob リクエストは、設定されたキャッシュチャンクサイズに関わらず32MiBのアプリケーションの読み取りバッファ容量を必要とします。 ネイティブクライアント割り当て、キュー作業、Pythonオブジェクト、モデル処理のために追加のポッドメモリを予約してください。 並行制限に入った後にのみバッファを取得し、各バッファを finally ブロックで解放します。
Warning
BoundedBufferManager(max_memory_limit=...) アプリケーションの読み取りバッファ容量を制限しますが、クライアント全体のメモリやネイティブアップロードバッファリングを制限しません。
0.1.0では、ネイティブクライアントのデフォルトの非同期アップロードバッファ予算は400 GiBです。 この予算は事前割り当てでも安全なポッドメモリ制限でもありません。 大規模なコールドキャッシュ読み取りは、小規模なPodのリソースを使い果たす可能性があります。 メモリ制約のあるワークロードでは、同期投入(SI_cacheEnableSyncUpload=true)または制限付き非同期アップロード予算(SI_cacheMaxUploadBufferSizePerProcessGB)をテストします。 これらの値は最初のクライアントを作成する前に設定し、ネイティブバッファは、アプリケーションの読み取りバッファやモデル処理とは別枠で見積もります。
限界を上げる前に測定してください
- 固定データセット、ポッド数、リソース制限でベースラインを確立しましょう。 スループット、読み取り遅延、CPU、メモリ、ネットワーク使用量を測定します。
- コールドキャッシュとウォームキャッシュの読み取りを別々にベンチマークします。
cache_server_request_counterを検査し、Azure Blobテレメトリを調査してキャッシュトラフィックとBlob への直接フォールバックを区別してください。 - 一度に一つの設定を変更します:並行性、アプリケーション読み取りサイズ、キャッシュチャンクサイズ。 結果を比較する際は、データセットやその他の設定を一貫性に保ちましょう。
- スループットが停滞したりレイテンシが増えたり、リソースプレッシャーやエラー、ブロブのスロットリングが増えたら、設定の上げをやめてください。
キャッシュ側の観測については、 クラウドストレージの分散アクセラレータ監視を参照してください。
エラーを処理する
すべてのDACS例外は以下から派生します:
from py_dacs_client import DACSError, DACSNotFoundError
try:
load_model_data()
except DACSNotFoundError:
handle_missing_model()
except DACSError as error:
handle_cache_or_storage_error(error)
| Exception | Meaning |
|---|---|
DACSNotFoundError |
Blobパスやディレクトリは存在しません。 |
DACSExistsError |
アプリケーションは上書きが有効でない状態で既存のブロブに書き込みを試みます。 |
DACSAuthenticationError |
資格情報が見つからないか無効です。 |
DACSPermissionError |
アイデンティティには必要な Blob ロールがありません。 |
DACSTimeoutError |
キャッシュやBlob操作がタイムアウトしました。 |
DACSNetworkError |
キャッシュサーバーまたは Blob エンドポイントにアクセスできません。 |
DACSInvalidInputError |
パスまたは引数の形式が正しくありません。 |
OperationNotSupportedError |
リクエストされたAPIはまだ実装されていません。 |
制御ログ
クライアントはPython層とネイティブ層ごとに別々のログ制御を持っています。 両方の変数を両層からの冗長な出力を抑制するように設定します:
env:
- name: DACS_DATA_LOG_LEVEL
value: WARNING
- name: STORAGE_LOG_LEVEL
value: WARNING
DACS_DATA_LOG_LEVELPythonログを制御します。
STORAGE_LOG_LEVEL はネイティブ クライアントのログ記録を制御します。