Потоковая обработка обучающих данных

Стриминг датасета во время обучения позволяет lerobot-train использовать LeRobotDataset v3 без предварительной загрузки полного содержимого Parquet и видео. Включите его с помощью существующего публичного переключателя:

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 управляет способом чтения существующего набора данных во время обучения. Кодировщики записи и rollout не используются в этом пути обучения.

Как читается эпоха

Каждый распределённый ранг владеет детерминированным, сбалансированным по кадрам набором полных эпизодов. Один пул с точным покрытием на каждый воркер DataLoader смешивает часть этих эпизодов, принадлежащую данному воркеру, при этом посещая каждый выбранный кадр один раз за эпоху покрытия. Колонки Parquet и сжатые байтовые диапазоны MP4 предварительно загружаются из того же байтово-ориентированного фронта допуска. Окна временной истории и будущего разрешаются внутри полного эпизода, включая те же маски отступов на границах, что и при загрузке в стиле map.

Поведение LeRobotDataset в режиме карты по умолчанию не изменяется, если задан параметр --dataset.streaming=false.

Рабочие процессы, фрагменты и параллелизм

Каждый ранг владеет сбалансированным по кадрам шардом полных эпизодов. Внутри ранга --num_workers задаёт количество процессов-воркеров DataLoader, как и для наборов данных map-style. Каждый воркер владеет непересекающейся, сбалансированной по кадрам частью эпизодов ранга, со своим пулом точного покрытия, своей долей байтового бюджета и своими пулами потоков:

Почему несколько воркеров. Сборка выборок и передача батчей выполняются кодом на Python, а один процесс выполняет код на Python в одном потоке за раз (из-за GIL). При одном воркере потоки декодирования ранга ждут друг друга, поэтому ранг использует лишь около 2 CPU-ядер. Каждый воркер — это отдельный процесс, поэтому N воркеров собирают и декодируют выборки параллельно.

Размер пула и разнообразие батчей. streaming_episode_pool_size задаёт пул для каждого воркера. Если вы не зададите это значение, ранг смешает 32 эпизода, и каждый из N воркеров получит 32/N из них. Каждый воркер также предзагружает streaming_prefetch_episodes эпизодов и является отдельным процессом Python, поэтому больше воркеров означает больше памяти: на одном датасете с 3 камерами 4 воркера использовали около 8 ГБ в сумме, тогда как один воркер — около 3 ГБ. DataLoader берёт каждый батч из одного воркера, поэтому батч содержит эпизоды из пула одного воркера. Больший пул смешивает больше эпизодов в каждом батче, но потребляет больше памяти и загружает больше данных до получения первого сэмпла.

Ограничения. Бюджет в байтах (streaming_byte_budget_gb) и явный video_decoder_cache_size остаются на ранг: каждый воркер получает 1/N от них. Воркеру нужен как минимум один эпизод, поэтому lerobot-train использует не больше воркеров, чем эпизодов на ранге. --num_workers=0 удерживает датасет в процессе обучения. Там поток обучения применяет преобразования изображений, поэтому один и тот же --seed даёт одни и те же преобразования.

MP4-файлы-спутники и первый запуск

Потоковая передача видео использует MP4-индекс в виде вспомогательного файла (sidecar). При инициализации датасета сначала проверяется локальный кэш, привязанный к ревизии, затем ищется опубликованный корректный sidecar. Если ни один из них недоступен, LeRobot локально строит sidecar, удерживая блокировку процесса, и устанавливает его атомарно. Сбой или прерывание процесса построения не заменяет предыдущий корректный файл. Другие процессы, которым требуется тот же sidecar, ожидают освобождения блокировки не более --dataset.streaming_sidecar_lock_timeout_s (по умолчанию 30 минут). Увеличьте это значение для больших датасетов, индексация которых занимает больше времени.

Чтение данных Hub привязано к коммиту снимка метаданных. Индексы Bucket проверяются по свежему списку объектов, включая хеши содержимого, поэтому замена видео по тому же пути инвалидирует его sidecar, даже если размер файла не изменился. Этот список добавляет работу при запуске, но не скачивает данные видео. Не изменяйте метаданные и данные во время выполнения. Универсальные удалённые файловые системы без стабильных идентификаторов объектов перестраивают свой индекс при каждом открытии. Существующие sidecar для Bucket без отпечатков объектов перестраиваются локально один раз; публикация обновлённого sidecar остаётся явным действием сопровождающего.

Обучение выполняется только для чтения: оно никогда не загружает sidecar и не изменяет репозиторий набора данных. На кластере с локальными для узлов кэшами первая задача может выполнять сборку один раз на узел. Общий кэш LeRobot позволяет избежать этого дублирования.

Сборка sidecar записывает каждый завершённый исходный индекс в временный архив. Он хранит не более заданного количества результатов разбора. Объём памяти для массивов зависит от наибольшего числа одновременных источников; метаданные архива продолжают расти с увеличением количества источников. Используйте --workers, чтобы задать параллелизм разбора при запуске lerobot-build-mp4-sidecar. При сбое сборки предыдущий sidecar сохраняется.

Опубликованные архивы NPZ со схемой 3 сохраняют канонические целочисленные массивы типа int64. При первом открытии архив преобразуется в индекс только для чтения с отображением в память (memory-mapped) в каталоге $HF_LEROBOT_HOME/streaming-indexes. Производный индекс использует наименьшие точные целочисленные типы. Постоянная длительность, смещение композиции, размер или байтовое смещение хранятся один раз. Массивы только для чтения сохраняют логическое количество образцов. Временные метки отображения (presentation timestamps) сохраняют точные значения типа float64. Массивы синхронизированных образцов сохраняют плотное хранение.

Производный индекс использует отдельное поколение кэша. Старые читатели сохраняют существующий формат кэша. При первом открытии с новым читателем архив может быть преобразован один раз; последующие открытия переиспользуют его. Преобразование использует блокировку процесса и атомарную замену. Ранги, разделяющие кэш, переиспользуют одни и те же страницы, основанные на файлах. Каждый ранг конструирует записи только для исходных файлов, используемых его эпизодами. Старые и новые поколения кэша могут требовать дополнительного места на диске. Замена sidecar-файла создает новое неизменяемое поколение без инвалидации активных читателей.

Это не делает все метаданные фиксированного размера: метаданные эпизодов и таблицы диапазонов по-прежнему растут вместе с набором данных, а перемешанные ранговые шарды могут затрагивать пересекающиеся исходные файлы. Отображенные страницы, находящиеся в памяти ОС, также учитываются в RSS; при измерении памяти, разделяемой рангами, используйте пропорциональный размер набора (PSS). При чтении Parquet группы строк отбраковываются с использованием статистики эпизодов, после чего выполняется фильтрация до полного эпизода. Группы без статистики остаются кандидатами, поэтому для файлов без полезных границ может потребоваться чтение всего файла с проекцией.

Разработчики датасетов могут заранее создать sidecar:

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 только после проверки sidecar-файла полного набора данных. Sidecar-файлы подмножеств не могут быть опубликованы.

Для репозиториев датасетов --push создаёт или обновляет выделенную ветку lerobot-sidecars. Исходная ветка/тег датасета не изменяется. Во время обучения система ищет там индекс, привязанный к зафиксированному коммиту источника видео, и проверяет его перед использованием. Тег и его разрешённый коммит используют один и тот же опубликованный индекс; существующие локальные кэши, привязанные к ревизии, остаются используемыми. Бакеты по-прежнему публикуются непосредственно в meta/mp4-sidecars/. Для публикации требуется право на запись в репозиторий или бакет; для обучения достаточно права на чтение, при этом не создаётся ветка и не загружаются файлы.

Схема sidecar v3 сохраняет тайминги композиции MP4 и правки, связанные с задержкой кодировщика, включая видео с B-кадрами. Старые sidecar-файлы автоматически пересобираются в отдельный элемент кэша, привязанный к ревизии; их также необходимо пересобрать перед явной публикацией. В переупорядоченные видео включается конец последнего GOP в каждом полученном диапазоне, чтобы индексы декодированных кадров оставались непрерывными. Повторяющиеся или нелинейные по времени таймлайны правок MP4 отклоняются, а не возвращают неверно выровненные кадры без предупреждения.

Конфигурация

Значения по умолчанию для продакшена:

Все параметры, кроме max_num_shards, являются флагами --dataset.*. max_num_shards — это аргумент StreamingLeRobotDataset; lerobot-train устанавливает его на основе --num_workers (не менее 1).

Для более равномерного смешивания эпизодов используйте --dataset.streaming_sampling_strategy=round_robin. Каждый раунд перемешивает текущие резидентные эпизоды и извлекает один перемешанный кадр из каждого; эпизоды, принятые в течение раунда, попадают в следующий. Обе стратегии обеспечивают покрытие эпохи ровно один раз и сохраняют те же границы ресурсов. Ни одна из них не является глобальной равномерной перестановкой: round-robin предпочитает более короткие эпизоды в конечных префиксах и может увеличивать оборот декодера и снижать пропускную способность. Порядок определяется параметром --seed (по умолчанию 1000), вместе с инициализацией политики и другими генераторами случайных чисел с заданным seed. Разные seed дают разные порядки эпизодов и якорей, а один и тот же seed повторяет порядок. --seed=null использует фиксированный seed стриминга 42. При возобновлении с контрольной точки сохраняйте ту же стратегию, --seed и настройки пула. Контрольная точка хранит seed, поэтому обычный --resume сохраняет порядок. Контрольная точка из запуска, начатого до того, как --seed достиг стримингового набора данных, использовала seed 42 для своего порядка: при возобновлении передавайте --seed=42, чтобы сохранить этот порядок.

Активный набор эпизодов ограничен как количеством эпизодов, так и точными размерами синтезированных мини-MP4, вычисленными из sidecar. Эпизод, превышающий полный бюджет ранга, завершается с ошибкой до того, как обучение запросит его данные. Резервирование охватывает ожидающие загрузки, завершённые предварительные загрузки и эпизоды, которые всё ещё декодируются. При полном использовании бюджета спекулятивная предварительная загрузка приостанавливается, а приём замены ожидает освобождения байтов последним декодированием. 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 в его каталог. Обучение извлекает метаданные и данные из одного и того же источника; отдельный флаг для указания местоположения потоковых данных не требуется.

Возобновление и перемешивание миграции

Ранее используемый потоковый читатель применял ограниченный буфер перемешивания строк. Читатель эпизодов, в свою очередь, использует детерминированный порядок полного покрытия, выводимый из seed и epoch. При восстановлении из контрольной точки смещение выборки для каждого ранга восстанавливается с использованием размера батча из контрольной точки. При наличии нескольких воркеров каждый воркер выводит собственное смещение на основе этого значения, поскольку DataLoader получает батчи от своих воркеров по очереди. Изменение размера распределённого мира, размера батча или --num_workers меняет распределение ответственности или границы батчей. Для сравнений с точностью до выборки восстанавливайте работу с тем же размером мира, размером батча и количеством воркеров. При сравнении производительности сохраняйте ту же внутреннюю параллельность получения данных.

Метаданные контрольной точки не идентифицируют ранее использовавшийся сэмплер для нескольких рабочих процессов. Контрольные точки не могут быть надёжно различены по проверкам размера мира и размера батча, поэтому их нельзя использовать для возобновления потоковой передачи с точным совпадением выборок. Вместо этого начните новый прогон из сохранённых весов политики. Гарантия возобновления касается порядка якорей, а не побитового воспроизведения стохастических преобразований изображений. Случайные преобразования изображений выполняются на потоках декодирования параллельно с декодированием, поэтому фиксированный сид не воспроизводит одни и те же аугментации от прогона к прогону.

Устранение неполадок