RFC: 支持 HuggingFace Streaming 数据集接入 Hyper-Parallel 训练流程

RFC：支持 HuggingFace Streaming 数据集接入 Hyper-Parallel 训练流程

1. 基本信息

项目	内容	备注
作者	Ruanzhixin
相关模块	trainer / data / scripts / examples / tests
相关 issue / PR	任务参考：#2094；实现 PR：feat/HF_Streaming_Datasets_Support
适用后端	PT 优先，MS 保持接口兼容	本期以 PyTorch 路径为主验证

2. 背景

HyperParallel 当前暂无内建的 HuggingFace Streaming 数据链路。对于超大语料预训练、SFT 以及多模态训练场景，用户通常希望直接通过 datasets.load_dataset(..., streaming=True) 读取数据，而不是先将完整数据集下载并落盘。

HF Streaming 的主要价值包括：

1. 降低本地存储压力，适合超大规模语料。
2. 复用 HuggingFace 原生数据集定义、解析和预处理能力。
3. 降低用户接入成本，使现有训练脚本可直接消费 streaming 数据集。
4. 为后续更通用的数据接入能力提供基础路径。

但 streaming 数据接入训练流程时存在几个核心问题：

1. 分布式训练下需要支持 rank 级分片，避免不同 rank 重复消费样本。
2. 需要支持 epoch reset，保证多轮训练行为正确。
3. 需要支持基础的一致性 shuffle。
4. 需要支持断点恢复时的数据状态回放。
5. 需要在尽量不重构现有 trainer 主流程的前提下完成接入。

本 RFC 要解决的问题是：在 HyperParallel 现有 trainer / integration 路径上接入 HuggingFace Streaming 数据能力，打通文本训练与多模态训练的基础链路，并提供最小可复现样例和基础测试。

完成后的成功标准包括：

1. 训练脚本可直接消费 HF Streaming 数据集。
2. 数据集加载和处理尽量调用 HuggingFace 原生能力。
3. 支持 rank-aware shard / filter 和 epoch reset。
4. 支持基础 dataset state roundtrip。
5. 提供至少 1 个端到端可复现样例和基础测试。
6. 跑通一个基础文本训练用例和一个变形用例。

3. 目标和非目标

3.1 目标

1. 提供 HF Streaming 数据集读取能力，支持 datasets.load_dataset(..., streaming=True)。
2. 在 HyperParallel trainer 流程中接入 HF iterable dataset，而不改变现有训练入口脚本。
3. 支持 rank-aware shard / filter，满足分布式训练场景的数据消费隔离。
4. 支持 epoch reset，保证多 epoch 训练时 streaming 行为正确。
5. 支持基础 dataset state 保存与恢复，用于断点恢复。
6. 支持文本训练场景的数据处理接入。
7. 支持多模态 / VL 场景的数据处理接入，保留 HuggingFace tokenizer / processor 相关能力。
8. 建立基础功能、样例和测试验证。

3.2 非目标

1. 本期不重构 HyperParallel 全量数据抽象层，优先在 trainer / integration 侧完成接入。
2. 本期不覆盖所有 HuggingFace dataset builder 类型，只优先支持当前任务所需主路径。
3. 本期不替代现有本地数据集训练方式，不启用 HF streaming 时现有行为保持不变。
4. 本期不优先做最终性能最优，先保证功能正确、训练可跑通、链路可恢复。
5. 本期不单独设计新的训练入口脚本。

4. 相关实现参考

来源	做法	限制
HuggingFace datasets streaming	使用 load_dataset(..., streaming=True) 构建 iterable dataset	需要上层自行管理分布式分片、epoch 语义和恢复状态
HuggingFace tokenizer / processor	复用原生 tokenizer / processor 做文本和多模态预处理	训练侧仍需做样本结构适配
HyperParallel trainer	现有 train_lm.py / train_vl.py 与 trainer 主流程	当前缺少 streaming 数据接入层
分布式数据分片	基于 rank / world_size 做 shard / filter	需保证多 rank 行为稳定、一致
本地 yaml 配置驱动训练	通过 yaml 指定模型、数据和训练参数	需扩展 HF streaming 相关配置项

来源	对本 RFC 的参考
HuggingFace datasets streaming	作为 streaming 数据源入口
HuggingFace tokenizer / processor	作为数据处理能力来源
HyperParallel trainer	作为主集成路径
分布式数据分片	作为分布式消费策略
本地 yaml 配置驱动训练	作为对外接入方式

5. 对外接口

5.1 接口定义

建议通过 yaml 配置扩展 HF Streaming 数据集能力，例如：

data:
  backend: hf
  streaming: true
  dataset_name_or_path: /path/or/hf_repo
  split: train
  shuffle: true
  shuffle_seed: 42
  resume_state_path: null

输入参数 / 配置项	类型	默认值
backend	str	现有本地实现
streaming	bool	false
dataset_name_or_path	str	无
split	str	train
shuffle	bool	false
shuffle_seed	int	42
resume_state_path	str/null	null

输入参数 / 配置项	是否必填	含义
backend	否	数据后端类型
streaming	否	是否启用 HF streaming
dataset_name_or_path	是	HF 数据集名称或本地路径
split	否	数据 split
shuffle	否	是否打乱数据
shuffle_seed	否	shuffle 种子
resume_state_path	否	dataset state 保存 / 恢复路径

输入参数 / 配置项	合法范围	错误处理
backend	hf / 现有本地后端	非法值报配置错误
streaming	true/false	与 backend=hf 组合校验
dataset_name_or_path	可被 HuggingFace 正常解析	无法加载时报错
split	合法 split 名称	split 不存在时报错
shuffle	true/false	非法值报配置错误
shuffle_seed	任意整数	类型错误时报错
resume_state_path	合法路径或空	恢复失败时报错或执行回退策略

5.2 使用示例

文本训练：

torchrun ... scripts/train_lm.py examples/qwen3_5_0_8b_base/train_hf_streaming.yaml

多模态训练：

torchrun ... scripts/train_vl.py examples/qwen3_vl_30b_a3b_instruct/train_hf_streaming.yaml

5.3 接口说明

接口沿用现有训练脚本和 yaml 配置方式，用户无需引入新的训练入口。启用 HF Streaming 后，trainer 在内部切换数据集构建逻辑，通过 HuggingFace iterable dataset 提供样本，并在训练侧完成分片、重置、恢复和格式适配。

6. 方案设计

6.1 总体流程

@startuml
start
:读取 YAML 配置;
if (是否启用 HF Streaming?) then (是)
:构建 HF Streaming 数据集;
:按 Rank 分片与过滤;
:执行 Shuffle 与 Epoch 管理;
:进行文本 / 多模态样本转换;
:接入训练循环;
:执行前向与反向训练;
if (是否保存断点?) then (是)
:保存数据集状态;
endif
if (是否进入新 Epoch?) then (是)
:重置 Streaming 迭代器;
endif
else (否)
:沿用现有本地数据集流程;
endif
stop
@enduml

6.2 架构参考

@startuml
left to right direction
rectangle "用户配置 YAML" as UserConfig
rectangle "训练配置解析" as ConfigParser
rectangle "HF Streaming 构建器" as HFBuilder
rectangle "HuggingFace 流式数据集加载" as HFLoader
rectangle "可迭代数据集" as IterableDataset
rectangle "按 Rank 分片与过滤" as RankShard
rectangle "Epoch 重置与 Shuffle" as EpochControl
rectangle "数据集状态" as DatasetState
rectangle "文本 / 多模态样本转换" as SampleTransform
rectangle "训练器基类" as TrainerBase
rectangle "训练循环" as TrainLoop
rectangle "断点保存与恢复" as CheckpointResume

UserConfig --> ConfigParser
ConfigParser --> HFBuilder
HFBuilder --> HFLoader
HFLoader --> IterableDataset
IterableDataset --> RankShard
RankShard --> EpochControl
EpochControl --> DatasetState
EpochControl --> SampleTransform
SampleTransform --> TrainerBase
TrainerBase --> TrainLoop
TrainLoop --> CheckpointResume
CheckpointResume --> DatasetState
DatasetState --> HFBuilder
@enduml
其中：

1. HF Streaming Builder 负责数据集加载、分片、shuffle 与 epoch reset。
2. Dataset State 负责最小必要状态的保存与恢复。
3. Text / VL Transform 负责将原始样本转换为训练可消费格式。
4. Trainer Base 负责在现有训练框架中组织整个数据链路。

6.3 时序参考

@startuml
participant "用户配置" as A
participant "训练器" as B
participant "HF Streaming 构建器" as C
participant "可迭代数据集" as D
participant "数据集状态" as E
participant "文本/多模态转换" as F
participant "训练循环" as G

A -> B: 读取 YAML 配置
B -> C: 构建 HF Streaming 数据集
C -> D: 加载流式数据集
C -> D: 按 Rank 分片与过滤
C -> E: 初始化数据集状态
B -> G: 启动训练循环

loop 每个训练步
G -> D: 拉取原始样本
D --> G: 返回样本
G -> F: 执行样本转换
F --> G: 返回训练输入
G -> E: 更新消费进度
end

opt Epoch 结束
G -> C: 重置迭代器
C -> D: 重新构建或重置数据流
end

opt 保存断点
G -> E: 保存数据集状态
end

opt 恢复训练
B -> E: 读取数据集状态
B -> C: 恢复构建器状态
C -> D: 从恢复位置继续消费
end
@enduml

6.4 关键逻辑

1. HF Streaming Builder
   根据配置创建 HuggingFace iterable dataset，并负责 rank-aware 分片与 epoch reset。

2. Dataset State
   保存当前 epoch、当前消费进度以及必要的 shuffle / cursor 相关状态。

3. Text / VL Transform
   将 streaming 样本转为模型训练所需结构，文本场景侧重 tokenizer 处理，多模态场景侧重 tokenizer / processor 和多模态字段保留。

4. Trainer Integration
   让现有 trainer 无需感知底层是本地数据集还是 HF streaming 数据集，只消费统一格式样本。

6.5 代码改动点

模块	改动内容	是否影响已有行为
hyper_parallel/data	新增或扩展 HF streaming 数据接入逻辑	否，新增能力
hyper_parallel/trainer/base.py	接入 streaming dataset 构建与恢复逻辑	仅在启用 HF streaming 时生效
hyper_parallel/trainer/config.py	增加 HF streaming 相关配置项	否
hyper_parallel/trainer/vl_trainer.py	适配多模态 HF 数据转换逻辑	仅影响对应场景
scripts/train_lm.py	支持文本 HF streaming 配置训练	否
scripts/train_vl.py	支持 VL HF streaming 配置训练	否
examples/...	增加基础样例和变形样例	否
tests/ut/trainer/...	增加基础测试	否

6.6 方案取舍

方案	优点	缺点
直接在底层重构统一 streaming 数据层	抽象更彻底、长远更通用	改动面大，超出本期任务范围
在 trainer / integration 侧优先接入 HF streaming	改动集中、落地快、验证路径清晰	通用性不如彻底重构
仅支持本地落盘数据集	实现简单	无法满足超大数据集 streaming 需求
完全自行实现数据处理逻辑	可完全受控	与 HF 原生能力重复建设，维护成本高

方案	是否选择	原因
直接在底层重构统一 streaming 数据层	否	本期目标是先完成能力接入
在 trainer / integration 侧优先接入 HF streaming	是	更符合当前任务目标
仅支持本地落盘数据集	否	不满足任务要求
完全自行实现数据处理逻辑	否	本期明确优先复用 HF 能力

本方案的主要代价在于 trainer 侧需要引入 streaming 状态管理与数据结构适配逻辑，但改动范围相对可控，且更符合当前任务验收目标。

7. 组件依赖

依赖组件	强依赖 / 弱依赖	当前状态
HuggingFace datasets	强依赖	已具备成熟 streaming 能力
transformers	强依赖	已具备 tokenizer / processor 能力
HyperParallel trainer	强依赖	已具备现有训练主流程
PyTorch 后端	强依赖	当前主验证路径
MindSpore 后端	弱依赖	本期保持接口兼容
Ascend 多卡环境	验证依赖	用于双卡训练验证

依赖组件	未 ready 时本期能力
HuggingFace datasets	无法完成 HF streaming 接入
transformers	文本 / 多模态处理受限
HyperParallel trainer	无法完成主流程集成
PyTorch 后端	可先完成主链路
MindSpore 后端	非主验收路径
Ascend 多卡环境	无法完成分布式验收

本期最小可交付能力包括：

1. 单卡文本 HF Streaming 训练可跑通。
2. 单卡多模态 HF 数据训练可跑通。
3. 双卡基础文本 HF Streaming 训练可验证。
4. 基础 dataset state roundtrip 可验证。

8. 约束与兼容性

类型	内容	备注
不支持项	本期不覆盖所有 HF dataset builder 类型，不重构全量数据抽象层
功能收益	提供 HF Streaming 数据接入能力，降低大规模数据集落盘需求
稳定性收益	可在现有训练入口下统一使用 HF 数据链路
性能风险	streaming 首期重点是功能正确，性能优化后续再评估
PT / MS 差异	本期以 PT 路径为主验证，MS 保持接口兼容
与现有行为不一致	启用 HF streaming 后，数据来源与迭代方式从本地数据集切换为 iterable dataset；不启用时现有行为保持不变

其他约束：

1. rank-aware shard / filter 需要保证分布式样本消费行为稳定。
2. epoch reset 需要显式管理，不可依赖 map-style dataset 语义。
3. dataset state 恢复仅保证基础训练可继续，不承诺恢复所有外部随机细节。
4. 多模态 transform 需要保留图像、文本等关键字段，不可在转换中丢失。
5. 若恢复路径与原路径的数据消费顺序不一致，应明确报错或给出可接受回退行为，而不是静默错用数据。

9. 验证设计

9.1 用例分层

用例级别	数量	覆盖内容
UT	4+	HF streaming builder、dataset state roundtrip、VL transform、样本结构保持
Level0	2+	文本 HF streaming 基础训练、多模态 HF 数据基础训练
Level1	2+	双卡文本 streaming 训练、变形用例验证

用例级别	通过标准
UT	单元测试通过
Level0	单卡训练可跑通
Level1	分布式训练可跑通

9.2 交互验证

组合	是否验证	通过标准
HF streaming + train_lm.py	是	文本训练可直接消费 streaming 数据
HF streaming + train_vl.py	是	多模态训练可直接消费 HF 数据
HF streaming + tokenizer / processor	是	数据处理链路正确
HF streaming + distributed training	是	rank-aware 分片行为正确
HF streaming + resume	是	dataset state 可恢复
PT / MS 对齐	否，本期	本期以 PT 主路径为主，MS 保持接口兼容

9.3 功能 / 稳定性验证

场景	基线	开启本特性
文本 streaming 训练	本地数据训练	HF streaming 训练
多模态 streaming 训练	本地数据训练	HF streaming 训练
双卡文本 streaming 训练	单卡可运行	双卡 streaming
resume 场景	无恢复	state roundtrip

场景	指标	通过标准
文本 streaming 训练	是否可完成训练	Training completed
多模态 streaming 训练	是否可完成训练	Training completed
双卡文本 streaming 训练	分布式训练是否正常	可完成多 step 训练
resume 场景	是否可继续训练	恢复链路有效

10. 实现计划

PR	内容	依赖
PR1	接入 HF Streaming Builder、基础配置项和文本训练主链路	无
PR2	增加 dataset state roundtrip、epoch reset 和多模态 transform 支持	PR1
PR3	完成双卡 Ascend 训练验证、补充样例与测试	PR2

PR	验证
PR1	UT + 单卡文本训练
PR2	UT + 单卡多模态训练
PR3	Level1 验证

schema_version: 1
source: gitcode
gitcode_repo: mindspore/hyper-parallel
gitcode_issue: 311
source_url: https://gitcode.com/mindspore/hyper-parallel/issues/311

Start by reading hyper_parallel/data and hyper_parallel/trainer/base.py, then inspect trainer/config.py, vl_trainer.py, scripts/train_lm.py, and scripts/train_vl.py to trace the existing dataset path. Run the tests under tests/ut/trainer and review the example YAML files. Done requires text and multimodal HF streaming training, rank-aware sharding, epoch reset, dataset state roundtrip, and the listed single- and dual-card validations.