编写transformers的自定义pytorch训练循环（Dataset和DataLoader解析和实例代码）_transformers dataloader

一、Dataset和DataLoader加载数据集

1.torch.utils.data

torch.utils.data主要包括以下三个类：

1. class torch.utils.data.Dataset
   其他的数据集类必须是torch.utils.data.Dataset的子类,比如说torchvision.ImageFolder.
2. class torch.utils.data.sampler.Sampler(data_source)
   参数: data_source (Dataset) – dataset to sample from
   作用: 创建一个采样器, class torch.utils.data.sampler.Sampler是所有的Sampler的基类, 其中,iter(self)函数来获取一个迭代器,对数据集中元素的索引进行迭代,len(self)方法返回迭代器中包含元素的长度.
3. class torch.utils.data.DataLoader

2. 加载数据流程

pytorch中加载数据的顺序是：

1. 加载数据，提取出feature和label，并转换成tensor
2. 创建一个dataset对象
3. 创建一个dataloader对象，dataloader类的作用就是实现数据以什么方式输入到什么网络中
4. 循环dataloader对象，将data,label拿到模型中去训练
   代码一般是这么写的：

# 定义学习集 DataLoader
train_data = torch.utils.data.DataLoader(各种设置...) 
# 将数据喂入神经网络进行训练
for i, (input, target) in enumerate(train_data): 
    循环代码行......
1
2
3
4
5

3. Dataset

Dataset是我们用的数据集的库，是Pytorch中所有数据集加载类中应该继承的父类。其中父类中的两个私有成员函数必须被重载，否则将会触发错误提示。其中__len__应该返回数据集的大小，而__getitem__应该编写支持数据集索引的函数

class Dataset(object):
    def __init__(self):
        ...       
    def __getitem__(self, index):
        return ...    
    def __len__(self):
        return ...
1
2
3
4
5
6
7
'
运行
运行

上面三个方法是最基本的，其中__getitem__是最主要的方法，它规定了如何读取数据。其主要作用是能让该类可以像list一样通过索引值对数据进行访问。

class FirstDataset(data.Dataset):#需要继承data.Dataset
    def __init__(self):
        # 初始化，定义你用于训练的数据集(文件路径或文件名列表)，以什么比例进行sample（多个数据集的情况），每个epoch训练样本的数目，预处理方法等等
        #也就是在这个模块里，我们所做的工作就是初始化该类的一些基本参数。
        pass
    def __getitem__(self, index):
         #从文件中读取一个数据（例如，使用numpy.fromfile，PIL.Image.open）。
         #预处理数据（例如torchvision.Transform）。
         #返回数据对（例如图像和标签）。
         #这里需要注意的是，第一步：read one data，是一个data
        pass
    def __len__(self):
        # 定义为数据集的总大小。
1
2
3
4
5
6
7
8
9
10
11
12
13

图片加载的dataset可以参考帖子：《带你详细了解并使用Dataset以及DataLoader》
人民币二分类参考：《pytorch - 数据读取机制中的Dataloader与Dataset》

4. dataloader类及其参数

dataloader类调用torch.utils.Data.DataLoader，实际过程中数据集往往很大，通过DataLoader加载数据集使用mini-batch的时候可以使用多线程并行处理，这样可以加快我们准备数据集的速度。Datasets就是构建这个工具函数的实例参数之一。一般可以这么写：

train_loader = DataLoader(dataset=train_data, batch_size=6, shuffle=True ，num_workers=4)
test_loader = DataLoader(dataset=test_data, batch_size=6, shuffle=False，num_workers=4)
1
2

下面看看dataloader代码：

class DataLoader(object):
    def __init__(self, dataset, batch_size=1, shuffle=False, sampler=None,
                 batch_sampler=None, num_workers=0, collate_fn=default_collate,
                 pin_memory=False, drop_last=False, timeout=0,
                 worker_init_fn=None)
    self.dataset = dataset
        self.batch_size = batch_size
        self.num_workers = num_workers
        self.collate_fn = collate_fn
        self.pin_memory = pin_memory
        self.drop_last = drop_last
        self.timeout = timeout
        self.worker_init_fn = worker_init_fn

1
2
3
4
5
6
7
8
9
10
11
12
13
14

• dataset:Dataset类，PyTorch已有的数据读取接口，决定数据从哪里读取及如何读取；
• batch_size：批大小；默认1
• num_works:是否多进程读取数据；默认0使用主进程来导入数据。大于0则多进程导入数据，加快数据导入速度
• shuffle：每个epoch是否乱序；默认False。输入数据的顺序打乱，是为了使数据更有独立性，但如果数据是有序列特征的，就不要设置成True了。一般shuffle训练集即可。
• drop_last:当样本数不能被batchsize整除时，是否舍弃最后一批数据；
• collate_fn:将得到的数据整理成一个batch。默认设置是False。如果设置成True，系统会在返回前会将张量数据（Tensors）复制到CUDA内存中。
• batch_sampler，批量采样，和batch_size、shuffle等参数是互斥的，一般采用默认None。batch_sampler，但每次返回的是一批数据的索引（注意：不是数据），应该是每次输入网络的数据是随机采样模式，这样能使数据更具有独立性质。所以，它和一捆一捆按顺序输入，数据洗牌，数据采样，等模式是不兼容的。
• sampler，默认False。根据定义的策略从数据集中采样输入。如果定义采样规则，则洗牌（shuffle）设置必须为False。
• pin_memory，内存寄存，默认为False。在数据返回前，是否将数据复制到CUDA内存中。
• timeout，是用来设置数据读取的超时时间的，但超过这个时间还没读取到数据的话就会报错。
• worker_init_fn（数据类型 callable），子进程导入模式，默认为Noun。在数据导入前和步长结束后，根据工作子进程的ID逐个按顺序导入数据。

想用随机抽取的模式加载输入，可以设置 sampler 或 batch_sampler。如何定义抽样规则，可以看sampler.py脚本，或者这篇帖子：《一文弄懂Pytorch的DataLoader, DataSet, Sampler之间的关系》

5. dataloader内部函数

5.1 __next__函数

DataLoader__next__函数用for循环来遍历数据进行读取。

def __next__(self): 
        if self.num_workers == 0:   
            indices = next(self.sample_iter)  
            batch = self.collate_fn([self.dataset[i] for i in indices]) # this line 
            if self.pin_memory: 
                batch = _utils.pin_memory.pin_memory_batch(batch) 
            return batch
1
2
3
4
5
6
7
'
运行
运行

仔细看可以发现，前面还有一个self.collate_fn方法，这个是干嘛用的呢?在介绍前我们需要知道每个参数的意义：

• indices: 表示每一个iteration，sampler返回的indices，即一个batch size大小的索引列表
• self.dataset[i]: 前面已经介绍了，这里就是对第i个数据进行读取操作，一般来说self.dataset[i]=(img, label)

看到这不难猜出collate_fn的作用就是将一个batch的数据进行合并操作。默认的collate_fn是将img和label分别合并成imgs和labels，所以如果你的__getitem__方法只是返回 img, label,那么你可以使用默认的collate_fn方法，但是如果你每次读取的数据有img, box, label等等，那么你就需要自定义collate_fn来将对应的数据合并成一个batch数据，这样方便后续的训练步骤。

5.2 DataLoaderIter函数

def __setattr__(self, attr, val):
        if self.__initialized and attr in ('batch_size', 'sampler', 'drop_last'):
            raise ValueError('{} attribute should not be set after {} is '
                             'initialized'.format(attr, self.__class__.__name__))
        super(DataLoader, self).__setattr__(attr, val)

def __iter__(self):
        return _DataLoaderIter(self)

def __len__(self):
        return len(self.batch_sampler)
1
2
3
4
5
6
7
8
9
10
11
'
运行
运行

当代码运行到要从torch.utils.data.DataLoader类生成的对象中取数据的时候，比如：

train_data=torch.utils.data.DataLoader(...)
for i, (input, target) in enumerate(train_data):
1
2

就会调用DataLoader类的__iter__方法：return DataLoaderIter(self)，此时牵扯到DataLoaderIter类：

def __iter__(self)：
	 if self.num_workers == 0:
            return _SingleProcessDataLoaderIter(self)
	 else:
            self.check_worker_number_rationality()
            return _MultiProcessingDataLoaderIter(self)
1
2
3
4
5
6

• SingleProcessDataLoaderIter：单线程数据迭代，采用普通方式来读取数据
• MultiProcessingDataLoaderIter：多进程数据迭代，采用队列的方式来读取。

MultiProcessingDataLoaderIter继承的是BaseDataLoaderIter,开始初始化，然后Dataloader进行初始化，然后进入 next __（）方法 随机生成索引，进而生成batch，最后调用 _get_data() 方法得到data。idx, data = self._get_data()， data = self.data_queue.get(timeout=timeout)

---

总结一下：

1. 调用了dataloader 的__iter__() 方法, 产生了一个DataLoaderIter
2. 反复调用DataLoaderIter 的__next__()来得到batch, 具体操作就是, 多次调用dataset的__getitem__()方法 (如果num_worker>0就多线程调用), 然后用collate_fn来把它们打包成batch. 中间还会涉及到shuffle , 以及sample 的方法等,
3. 当数据读完后, next()抛出一个StopIteration异常, for循环结束, dataloader 失效.

DataLoaderIter的源码及详细解读参考：《PyTorch源码解读之torch.utils.data.DataLoader》

6. dataloader循环

ataloader本质上是一个可迭代对象，但是dataloader不能像列表那样用索引的形式去访问，而是使用迭代遍历的方式。

for i in dataLoader:
	print(i.keys())
1
2

也可以使用enumerate(dataloader)的形式访问。
在计算i的类型时，发现其为一个字典，打印这个字典的关键字可得到

for i in dataLoader:
	print(i.keys())
1
2

dict_keys(['text', 'audio', 'vision', 'labels'])
1

同理，计算 **i[‘text’]**发现其为一个张量，打印该张量信息

print(i['text'].shape)  #64*39*768
1

此时的64恰好就是我们设置的batchsize，并且最后一个i值的text的shape为2439768，即24个数据

二、代码示例

1. transformer单句文本分类（HF教程）

1.1使用Trainer训练

GLUE榜单包含了9个句子级别的分类任务，分别是：

• CoLA (Corpus of Linguistic Acceptability) 鉴别一个句子是否语法正确.
• MNLI (Multi-Genre Natural Language Inference) 给定一个假设，判断另一个句子与该假设的关系：entails, contradicts 或者 unrelated。
• MRPC (Microsoft Research Paraphrase Corpus) 判断两个句子是否互为paraphrases.
• QNLI (Question-answering Natural Language Inference) 判断第2句是否包含第1句问题的答案。
• QQP (Quora Question Pairs2) 判断两个问句是否语义相同。
• RTE (Recognizing Textual Entailment)判断一个句子是否与假设成entail关系。
• SST-2 (Stanford Sentiment Treebank) 判断一个句子的情感正负向.
• STS-B (Semantic Textual Similarity Benchmark) 判断两个句子的相似性（分数为1-5分）。
• WNLI (Winograd Natural Language Inference) Determine if a sentence with an anonymous pronoun and a sentence with this pronoun replaced are entailed or not.

加载数据集

from datasets import load_dataset
raw_datasets = load_dataset("glue","sst2")
1
2

预处理数据

from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained("bert-base-cased")

def tokenize_function(examples):
    return tokenizer(examples["sentence"], padding="max_length", truncation=True)

tokenized_datasets = raw_datasets.map(tokenize_function, batched=True)

small_train_dataset = tokenized_datasets["train"].shuffle(seed=42).select(range(1000))
small_eval_dataset = tokenized_datasets["test"].shuffle(seed=42).select(range(1000))
full_train_dataset = tokenized_datasets["train"]
full_eval_dataset = tokenized_datasets["test"]
1
2
3
4
5
6
7
8
9
10
11
12

定义评估函数

import numpy as np
from datasets import load_metric

metric = load_metric("glue","sst2")#改成"accuracy"效果一样吗？

def compute_metrics(eval_pred):
    logits, labels = eval_pred
    predictions = np.argmax(logits, axis=-1)
    return metric.compute(predictions=predictions, references=labels)
1
2
3
4
5
6
7
8
9

加载模型

from transformers import AutoModelForSequenceClassification
model = AutoModelForSequenceClassification.from_pretrained("bert-base-cased", num_labels=2)
1
2

配置 Trainer参数：

from transformers import TrainingArguments，Trainer

args = TrainingArguments(
    "ft-sst2",                          # 输出路径，存放检查点和其他输出文件
    evaluation_strategy="epoch",        # 定义每轮结束后进行评价
    learning_rate=2e-5,                 # 定义初始学习率
    per_device_train_batch_size=16,     # 定义训练批次大小
    per_device_eval_batch_size=16,      # 定义测试批次大小
    num_train_epochs=2,                 # 定义训练轮数
)

trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=small_train_dataset,
    eval_dataset=small_eval_dataset,
    compute_metrics=compute_metrics,
)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

开始训练：

trainer.train()
1

训练完毕后，执行以下代码，得到模型在验证集上的效果：

trainer.evaluate()
1

{'epoch': 2,
 'eval_loss': 0.9351930022239685，
 'eval_accuracy'': 0.7350917431192661
 }
1
2
3
4

1.2 使用 PyTorch进行训练

重新启动笔记本以释放一些内存，或执行以下代码：

del model
del pytorch_model
del trainer
torch.cuda.empty_cache()
1
2
3
4

首先，我们需要定义数据加载器，我们将使用它来迭代批次。 在这样做之前，我们只需要对我们的 tokenized_datasets 应用一些后处理：

1. 删除与模型不期望的值相对应的列（此处为“text”列）
2. 将列“label”重命名为“labels”（因为模型期望参数被命名为标签）
3. 设置数据集的格式，以便它们返回 PyTorch 张量而不是列表。

tokenized_datasets 对每个步骤处理如下：

tokenized_datasets = tokenized_datasets.remove_columns(["sentence","idx"])#删除多余的“sebtence”列和“idx”列,否则会报错forward() got an unexpected keyword argument 'idx'
tokenized_datasets = tokenized_datasets.rename_column("label", "labels")#列“label”重命名为“labels”，否则报错forward() got an unexpected keyword argument 'label'
tokenized_datasets.set_format("torch")#返回 PyTorch 张量，否则报错'list' object has no attribute 'size'
1
2
3

二三步也可以合并：

columns = ['input_ids', 'token_type_ids', 'attention_mask', 'labels']
tokenized_datasets.set_format(type='torch', columns=columns)
1
2

切出一部分数据集

small_train_dataset = tokenized_datasets["train"].shuffle(seed=42).select(range(1000))
small_eval_dataset = tokenized_datasets["test"].shuffle(seed=42).select(range(1000))
1
2

定义dataloaders：

from torch.utils.data import DataLoader
train_dataloader = DataLoader(small_train_dataset, shuffle=True, batch_size=8)
eval_dataloader = DataLoader(small_eval_dataset, batch_size=8)
1
2
3

定义模型：

from transformers import AutoModelForSequenceClassification
model = AutoModelForSequenceClassification.from_pretrained("bert-base-cased", num_labels=2)
1
2

定义优化器optimizer 和学习率调度器scheduler：

from transformers import AdamW
optimizer = AdamW(model.parameters(), lr=5e-5)

#默认使用的学习率调度器只是线性衰减从最大值（此处为 5e-5）到 0：
from transformers import get_scheduler
num_epochs = 3
num_training_steps = num_epochs * len(train_dataloader)
lr_scheduler = get_scheduler(
    "linear",
    optimizer=optimizer,
    num_warmup_steps=0,
    num_training_steps=num_training_steps
)
1
2
3
4
5
6
7
8
9
10
11
12
13

使用GPU进行训练：

import torch
device = torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu")
model.to(device)
1
2
3

使用 tqdm 库在训练步骤数上添加了一个进度条，并定义训练循环：

from tqdm.auto import tqdm
progress_bar = tqdm(range(num_training_steps))

model.train()#设置train状态，启用 Batch Normalization 和 Dropout。
for epoch in range(num_epochs):
    for batch in train_dataloader:
        batch = {k: v.to(device) for k, v in batch.items()}
        outputs = model(**batch)
        loss = outputs.loss
        loss.backward()

        optimizer.step()
        lr_scheduler.step()
        optimizer.zero_grad()
        progress_bar.update(1)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

编写评估循环，在循环完成时计算最终结果之前累积每个批次的预测：

metric= load_metric("accuracy")
model.eval()
for batch in eval_dataloader:
    batch = {k: v.to(device) for k, v in batch.items()}
    with torch.no_grad():
        outputs = model(**batch)

    logits = outputs.logits
    predictions = torch.argmax(logits, dim=-1)
    metric.add_batch(predictions=predictions, references=batch["labels"])

metric.compute()
1
2
3
4
5
6
7
8
9
10
11
12

1.3 句子对文本分类（rte）：

dataset = load_dataset('glue', 'rte')
metric = load_metric('glue', 'rte')
tokenizer = BertTokenizerFast.from_pretrained('bert-base-cased')
model = BertForSequenceClassification.from_pretrained('bert-base-cased', return_dict=True)
def tokenize(examples):
    return tokenizer(examples['hypothesis'],examples['premiere'] truncation=True, padding='max_length')
dataset = dataset.map(tokenize, batched=True)
1
2
3
4
5
6
7

其它代码一样.更多文本分类参考datawhale-transformer教程4.1：《文本分类》

1.4 更多示例

要查看更多微调示例，您可以参考：