MindSpore大模型数据预处理：mindspore.dataset变换与性能优化实践

做了一段时间大模型微调和AI落地项目之后，我有一个很深的感受：模型结构选型固然重要，但真正决定项目能跑多远、效果稳不稳定的，往往是被很多人忽视的数据预处理环节。尤其是当你盯上昇思MindSpore这套框架时，mindspore.dataset这个看似“只是喂数据”的模块，实际上藏着整套数据处理管线的设计哲学。你数据清洗得好不好、batch怎么组、并行度怎么调、分布式下要不要切片，这些都会直接影响训练收敛速度和最终指标。

今天这篇东西，我就以MindSpore大模型场景为主线，把基于mindspore.dataset的数据变换与预处理方案完整拆开讲一遍。里面有最基础的数据集构建逻辑，也有文本、图像、多模态场景下我实际验证过的预处理流程，还会聊到性能优化和一批典型的坑。如果你是刚开始接触MindSpore，或者正在用它对大模型做微调，这篇应该能帮你少走不少弯路。

1. 数据处理这关，为什么在大模型场景里尤其要重视

很多人一上来就关注模型怎么改、学习率怎么设，觉得数据处理无非是“把文件读进来然后喂给模型”。这种想法在小数据、小模型时代勉强够用，但放到大模型的训练和微调场景里，基本是行不通的。

1.1 大模型对数据的“质”和“量”要求完全不同

大模型训练数据动辄几十GB甚至TB级，光是把这些数据完整读进内存就根本不现实。更重要的是，大模型对数据质量极其敏感。举例来说，你做指令微调，如果prompt和response之间的分隔符处理得不统一，模型就可能学到“乱接话”的坏习惯；你做文本续写预训练，如果文本被截断得不是地方，模型学会的就可能是“话说到一半强行结束”。

MindSpore里的mindspore.dataset就是为此设计的。它不是简单的文件读取器，而是一整套惰性执行的、可组合的数据处理管线。你可以把它想象成一条流水线：原始文件像原材料从一端放进去，中间经过解码、清洗、分词、截断、补齐、batch等一系列工位，最后从另一端出来的是已经可以直接喂给模型的标准化Tensor。这条流水线不会在定义时就执行，而是在迭代时才真正跑起来，好处是既能节省内存，又方便在训练过程中做各种动态增强和扰动。

1.2 mindspore.dataset 能解决什么，不能解决什么

先说能解决的：大规模数据的流式读取、多进程并行、随机shuffle、按step做batch、各种常见变换算子、分布式训练时的数据切分，还有基于缓存的数据加速。这些恰恰是你在做大模型训练时的高频需求。

不能解决或者说不太适合的：如果你的原始数据本身就是混乱到需要弹性解析的格式，比如某些日志里嵌套了异常字段，你多半还得先在数据集进入MindSpore之前做一次离线清洗。硬要把所有清洗逻辑塞进dataset管线里，调试起来会非常痛苦。我的习惯是“离线粗清洗 + 在线细变换”两层结构：离线把明显脏的数据、格式不统一的文件处理掉，在线用mindspore.dataset做标准化、增强和batch组织。

注意：mindspore.dataset里的很多算子支持Python函数作为回调，比如map操作可以传一个自定义函数进去。这给灵活性带来了很大提升，但如果你在回调里写了特别重的逻辑（比如逐样本调一次远程接口），整个训练Pipeline的性能会被拖垮，后面我会专门讲这个问题。

2. 环境准备与基础API速览，先把地基打好

在展开预处理方案之前，我建议你先确认自己的MindSpore环境是干净可用的。版本差异带来的API变化往往比你想的更折腾人，尤其是2.x之后，mindspore.dataset的接口细节有过不少调整。

2.1 安装MindSpore与核心依赖

MindSpore支持CPU、GPU、昇腾三种后端。做大模型微调，GPU或昇腾基本是刚需，CPU只能拿来跑跑小样例和语法验证。安装命令比较直接，官方源用pip就能装：

# CPU版本，用于验证代码逻辑 pip install mindspore==2.2.0 # CUDA 11.6环境，GPU版本 pip install mindspore==2.2.0-cuda116-cp39-cp39-linux_x86_64.whl

我建议你装完之后先跑一个最小用例确认后端注册成功：

import mindspore as ms from mindspore import Tensor import mindspore.dataset as ds ms.set_context(device_target="GPU") print(ms.__version__) print(Tensor([1, 2, 3]).sum())

能正常输出版本号和求和结果，说明环境基本没问题。如果你在VS Code里用MindSpore内核做开发，注意Python解释器一定要选对虚拟环境，否则你会发现明明装了mindspore却一直import失败。

2.2 mindspore.dataset 的核心API地图

初次接触mindspore.dataset的人，最容易被一堆Dataset类搞晕。其实梳理一下，核心就几类：

其中GeneratorDataset是绕不开的核心出入口。它允许你用一个Python生成器或自定义类来产生数据，这意味着你可以从任意数据源（内存数组、数据库游标、文件列表）构建数据集。它内部会做大量优化，但在你写回调函数时还是有很明显的注意事项，这个我放到后面问题章节细说。

一个最朴素的数据集构建长这样：

import numpy as np def generator_func(): for i in range(100): yield np.array([i, i + 1]), np.array([i * 2]) dataset = ds.GeneratorDataset(generator_func, column_names=["input", "label"]) dataset = dataset.batch(batch_size=8) for data in dataset.create_dict_iterator(): print(data["input"].shape, data["label"].shape)

这段代码就完成了从Python生成器到可迭代Tensor数据集的全过程。接下来我们就要在这个基础上，叠加真正的数据变换。

3. 文本数据预处理全流程：从原始文本到模型输入Tensor

大模型场景里，文本预处理是最核心也是最容易被忽视的一块。很多微调脚本里就写一个tokenizer调用完事，但真正落地时，你还要处理截断、补齐、注意力掩码、标签错位等一系列问题。

3.1 从原始文件构建可迭代的文本数据集

假设你手里有一批JSONL格式的指令微调数据，每一行是{"instruction": "请解释...", "output": "..."}。用MindSpore读取的基本思路是：先用Python把文件解析成一个生成器，再交给GeneratorDataset。

import json def load_jsonl(file_path): with open(file_path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue item = json.loads(line) yield item["instruction"], item["output"] dataset = ds.GeneratorDataset( load_jsonl("train.jsonl"), column_names=["instruction", "output"] )

这里有个很容易犯的错：GeneratorDataset的source参数如果传的是一个生成器对象，那么它在多次迭代时会出问题，因为生成器是一次性的。正确做法是传一个可调用的函数（像上面这样），或者传一个实现了__iter__的类实例。这样每次epoch启动时，MindSpore都能重新创建迭代器，数据才能反复使用。

3.2 分词与tokenize的落地写法

MindSpore本身没有像HuggingFace Transformers那样的完整分词器库，但它提供了足够灵活的接口让你集成外部分词器。常见做法是直接用transformers的tokenizer，然后在map操作里包装成一个Python函数。

from transformers import AutoTokenizer tokenizer = AutoTokenizer.from_pretrained("/path/to/tokenizer") def tokenize(instruction, output): text = "指令：" + instruction + "\n回答：" + output tokenized = tokenizer( text, truncation=True, max_length=512, padding=False, return_tensors=None ) input_ids = tokenized["input_ids"] attention_mask = tokenized["attention_mask"] labels = input_ids.copy() return input_ids, attention_mask, labels dataset = dataset.map(tokenize, input_columns=["instruction", "output"], output_columns=["input_ids", "attention_mask", "labels"])

这里有几个关键点：

• map的input_columns要跟数据集已有的列名完全对应，输出列名可以通过output_columns指定。如果我不指定output_columns，MindSpore会用函数返回值自动推断，但这种情况下列名往往是output_1、output_2这种，后面引用起来很别扭，最好显式命名。
• 每条样本长度不一，所以这时的input_ids还是一个变长列表。直接进batch会报错，必须补齐或做动态padding，这是下一步的重点。
• tokenizer里不要急着做padding。先把padding=False留到batch阶段再统一处理，否则每条样本都会pad到相同长度，显存和算力都浪费。

3.3 batch阶段如何优雅地做padding

这是文本预处理里最容易踩坑的地方。PyTorch里你可以用DataCollatorWithPadding自动搞定，但MindSpore的batch算子本身不会自动padding，它默认要求batch内所有样本形状一致。

我的做法是自定义一个batch函数，在组batch时动态计算当前batch的最大长度，然后把所有样本补齐到这个长度，这样batch内部的冗余最少。

def batch_padding(batch_data): inp_ids, attn_mask, labels = zip(*batch_data) max_len = max(len(ids) for ids in inp_ids) padded_ids = [] padded_mask = [] padded_labels = [] for ids, mask, lab in zip(inp_ids, attn_mask, labels): pad_len = max_len - len(ids) padded_ids.append(ids + [tokenizer.pad_token_id] * pad_len) padded_mask.append(mask + [0] * pad_len) padded_labels.append(lab + [-100] * pad_len) return np.array(padded_ids, dtype=np.int32), \ np.array(padded_mask, dtype=np.int32), \ np.array(padded_labels, dtype=np.int32) dataset = dataset.batch(batch_size=16, per_batch_map=batch_padding)

labels里pad的部分用-100填充，这是业界通用做法，计算loss时忽略这些位置。注意per_batch_map是MindSpore batch算子专门用来支持自定义batch逻辑的一个参数，很多刚从PyTorch转过来的人不知道这个，误以为只能老老实实录完所有样本再手动拼。

3.4 缓存、shuffle与repeat的顺序问题

文本数据集处理完tokenize之后，shuffle和repeat的顺序是有讲究的。正确顺序是：先shuffle再repeat。如果你先repeat再shuffle，那么shuffle的单位变成了一整轮epoch的拼接，不仅内存压力大，而且数据随机性并没有变得更好。还有一个更实际的问题：shuffle是在map之后做，还是在map之前做？

我的经验是：如果原始文本很短，map（也就是tokenize）开销很大，提前shuffle会更好。因为先shuffle再map，每条数据被随机打乱后进入tokenize，每个worker处理的样本分布更随机；如果你先map再shuffle，内存里缓存的全是tokenize后的Tensor，占的空间更大，shuffle时的内存压力也更明显。当然，如果你的原始数据文件本身就很大，优先考虑的是减少IO次数，那就要结合cache来看了。

一个典型的完整pipeline如下：

dataset = ds.GeneratorDataset(load_jsonl("train.jsonl"), column_names=["instruction", "output"]) dataset = dataset.shuffle(buffer_size=10000) dataset = dataset.map(tokenize, input_columns=["instruction", "output"], output_columns=["input_ids", "attention_mask", "labels"]) dataset = dataset.batch(batch_size=16, per_batch_map=batch_padding) dataset = dataset.repeat(epochs)

很多人会把repeat放在最前面，或者放在map之前，我建议你统一放在batch后面。这样做的好处是：每个epoch的数据经过相同的shuffle + tokenize + batch流程，语义清晰，调试起来也不容易出幺蛾子。

4. 图像与多模态数据的预处理：别把简单事情复杂化

大模型不只有文本，多模态场景越来越常见。MindSpore的mindspore.dataset对图像数据也有完整的支持，从读取到增强再到归一化，都有现成算子。不过很多人在多模态数据对齐上栽过跟头，这里一并讲透。

4.1 图像数据集的构建与基础增强

假设你有一批图像分类任务的数据，目录结构是标准的train/class1/*.jpg，直接用ImageFolderDataset最省事：

import mindspore.dataset.vision as vision image_folder = ds.ImageFolderDataset("/data/train", class_indexing={"cat": 0, "dog": 1}) image_folder = image_folder.map(operations=[ vision.Decode(), vision.Resize((224, 224)), vision.RandomHorizontalFlip(prob=0.5), vision.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]), vision.HWC2CHW() ], input_columns=["image"])

这里的核心逻辑是把一组变换算子传给map的operations参数。MindSpore会按顺序依次应用，而且这些算子在底层做了C++层面的优化，比你自己写Python循环再逐张处理要快很多。

我在实际项目里有个心得：Decode算子一定要放在所有图像变换之前。很多人喜欢先把图读了再做别的，但如果你的原始数据是JPEG，让MindSpore的Decode在流水线里做解码，就能利用底层的解码优化。不要觉得都是读图，顺序无所谓，解码位置放得不对，性能差距能到2到3倍。

4.2 多模态样本对齐的正确姿势

真正搞多模态大模型时，一条样本可能包含图像路径和文本描述两部分。这时最合适的入口还是GeneratorDataset：

def load_multimodal(meta_list): for item in meta_list: img = cv2.imread(item["image_path"]) img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) yield img, item["text"], item["label"] dataset = ds.GeneratorDataset( lambda: load_multimodal(meta_list), column_names=["image", "text", "label"] )

然后对image列和text列分别做不同的map变换。MindSpore的map支持按列名分别传入不同算子，互不干扰，这点和PyTorch里手动写collate_fn的体验完全不同，写起来清晰很多。

这里有个特别重要的细节：在多模态场景下，图像解码和文本tokenize在同一个pipeline里，但两者耗时差异巨大。图像解码往往比文本tokenize慢一个数量级，如果你只开了默认的并行worker数，整个流水线会被图像处理拖住。后面性能章节我会说怎么针对不同列分别调整并行度。

4.3 用tfrecord还是直接用原始文件，这是个方案选择题

MindSpore原生支持TFRecordDataset，性能也不错。如果你的数据规模到了几十GB，而且有反复读取的需求，我建议把预处理后的结果落成TFRecord或MindRecord格式，这样后续训练时IO开销会小很多。

但如果你只是小规模微调，或者数据还在不断迭代清洗中，完全没必要多此一举。直接用GeneratorDataset加map就够，落盘反而增加维护成本。说到底，数据格式选型取决于数据量和迭代频率，不是越重越好。

5. 数据变换的性能优化：别让预处理成为训练瓶颈

大模型训练时，GPU的计算速度非常快，如果数据喂得不够快，GPU就会在那里干等。硬件那么贵，等一秒都是钱。所以我专门把性能优化拿出来单独讲，这部分恰恰是最容易被忽略又最影响训练效率的地方。

5.1 num_parallel_workers到底设多少合适

map操作里有个num_parallel_workers参数，默认值在不同版本里不一样，而且它只管map内部开几个worker。很多人觉得这个值设得越大越好，其实不然。

每个worker都会独立占用CPU资源，如果你的机器CPU核数有限，开太多worker反而会导致频繁的上下文切换，性能不升反降。我自己的经验是：对于单机训练，文本tokenize这类操作，worker数设为CPU物理核数的一半到三分之二比较合适；图像解码这种偏IO的操作，可以稍微再多开一点，因为worker经常在等IO而CPU利用率不高。

一个粗略的设置参考：

还有一点值得注意：如果你在map的回调里用了HuggingFace的tokenizer，它的底层可能自己也会开线程。这时MindSpore的worker和tokenizer内部线程叠加起来，CPU资源争抢会非常明显。我的处理办法是，在tokenize函数外层加一个线程锁，或者干脆在构造tokenizer时设置local_files_only=True并且避免每次都重新加载。

5.2 用DatasetCache缓存重复处理的结果

在微调场景中，数据经过shuffle后往往会被多个epoch重复读取。如果每次epoch都重新做一遍tokenize和图像解码，CPU和磁盘IO都在做大量无用功。MindSpore提供了DatasetCache，可以把第一次迭代时处理过的数据缓存到共享内存或本地磁盘，之后的epoch直接复用。

cache = ds.DatasetCache(session_size=10, spilling_size=20*1024*1024*1024) dataset = dataset.map(tokenize, input_columns=["instruction", "output"], output_columns=["input_ids", "attention_mask", "labels"], cache=cache)

这个方案在做大模型微调时收益极其明显。我实测过一个文本微调任务，加缓存前每个epoch的数据准备时间是8分钟左右，加缓存后第二次epoch开始降到几十秒。不过要注意：如果你的数据增强算子带有随机性（比如随机裁剪、随机mask），那这些随机操作不能放到缓存前面，否则缓存出来的永远是同一份增强结果，模型的泛化能力会受影响。

5.3 分布式训练中的数据切分与打乱

多卡训练大模型时，每张卡应该拿到不同的数据切片，否则相当于所有卡在重复学习同一批东西，batch size等于没变。MindSpore的分布式数据并行处理通过num_shards和shard_id实现：

dataset = ds.GeneratorDataset(load_jsonl("train.jsonl"), column_names=["instruction", "output"], num_shards=rank_size, shard_id=rank_id) dataset = dataset.shuffle(buffer_size=10000)

注意shuffle必须放在sharding之后，否则每张卡拿到的数据切片是固定的，不同epoch之间数据分布差异会非常大。另外在数据集较大的情况下，sharding后再shuffle，每张卡只需要维护自己那份数据的buffer，内存占用会小很多。

5.4 用Profiler定位真正的瓶颈在哪

性能问题不能靠猜。MindSpore提供了Profiler工具，可以统计每个数据集算子花费的时间和吞吐量。我之前遇到过一个很隐蔽的问题：数据管线速度正常，但每隔一段时间GPU利用率掉到0。用Profiler一查，发现是某个map操作里回调函数偶尔会触发一次磁盘IO（因为有缓存未命中），导致整个管线在那一瞬间卡住。

排查这类问题的手段其实很简单：先逐步注释掉map里的操作，看哪一步耗时占比最大。然后针对那一步去优化，比如换成更轻量的算法、增加缓存或者提高并行度。先窄化范围再动手优化，才能避免做无用功。

6. 常见问题与排查技巧实录

最后这部分是我反复帮人调MindSpore数据管线时总结出来的高频问题，写出来给大家避坑。每一个我都亲自踩过或者帮人排过，实战价值非常高。

6.1 GeneratorDataset迭代一次之后就没数据了

这是最经典的问题。很多人把生成器对象直接传给GeneratorDataset，第一轮epoch跑得很正常，到第二轮就报StopIteration或者数据为空。原因就是生成器一次性消费掉了。解决方案在前面提过：传函数而不是生成器对象，或者自定义类实现__iter__方法。

# 错误示例 gen = load_jsonl("train.jsonl") dataset = ds.GeneratorDataset(gen, column_names=["instruction", "output"]) # 第二轮迭代会出问题 # 正确示例 dataset = ds.GeneratorDataset(load_jsonl, column_names=["instruction", "output"]) # 传函数引用，每次迭代都会重新调用

如果你非要传生成器对象，那就只能自己写一个包装类，在__iter__里重新创建生成器。但说实话，直接传函数引用是最简洁的。

6.2 batch时报错说维度不一致

这个问题在文本场景里几乎必现。原因是batch算子默认要求batch内所有样本的shape完全一致，而文本tokenize后的长度天然就是变长的。解决方案就是前面写的per_batch_map动态padding，没有别的捷径。

还有个隐蔽的情况：你用了per_batch_map，但函数返回值是Python list，而不是numpy数组。MindSpore在某些版本里对Python list转Tensor的自动推断并不友好，会报奇怪的类型错误。建议在batch函数末尾统一转成numpy数组，并明确指定dtype。

6.3 map回调函数里的随机数在每个epoch都一样

如果你的map里用了Python的random模块，比如随机mask一段文本，你会发现不同epoch之间采到的mask位置居然一模一样，等于没做随机增强。原因很简单：MindSpore的多worker进程环境下，每个worker的随机种子被固定了。解决办法是在回调函数里不要用全局random，而是用numpy.random并基于当前epoch/样本索引生成随机因子，或者干脆把随机性放到dataset外部。

def tokenize_with_random_mask(text): np.random.seed(hash(text) % (2**32)) # 基于样本内容生成随机种子 mask_start = np.random.randint(0, len(text.split())) ...

这个方法不算完美，但至少保证同一条样本在不同epoch里得到的随机结果不同，实测有效。

6.4 自定义Python函数处理太慢

这是最让人头疼的问题。MindSpore的map允许Python回调，但Python函数跑得再快也有上限，尤其是对比框架内置的C++算子。如果发现自定义函数成了瓶颈，有两条路：

一是能换用内置算子的就换内置算子，比如分词后的截断补齐，用ds.pad或者batch阶段的per_batch_map来处理，不要写Python循环。

二是如果必须用Python函数，就尽量减少函数内部开销。比如避免在函数里重复加载模型/分词器，避免频繁的dict创建和复制，把重计算提前。

我见过最夸张的例子，有人在map回调里每次创建一个新的tokenizer实例，直接导致训练速度慢了20倍。把tokenizer放到函数外部复用之后，速度立刻回归正常。

6.5 数据管线的内存占用不断上升

训练过程中内存持续增长，通常不是因为数据本身太大，而是shuffle的buffer_size设置得不合理，或者缓存没有被正确清理。如果shuffle的buffer_size设成几百万条，每条样本都是tokenize后的长Tensor，那么内存占用很容易把机器压垮。

我的建议是：先把buffer_size调到能保证随机性的最小值，然后逐级往上测试，直到内存占用达到训练可接受上限。一般来说，文本微调任务里buffer_size几千到一万就足够，没必要设成整个数据集大小。

6.6 多卡训练时每张卡的数据一模一样

这个问题通常是因为忘了设置num_shards和shard_id，或者shard_id在所有卡上设置成了相同值。另外一个容易忽略的点是：在分布式模式下，如果shuffle的buffer_size小于num_shards，shuffle反而会破坏sharding的均匀性，导致不同卡之间数据分配不均。建议shuffle的buffer_size至少是num_shards的几倍以上。

7. 一段可直接参考的完整文本预处理管线

光讲散点不做汇总，阅读体验还是差了点。我在这里贴一段我实际用过的、比较完整的文本微调数据管线，把前面说的东西串起来，你可以直接改路径和配置后跑。

import os import json import numpy as np import mindspore as ms import mindspore.dataset as ds from transformers import AutoTokenizer TOKENIZER_PATH = "/path/to/tokenizer" DATA_PATH = "/path/to/train.jsonl" MAX_LENGTH = 512 BATCH_SIZE = 16 EPOCHS = 3 RANK_SIZE = 1 RANK_ID = 0 tokenizer = AutoTokenizer.from_pretrained(TOKENIZER_PATH) def load_jsonl(file_path): def generator(): with open(file_path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue item = json.loads(line) yield item["instruction"], item["output"] return generator def tokenize(instruction, output): text = "指令：" + instruction + "\n回答：" + output tokenized = tokenizer( text, truncation=True, max_length=MAX_LENGTH, padding=False, return_tensors=None ) input_ids = tokenized["input_ids"] attention_mask = tokenized["attention_mask"] labels = input_ids.copy() return input_ids, attention_mask, labels def batch_padding(batch_data): inp_ids, attn_mask, labels = zip(*batch_data) max_len = max(len(ids) for ids in inp_ids) padded_ids = [] padded_mask = [] padded_labels = [] for ids, mask, lab in zip(inp_ids, attn_mask, labels): pad_len = max_len - len(ids) padded_ids.append(ids + [tokenizer.pad_token_id] * pad_len) padded_mask.append(mask + [0] * pad_len) padded_labels.append(lab + [-100] * pad_len) return (np.array(padded_ids, dtype=np.int32), np.array(padded_mask, dtype=np.int32), np.array(padded_labels, dtype=np.int32)) def build_train_dataset(): dataset = ds.GeneratorDataset( load_jsonl(DATA_PATH), column_names=["instruction", "output"], num_shards=RANK_SIZE, shard_id=RANK_ID ) dataset = dataset.shuffle(buffer_size=10000) dataset = dataset.map( tokenize, input_columns=["instruction", "output"], output_columns=["input_ids", "attention_mask", "labels"], num_parallel_workers=8 ) dataset = dataset.batch(batch_size=BATCH_SIZE, per_batch_map=batch_padding) dataset = dataset.repeat(EPOCHS) return dataset if __name__ == "__main__": ms.set_context(device_target="GPU") train_data = build_train_dataset() for step, data in enumerate(train_data.create_dict_iterator()): print(f"step {step}: {data['input_ids'].shape}")

这段代码覆盖了本节前面90%的注意点，唯一没包含的是DatasetCache的用法（因为分布式场景下缓存共享逻辑要更小心）。你如果要加缓存，参考我上一节给的参数补上就行。

8. 最后再分享一点我的实操心得

数据预处理这事，看起来不如模型结构那么“高大上”，但在大模型项目里它就是那个最决定下限的环节。数据管线写好了，模型训练顺风顺水；管线写拉胯了，再好的模型也救不回来。

我个人在MindSpore上做数据处理的习惯可以总结成几句话：先想清楚数据长什么样、要变成什么样，再动手写管线；凡是能在离线阶段做掉的脏活累活绝不放在线pipeline里；凡是能用框架内置算子解决的绝不用Python硬写；凡是涉及随机增强的都要确认每个epoch真的在“随机”。你按这几个原则去搭管线，大概率不会出大问题。

像是vscode里跑MindSpore内核调试之类的场景，我也遇到过不少次，核心思想都一样：先把数据集单独拿出来迭代一遍，确定数据没问题，再挂到模型上训练。这样定位问题会快很多，也不会被训练日志里的各种信息干扰。

如果你正在做的项目刚好也是大模型微调或全参数训练，希望这篇文章里关于mindspore.dataset的整套方案，能帮你把数据这部分省下一点折腾的时间。毕竟大模型训练这件事，时间才是最贵的资源。