トレーニングデータセットのストリーミング

トレーニング時のデータセットストリーミングにより、lerobot-train は、完全な Parquet および動画ペイロードをダウンロードする前に LeRobotDataset v3 を消費できます。既存の公開スイッチを介して有効にします:

lerobot-train \
  --dataset.repo_id=OWNER/DATASET \
  --dataset.streaming=true \
  --policy.type=act \
  --output_dir=outputs/train/act_streaming

この機能は--dataset.streaming_encoding=trueとは独立しています。streaming_encodingは録画時の動画の書き込み方法を制御し、dataset.streamingは学習時に既存のデータセットを読み込む方法を制御します。この学習パスでは、録画およびロールアウトのエンコーダーは使用されません。

エポックの読み取り方法

各分散ランクは、決定論的かつフレームバランスの取れた完全なエピソードの集合を所有します。DataLoaderワーカーごとに1つの完全カバレッジプールが用意され、そのワーカーが担当するエピソード部分をミックスしながら、カバレッジエポックごとに選択されたすべてのフレームを1回ずつ訪れます。Parquetの列と圧縮MP4のバイト範囲は、同じバイト認識型アドミッションフロンティアからプリフェッチされます。時間履歴と未来ウィンドウは、完全なエピソード内で解決され、mapスタイルの読み込みと同様の境界パディングマスクも含まれます。

--dataset.streaming=false の場合、デフォルトのマップスタイルの LeRobotDataset の動作は変更されません。

ワーカー、シャード、並行性

各ランクは、完全なエピソードからなるフレームバランスの取れたシャードを所有します。ランク内では、--num_workersがDataLoaderのワーカープロセス数を設定します。これはmapスタイルのデータセットの場合と同様です。各ワーカーは、そのランクのエピソードの互いに重複しないフレームバランスの取れた部分を所有し、それぞれに正確なカバレッジプール、バイト予算の割り当て、スレッドプールが用意されています。

複数のワーカーが必要な理由 サンプルの組み立てとバッチの引き継ぎは Python コードを実行しますが、1 つのプロセスは 1 つのスレッドでしか Python コードを実行できません（GIL の制約）。ワーカーが 1 つの場合、あるランクのデコードスレッドは互いに待機するため、そのランクは約 2 つの CPU コアしか使用できません。各ワーカーは独立したプロセスであるため、N 個のワーカーが並行してサンプルの組み立てとデコードを行います。

プールサイズとバッチの多様性。 streaming_episode_pool_size は各ワーカーのプールを設定します。設定しない場合、ランクは 32 エピソードを混合し、N 個のワーカーのそれぞれがその 32/N を受け取ります。各ワーカーは streaming_prefetch_episodes エピソードをプリフェッチし、独立した Python プロセスとして動作するため、ワーカー数が増えるとメモリ使用量も増加します。3 カメラを持つ 1 つのデータセットでは、4 ワーカーで合計約 8 GB、1 ワーカーで約 3 GB でした。DataLoader は各バッチを 1 つのワーカーから取得するため、バッチは 1 つのワーカーのプールからエピソードが混合されます。プールを大きくすると、各バッチでより多くのエピソードが混合されますが、メモリ使用量が増え、最初のサンプル取得までに多くのデータをフェッチすることになります。

制限。 バイト予算 (streaming_byte_budget_gb) と明示的な video_decoder_cache_size はランクごとに保持されます。各ワーカーにはその 1/N が割り当てられます。ワーカーには少なくとも 1 つのエピソードが必要であるため、lerobot-train は、ランクが持つエピソード数を超える数のワーカーを使用しません。 --num_workers=0 を指定すると、データセットはトレーニングプロセス内に保持されます。この場合、トレーニングスレッドが画像変換を適用するため、同じ --seed を使用すると同じ変換が得られます。

MP4サイドカーと初回実行

Video streaming は MP4 インデックスのサイドカーファイルを使用します。Dataset の初期化では、まずリビジョンをキーとするローカルキャッシュを確認し、次に有効な公開済みサイドカーファイルを探します。どちらも利用できない場合、LeRobot はプロセスロックを保持しながらローカルでサイドカーファイルを構築し、アトミックにインストールします。失敗または中断された構築では、以前の有効なファイルは置き換えられません。同じサイドカーファイルが必要な他のプロセスは、最大 --dataset.streaming_sidecar_lock_timeout_s（デフォルトで 30 分）の間、ロックを待機します。インデックス作成に時間がかかる大規模な Dataset では、この値を引き上げてください。

Hubからのペイロード読み取りは、メタデータスナップショットのコミットに固定されます。バケットインデックスは、コンテンツハッシュを含む新しいオブジェクト一覧と照合されるため、同じパスで動画を置き換えた場合、ファイルサイズが変更されていなくても、そのサイドカーは無効になります。この一覧取得は起動時の処理量を増やしますが、動画ペイロードのダウンロードは行いません。実行中はメタデータとペイロードを変更しないでください。安定したオブジェクトアイデンティティを持たない汎用のリモートファイルシステムでは、各オープン時にインデックスが再構築されます。オブジェクトフィンガープリントを持たない既存のバケットサイドカーは、ローカルで一度だけ再構築されます。更新されたサイドカーの公開は、引き続き明示的なメンテナーの操作です。

Training は読み取り専用です。サイドカーをアップロードしたり、データセットリポジトリを変更したりすることはありません。ノードローカルキャッシュを持つクラスターでは、最初のジョブがノードごとに一度だけビルドを行う場合があります。共有 LeRobot キャッシュを使用することで、この重複を回避できます。

Sidecarの構築では、各ソースインデックスの完了ごとに一時アーカイブに書き込みます。保持されるパース結果は、設定された数を超えません。配列メモリは同時に処理される最大のソース数に依存しますが、アーカイブのメタデータはソース数に応じて増加し続けます。lerobot-build-mp4-sidecarを実行する際は、--workersオプションでパースの並列度を設定してください。ビルドが失敗した場合、以前のSidecarは保持されます。

公開された schema-3 の NPZ アーカイブは、正規の int64 整数配列を保持します。初回のオープン時に、アーカイブは $HF_LEROBOT_HOME/streaming-indexes 配下の読み取り専用メモリマップインデックスに変換されます。派生インデックスは、最小の正確な整数型を使用します。一定の期間、構成オフセット、サイズ、またはバイトオフセットは一度だけ保存されます。読み取り専用の配列ビューは、論理的なサンプル数を保持します。プレゼンテーションタイムスタンプは、正確な float64 値を保持します。同期サンプル配列は、密集型ストレージを保持します。

派生インデックスは別のキャッシュ世代を使用します。古いリーダーは既存のキャッシュ形式を維持します。新しいリーダーによる最初のオープンでアーカイブを一度変換でき、その後のオープンでは再利用されます。変換にはプロセスロックとアトミックな置換を使用します。キャッシュを共有するランクは、同じファイルバックのページを再利用します。各ランクは、そのエピソードで使用されるソースファイルに対してのみレコードを構築します。古い世代と新しい世代のキャッシュは、追加のディスク容量を必要とする場合があります。サイドカーの置換は、ライブリーダーを無効にすることなく、新しい不変世代を作成します。

これにより、すべてのメタデータが一定サイズになるわけではありません。エピソードのメタデータやスパンテーブルは、データセットの規模に応じて依然として増加します。また、シャッフルされたランクシャードは、重複するソースファイルにアクセスする可能性があります。OSに常駐するマッピングされたページもRSSにカウントされるため、ランク間で共有されるメモリを測定する際は、比例セットサイズ（PSS）を使用してください。Parquetの読み取りでは、エピソードの統計情報を使用して行グループをプルーニングし、その後、完全なエピソードに絞り込みます。統計情報がないグループは候補として残るため、有用な境界値がないファイルでは、ファイル全体の予測読み取りが必要になる場合があります。

Datasetのメンテナーは、サイドカーを事前に構築できます:

lerobot-build-mp4-sidecar \
  --repo-id=OWNER/DATASET \
  --revision=COMMIT_SHA \
  --data-root=hf://datasets/OWNER/DATASET@COMMIT_SHA \
  --output=/tmp/dataset-mp4-sidecar.npz

公開は常に明示的に行われます。完全なデータセットのサイドカーを検証した後にのみ、--push を追加してください。 サブセットのサイドカーは公開できません。

データセットリポジトリの場合、--push は専用の lerobot-sidecars ブランチを作成または更新します。 ソースのデータセットブランチ/タグは変更されません。トレーニングは、固定されたビデオソースコミットに紐付けられたインデックスをそこで検索し、使用前に検証します。タグとその解決されたコミットは同じ公開インデックスを共有し、既存のリビジョンキー付きのローカルキャッシュは引き続き再利用可能です。バケットは引き続き meta/mp4-sidecars/ の直下に直接公開します。公開にはリポジトリまたはバケットへの書き込みアクセスが必要ですが、トレーニングには読み取りアクセスのみで十分であり、ブランチの作成やファイルのアップロードは行いません。

Sidecar スキーマ v3 は、B フレームを含む MP4 の合成タイミングおよびエンコーダー遅延編集を保持します。古い Sidecar は自動的に別のリビジョンキー付きキャッシュエントリとして再構築されますが、明示的な公開前にも再構築が必要です。並べ替えられた動画では、取得された各スパンの最後の GOP の末尾が含まれるため、デコードされたフレームのインデックスが連続したままになります。繰り返しのある MP4 編集タイムラインや非単位レートの MP4 編集タイムラインは、正しくアラインされていないフレームを無言のうちに返すのではなく、拒否されます。

設定

本番環境のデフォルト値は以下の通りです:

max_num_shards 以外のすべてのオプションは --dataset.* フラグです。max_num_shards は StreamingLeRobotDataset の引数であり、lerobot-train は --num_workers からこれを設定します（最低 1）。

より均一なエピソードの混合を実現するには、--dataset.streaming_sampling_strategy=round_robin を指定してください。 各ラウンドでは、現在レジデント状態のエピソードがシャッフルされ、それぞれからシャッフルされた1フレームがサンプリングされます。 ラウンド中にアドミットされたエピソードは、次のラウンドに参加します。どちらの戦略も、エポックのカバレッジがちょうど1回ずつであることを保証し、同じリソースの上限を維持します。どちらもグローバルな一様パーミュテーションではありません。ラウンドロビンは有限のプレフィックスにおいて短いエピソードを優先するため、デコーダーのチャーンが増加し、スループットが低下する可能性があります。 順序は --seed (デフォルトは 1000) によって決まり、ポリシーの初期化および他のシード付き乱数生成器と組み合わされます。異なるシードを使用すると、エピソードとアンカーの順序が異なり、同じシードを使用すると順序が再現されます。--seed=null は、固定のストリーミングシード 42 を使用します。チェックポイントから再開する際は、同じ戦略、--seed およびプール設定を維持してください。チェックポイントにはシードが保存されているため、通常の --resume では順序が維持されます。--seed がストリーミングデータセットに到達する前に開始された実行からのチェックポイントでは、その順序にシード 42 が使用されていました: そのチェックポイントを再開する際は、その順序を維持するために --seed=42 を渡してください。

アクティブなエピソードセットは、エピソード数と、サイドカーから計算される正確な合成ミニ MP4 サイズの両方で上限が設定されます。完全なランク予算を超えるエピソードは、トレーニングがそのペイロードを取得する前に失敗します。予約は、保留中のフェッチ、完了したプリフェッチ、およびまだデコード中のエピソードをカバーします。予算が満杯になると、投機的プリフェッチは一時停止し、置換の受け入れは最後のデコードがそのバイトを解放するのを待ちます。EpisodeByteCache.reserved_bytes はこれらの予約を報告し、resident_bytes は、最初のサンプルが要求される前であっても、すべての完了したペイロードを報告します。

これは圧縮ペイロードの予算であり、プロセスのRSS制限ではありません: range-fetch/合成用のスクラッチバッファ、デコーダーの状態、Parquetテーブル、デコードされたテンソルには追加のメモリが必要です。デコーダーのLRUとデコード済みサンプルのキューには、それぞれ独立した制限があります。 メモリに制約のあるホストでは、より小さいプールまたは予算から開始してください:

lerobot-train \
  --dataset.repo_id=OWNER/DATASET \
  --dataset.streaming=true \
  --dataset.streaming_episode_pool_size=16 \
  --dataset.streaming_prefetch_episodes=4 \
  --dataset.streaming_byte_budget_gb=4 \
  --dataset.streaming_decode_threads=2 \
  --dataset.streaming_decoded_queue_size=8 \
  --num_workers=4 \
  --policy.type=act \
  --output_dir=outputs/train/act_streaming

HF Storage Bucket に保存された完全なデータセットについては、バケットの識別子とリポジトリタイプを使用するか、バケット URI を --dataset.root=hf://buckets/OWNER/BUCKET として渡します。メタデータ（meta/）、Parquet ファイル、動画ファイルはすべて、そのバケットのルートに存在する必要があります。

lerobot-train \
  --dataset.repo_id=OWNER/BUCKET \
  --dataset.repo_type=bucket \
  --dataset.streaming=true \
  --policy.type=act \
  --output_dir=outputs/train/act_bucket_streaming

Hubのデータセットはデフォルトでrepo_type=datasetを使用します。完全なローカルデータセットの場合、--dataset.rootをそのディレクトリに設定してください。学習では、メタデータとペイロードを同じソースから解決します。そのため、ストリーミング用のペイロード位置を指定するフラグは不要です。

再開とシャッフルの移行

以前のストリーミングリーダーは、有界な行シャッフルバッファを使用していました。一方、エピソードリーダーは、シードとエポックから導出される決定論的な完全カバレッジ順序を持っています。チェックポイントからの再開では、チェックポイントのバッチサイズを使用して、ランクごとのサンプルオフセットを復元します。複数のワーカーを使用する場合、DataLoaderがワーカーから順番にバッチを取得するため、各ワーカーはそれに基づいて独自のオフセットを導出します。分散ワールドサイズ、バッチサイズ、または--num_workersを変更すると、所有権やバッチの境界が変わります。サンプル単位の正確な比較を行う場合は、同じワールドサイズ、バッチサイズ、ワーカー数で再開してください。パフォーマンスを比較する場合は、内部フェッチの並行度も同じに保ってください。

チェックポイントのメタデータには、以前のマルチワーカーサンプラーの識別情報が含まれていません。これらのチェックポイントは、ワールドサイズとバッチサイズのチェックによって確実に区別することができず、サンプル単位の正確なストリーミング再開には使用しないでください。代わりに、保存されたポリシーの重みから新しい実行を開始してください。再開の保証はアンカーの順序に関するものであり、確率的な画像変換のビット単位の再現ではありません。ランダムな画像変換はデコードスレッド上で、デコードと並行して実行されるため、固定されたシードでは、実行から実行へ同じ拡張を再現することはできません。

トラブルシューティング