在大语言模型、多模态大模型的训练、微调与推理全流程中数据预处理是决定模型收敛速度、泛化能力与最终效果的前置核心环节。原始文本、图像、音频等原始数据格式杂乱、长度不一、存在噪声无法直接输入大模型网络必须经过清洗、分词、编码、对齐、增强、批量组织等一系列变换操作。昇思 MindSpore 内置的mindspore.dataset模块是一套面向高性能、分布式、国产硬件深度适配的数据处理引擎专为深度学习场景设计不仅提供高效的数据加载、流水线调度能力还集成了丰富的数据变换Transform接口支持文本、图像、多模态等各类大模型数据预处理需求。相较于传统 Python 原生数据处理方案mindspore.dataset采用C 后端加速、多线程并行、流水线预取、硬件亲和调度架构可大幅降低 Python 解释层开销实现数据处理与模型计算并行执行充分释放昇腾 AI 芯片、鲲鹏服务器的算力避免出现 “算力空转、数据拖后腿” 的瓶颈。围绕mindspore.dataset的架构原理、数据变换分类、大模型专属预处理流程、完整代码实战、性能调优与工程化落地展开全面讲解覆盖文本大模型、多模态大模型主流预处理场景。一、mindspore.dataset 整体架构与数据变换原理1. 模块整体架构mindspore.dataset是 MindSpore 生态独立的数据处理组件整体分为数据源层、数据加载层、数据变换层、批次组织层、输出对接层五大层级形成端到端流水线处理链路各环节异步执行、互不阻塞。数据源层支持读取本地文本文件、JSON、CSV、Parquet、标准数据集、分布式共享存储等多种数据源适配大模型海量离线数据集存储形态数据加载层依托多线程 / 多进程实现并行读取支持分片加载天然适配分布式训练场景数据变换层是本文核心集成通用变换、文本变换、图像变换、多模态变换、自定义变换五大类算子所有变换算子支持链式调用、组合编排批次组织层完成数据补齐、截断、动态 Batch、静态 Batch、打乱重排等操作输出对接层将处理完成的张量数据直接输送至模型计算图支持数据下沉Dataset Sink减少主机与昇腾设备之间的数据拷贝开销。整个数据处理流水线采用懒执行机制仅定义变换流程不立即执行计算只有当模型开始迭代取数时才按流水线顺序批量执行所有变换逻辑有效节省内存资源适合 TB 级大模型数据集处理。2. 数据变换核心分类针对大模型主流场景mindspore.dataset的数据变换算子可划分为四大类别覆盖全预处理链路第一类为基础通用变换适用于所有数据类型包含类型转换、维度变换、数据打乱、重命名列、字段过滤等用于统一数据格式第二类为文本专属变换是大语言模型LLM核心算子包含分词、词表映射、文本截断、填充、特殊符添加、序列掩码、正负样本构造等完成自然语言数据向模型输入张量的转换第三类为图像 / 音频变换面向文生图、图文理解等多模态大模型实现图像解码、归一化、缩放、裁剪、数据增强、张量转换等操作第四类为自定义变换支持用户编写 Python 函数或 C 算子接入流水线灵活实现业务专属清洗规则弥补内置算子无法覆盖的个性化需求。同时所有内置变换算子均经过昇腾 CANN、鲲鹏指令集深度优化执行效率远高于原生 Python 循环、Pandas、Torchvision 等方案在大规模数据集下性能优势尤为明显。3. 大模型预处理核心痛点与优化思路大模型预处理普遍存在三大痛点一是海量文本数据分词、编码耗时久单线程处理速度无法匹配模型训练吞吐二是长文本序列长度不统一直接输入网络会造成计算浪费三是分布式训练时数据集分片不均、打乱失效影响训练效果。mindspore.dataset针对性给出解决方案通过多线程并行变换加速分词与编码内置动态截断、动态 Padding 算子统一序列长度支持全局打乱、分布式分片加载保证多卡训练数据分布均匀结合流水线预取在模型执行前提前完成数据预处理实现 “计算、数据” 双流水线并行。二、大语言模型LLM文本数据预处理流程大语言模型的标准预处理流程为原始文本读取 → 数据清洗过滤 → 文本分词 → Token 编码 → 序列截断 / 补齐 → 注意力掩码生成 → 构造训练样本 → 批量输出张量。该流程全部可基于mindspore.dataset链式完成也是对话模型、预训练模型、微调模型最常用的链路。结合 MindSpore 设计规范文本预处理分为离线预处理与在线预处理。离线预处理适合固定数据集提前完成分词编码并存储训练时直接加载在线预处理即训练过程中实时执行变换灵活性更高支持动态数据增强是动态数据集、增量训练的首选方案下文代码以在线预处理为核心演示。2.1 关键文本变换算子说明TextTokenizer分词算子支持 BPE、WordPiece、SentencePiece 等大模型主流分词算法对接开源词表Lookup词表映射算子将分词后的 Token 字符串转换为模型对应的整数 IDPadSequence序列补齐算子将短文本统一填充至固定长度Slice序列截断算子对超长文本进行头部 / 尾部截断控制输入序列长度Mask掩码算子生成注意力掩码、损失掩码区分有效 Token 与 Padding 占位符Rename、Filter字段重命名、脏数据过滤完成原始数据清洗。三、完整代码实战LLM 文本数据变换与预处理本章节基于 MindSpore 最新版本实现数据集加载、数据清洗、分词编码、序列处理、掩码生成、批量输出全流程代码模拟通用大语言模型微调场景代码可直接在昇腾服务器、openEuler 系统中运行。3.1 环境准备与依赖导入首先完成环境初始化指定硬件平台、运行模式导入数据集、文本变换、分词器等核心模块import mindspore as ms from mindspore import context import mindspore.dataset as ds import mindspore.dataset.text as text from mindspore.dataset.transforms import transforms as C from mindspore.dataset.text import transforms as T # 全局环境配置昇腾硬件 静态图模式适配大模型训练 context.set_context( modecontext.GRAPH_MODE, device_targetAscend, device_id0, max_call_depth2000 ) # 全局参数配置大模型常用超参 VOCAB_FILE ./vocab.txt # 词表文件路径 DATA_PATH ./train_data.txt # 原始文本数据集 MAX_SEQ_LEN 256 # 模型最大输入序列长度 BATCH_SIZE 16 # 批次大小 PAD_ID 0 # 填充符ID UNK_ID 1 # 未知词ID3.2 自定义数据清洗函数自定义变换原始文本往往包含空行、特殊符号、过长 / 过短脏数据通过pyfunc接入自定义清洗逻辑作为数据变换的第一步def text_clean_func(text_data): 自定义文本清洗过滤空文本、去除首尾空格、过滤极短文本 text_data text_data.strip() # 过滤空数据与长度小于5的无效文本 if len(text_data) 5: return None return text_data # 将清洗函数封装为dataset可调用的自定义变换 clean_op C.pyfunc( functext_clean_func, output_typesms.string, output_shapes() )3.3 构建完整数据预处理流水线采用链式调用方式依次组合清洗、分词、编码、截断、补齐、掩码等所有变换算子构建标准 LLM 预处理流水线def create_llm_dataset(data_path, vocab_path): # 1. 加载纯文本数据集单列为原始文本 dataset ds.TextFileDataset( dataset_filesdata_path, shuffleTrue, # 全局打乱数据 num_parallel_workers8 # 8线程并行处理加速变换 ) # 2. 第一步执行自定义文本清洗变换 dataset dataset.map( operationsclean_op, input_columns[text], output_columns[clean_text], num_parallel_workers8 ) # 过滤清洗后为空的数据行 dataset dataset.drop(columns[text]) dataset dataset.filter(predicatelambda x: x is not None, input_columns[clean_text]) # 3. 初始化分词器与词表映射器 tokenizer text.SentencePieceTokenizer(vocab_path) vocab text.Vocab.from_file(vocab_path, unknown_tokenunk) lookup_op text.Lookup(vocab, unk_idUNK_ID) # 4. 第二步分词变换文本转Token列表 dataset dataset.map( operationstokenizer, input_columns[clean_text], output_columns[tokens], num_parallel_workers8 ) # 5. 第三步Token转整数ID词表编码 dataset dataset.map( operationslookup_op, input_columns[tokens], output_columns[token_ids], num_parallel_workers8 ) # 6. 第四步序列截断超长文本截断至最大长度 slice_op T.Slice(start0, endMAX_SEQ_LEN) dataset dataset.map( operationsslice_op, input_columns[token_ids], output_columns[token_ids], num_parallel_workers8 ) # 7. 第五步序列补齐短文本填充至统一长度 pad_op T.PadSequence( pad_shape[MAX_SEQ_LEN], padding_valuePAD_ID, pad_endTrue ) dataset dataset.map( operationspad_op, input_columns[token_ids], output_columns[token_ids], num_parallel_workers8 ) # 8. 第六步生成注意力掩码Padding位置置0有效Token置1 def create_mask(ids): mask (ids ! PAD_ID).astype(ms.int32) return mask mask_op C.pyfunc(funccreate_mask, output_typesms.int32, output_shapes[MAX_SEQ_LEN]) dataset dataset.map( operationsmask_op, input_columns[token_ids], output_columns[attention_mask], num_parallel_workers8 ) # 9. 转换数据类型为模型所需张量格式 type_cast_op C.TypeCast(ms.int32) dataset dataset.map( operationstype_cast_op, input_columns[token_ids, attention_mask], num_parallel_workers8 ) # 10. 分批、预取数据完成流水线收尾 dataset dataset.batch(BATCH_SIZE, drop_remainderTrue) dataset dataset.prefetch(buffer_size16) # 预取16个Batch流水线加速 return dataset3.4 数据集调用与结果验证初始化数据集并遍历输出验证所有数据变换是否生效查看预处理后张量形状与内容if __name__ __main__: # 初始化完整预处理数据集 train_dataset create_llm_dataset(DATA_PATH, VOCAB_FILE) # 遍历数据集查看预处理结果 for batch_data in train_dataset.create_tuple_iterator(): token_ids, attn_mask batch_data print(f批次Token ID形状: {token_ids.shape}) print(f批次注意力掩码形状: {attn_mask.shape}) print(f首行Token ID: {token_ids[0][:10]}) print(f首行注意力掩码: {attn_mask[0][:10]}) print(- * 60) # 仅展示前2个批次避免输出过多 break四、多模态大模型数据变换扩展图文联合预处理对于文图生成、图文问答等多模态大模型需要同时处理文本 图像两类数据mindspore.dataset支持多列数据并行变换在文本处理基础上叠加图像变换算子构建多模态预处理流水线。以下为核心图像变换示例可与上文文本流水线组合使用# 图像基础变换算子解码、缩放、归一化、维度转换 image_transform [ C.Decode(), # 图像解码 C.Resize((224, 224)), # 统一图像尺寸 C.Normalize(mean[127.5, 127.5, 127.5], std[127.5, 127.5, 127.5]), # 归一化 C.HWC2CHW() # 维度转换 HWC - CHW模型标准格式 ] # 在多模态数据集中应用图像变换 # dataset dataset.map(operationsimage_transform, input_columns[image])多模态场景下文本变换与图像变换相互独立、并行执行多线程调度进一步提升整体处理效率。五、分布式训练场景下的数据变换适配大模型分布式训练是主流部署形态mindspore.dataset原生支持分布式分片保证多卡之间数据不重复、不遗漏只需在数据集加载阶段增加分片配置from mindspore.communication import init, get_rank, get_group_size # 初始化分布式通信 init() rank_id get_rank() # 当前卡编号 rank_size get_group_size() # 总卡数 # 加载数据集时添加分片参数 dataset ds.TextFileDataset( dataset_filesDATA_PATH, shuffleTrue, num_shardsrank_size, shard_idrank_id, num_parallel_workers8 )分布式模式下每张卡仅加载对应分片数据数据变换逻辑完全复用单机代码无需额外修改大幅降低分布式适配成本。六、性能调优与工程化最佳实践6.1 并行线程数调优num_parallel_workers是影响预处理速度的核心参数建议设置为CPU 核心数的 1~1.5 倍。鲲鹏多核服务器可根据物理核数量调高线程数充分利用 CPU 算力线程数过高会引发线程竞争反而降低效率。6.2 预取与数据下沉prefetch开启数据预取在模型计算时提前加载下一批数据消除数据等待时延训练时开启dataset_sink_modeTrue数据下沉将数据直接送入昇腾设备显存避免 CPU 与 NPU 频繁数据拷贝训练整体吞吐可提升 20%~40%。6.3 离线预处理缓存对于固定数据集建议执行离线预处理运行一次完整变换流程将编码后的 Token ID、掩码张量保存为二进制文件训练时直接加载跳过重复的分词、清洗逻辑节省训练启动耗时。6.4 算子组合顺序优化遵循先清洗、后分词、再编码、最后序列处理的顺序编排变换算子将计算密集型算子分词、解码放在前面轻量算子类型转换、重命名放在后面符合流水线调度最优逻辑。6.5 常见问题排查序列长度报错检查MAX_SEQ_LEN与模型输入维度是否匹配确保截断、补齐生效分词结果异常核对词表文件路径、分词算法与模型预训练词表保持一致分布式数据重复确认num_shards与集群总卡数一致分片 ID 配置正确。七、总结mindspore.dataset是昇思 MindSpore 大模型体系中不可或缺的数据处理引擎其内置的丰富数据变换算子覆盖了大语言模型、多模态大模型从原始数据到模型输入张量的全预处理链路。依托 C 后端加速、多线程并行、流水线预取、分布式分片四大核心能力彻底解决了传统 Python 预处理方案速度慢、资源占用高、分布式适配复杂的问题完美适配昇腾 AI 集群与鲲鹏服务器的国产软硬件生态。本文通过完整可运行代码实现了文本清洗、分词、编码、序列截断补齐、注意力掩码生成等 LLM 核心变换逻辑并扩展了多模态、分布式场景的适配方案。在实际工程落地中开发者可根据业务需求自由组合内置算子与自定义变换函数灵活搭建专属预处理流水线。