快速從零構建成DataLoader
- 2022 年 1 月 17 日
- AI
各位朋友大家好，歡迎來到月來客棧。由於公眾號推文不支援後續修訂，以及為了方便大家在PC端進行閱讀，本文將同步推送至網站www.ylkz.life。歡迎大家關注，謝謝！
1 引言
在上一篇文章中，筆者詳細介紹了在文本處理過程中如何通過torch.vocab來快速根據原始語料構建我們所需要的詞表。在接下來的這篇文章中，筆者就來詳細介紹一下如何在上一步的基礎上快速構建Pytorch中的DataLoader，以便於後續網路的訓練。
下面，筆者將會詳細介紹在兩種場景下文本數據集的構建過程。第一種場景是輸入為序列輸出為標籤，也就是類似於文本分類數據集；第二種場景就是輸入輸出均序列，例如翻譯模型，或者文本生成模型等。
2 構建類文本分類數據集
對於構建類似文本分類的數據集來說，總體上可以分為4個步驟：①構建字典；②將文本中的每一個詞（字）轉換為Token序列；③對不同長度的樣本序列按照某個標準進行padding處理；④構建DataLoader類。現在假設我們有如下格式的原始數據：
問君能有幾多愁，恰似一江春水向東流。 0
年年歲歲花相似，歲歲年年人不同。 0
去年今日此門中，人面桃花相映紅。 2
人面不知何處去，桃花依舊笑春風。 1
渺滄海之一粟，羨長江之無窮。3
人面不知何處去，桃花依舊笑春風。 1
月來客棧。 1
其中文本與標籤之間通過一個控制符進行分割，下面我們開始來一步步構建數據集。
2.1 構建字典
由於這部分內容在上一篇文章中已經具體介紹過，所以這裡直接貼出程式碼即可，如下：
def tokenizer(s, word=False):
"""
word: 是否採用分字模式
"""
if word:
r = [w for w in s]
else:
s = jieba.cut(s, cut_all=False)
r = " ".join(s).split()
return r
def build_vocab(tokenizer, filepath, word, min_freq, specials=None):
"""
根據給定的tokenizer和對應參數返回一個Vocab類
Args:
tokenizer: 分詞器
filepath: 文本的路徑
word: 是否採用分字的模式對漢字進行處理
min_freq: 最小詞頻，去掉小於min_freq的詞
specials: 特殊的字元，如<pad>，<unk>等
Returns:
"""
if specials is None:
specials = ['<unk>', '<pad>', '<bos>', '<eos>']
counter = Counter()
with open(filepath, encoding='utf8') as f:
for string_ in f:
counter.update(tokenizer(string_.strip(), word))
return Vocab(counter, min_freq=min_freq, specials=specials)
在完成上述過程後，我們將得到一個Vocab類的實例化對象，通過它便可以得到最終生成的字典：
{'<unk>': 0, '<pad>': 1, '歲': 2, '年': 3, '。': 4, '，': 5, '不': 6, '人': 7, '似': 8, '春': 9, '花': 10, '一': 11, '東': 12,...}
此時，我們就需要定義一個類，並在類的初始化過程中根據訓練語料完成字典的構建，程式碼如下：
class LoadSentenceClassificationDataset():
def __init__(self, train_file_path=None, # 訓練集路徑
tokenizer=None,
batch_size=2,
word=True, # 是否採用分字的模式對漢字進行處理
min_freq=1, # 最小詞頻，去掉小於min_freq的詞
max_sen_len='same'): #最大句子長度，默認設置其長度為整個數據集中最長樣本的長度
# 根據訓練預料建立英語和德語各自的字典
self.tokenizer = tokenizer
self.min_freq = min_freq
self.specials = ['<unk>', '<pad>']
self.word = word
self.vocab = build_vocab(self.tokenizer,
filepath=train_file_path,
word=self.word,
min_freq=self.min_freq,
specials=self.specials)
self.PAD_IDX = self.vocab['<pad>']
self.UNK_IDX = self.vocab['<unk>']
self.batch_size = batch_size
self.max_sen_len = max_sen_len
2.2 轉換為Token序列
在得到構建的字典後，便可以通過如下函數來將訓練集、驗證集和測試集轉換成Token序列：
def data_process(self, filepath):
"""
將每一句話中的每一個詞根據字典轉換成索引的形式，同時返回所有樣本中最長樣本的長度
:param filepath: 數據集路徑
:return:
"""
raw_iter = iter(open(filepath, encoding="utf8"))
data = []
max_len = 0
for raw in raw_iter:
line = raw.rstrip("\n")
s, l = line.split('\t')
tensor_ = torch.tensor([self.vocab[token] for token in
self.tokenizer(s, self.word)], dtype=torch.long)
l = torch.tensor(int(l), dtype=torch.long)
max_len = max(max_len, tensor_.size(0))
data.append((tensor_, l))
return data, max_len
在上述程式碼中，其中第9行用來保存當前數據中最長樣本的長度，在後續padding時會用到；第13-14行是先將原始文本序列tokenize，然後再轉換成每個詞（字）對應的token；第18行將返回包含所有樣本的一個列表，以及當前語料中最長樣本的長度。
例如如下兩行樣本
問君能有幾多愁，恰似一江春水向東流。 0
年年歲歲花相似，歲歲年年人不同。 0
在經過該函數處理後得到的結果為
[(tensor([61, 36, 58, 45, 33, 37, 40, 4, 39, 18, 16, 23, 12, 49, 35, 30, 51, 3]), tensor(0)),
(tensor([ 5, 5, 7, 7, 8, 24, 18, 4, 7, 7, 5, 5, 6, 10, 34, 3]), tensor(0)) ...]
2.3 padding處理
從上面的輸出結果也可以看到，對於不同的樣本來說其對應的長度通常來說都是不同的。但是在將數據輸入到相應模型時卻需要保持同樣的長度，因此在這裡我們就需要對Token序列化後的樣本進行padding處理，具體程式碼如下：
def pad_sequence(sequences, batch_first=False, max_len=None, padding_value=0):
"""
對一個List中的元素進行padding
sequences:
batch_first: 是否把batch_size放到第一個維度
padding_value:
max_len : 最大句子長度，默認為None，即在每個batch中以最長樣本的長度對其它樣本進行padding；
當指定max_len的長度小於一個batch中某個樣本的長度，那麼在這個batch中還是會以最長樣本的長度對其它樣本進行padding
建議指定max_len的值為整個數據集中最長樣本的長度
Returns:
"""
max_size = sequences[0].size()
trailing_dims = max_size[1:]
length = max_len
max_len = max([s.size(0) for s in sequences])
if length is not None:
max_len = max(length, max_len)
if batch_first:
out_dims = (len(sequences), max_len) + trailing_dims
else:
out_dims = (max_len, len(sequences)) + trailing_dims
out_tensor = sequences[0].data.new(*out_dims).fill_(padding_value)
for i, tensor in enumerate(sequences):
length = tensor.size(0)
# use index notation to prevent duplicate references to the tensor
if batch_first:
out_tensor[i, :length, ...] = tensor
else:
out_tensor[:length, i, ...] = tensor
return out_tensor
上述程式碼是根據torch.nn.utils.rnn中的pad_sequence函數修改而來，增加了可以指定一個全局最大長度的參數max_len。在經過pad_sequence函數處理後，所有的樣本就會保持同樣的長度。例如上面的tokenize後的結果在經過padding處理後將變為
[(tensor([61, 36, 58, 45, 33, 37, 40, 4, 39, 18, 16, 23, 12, 49, 35, 30, 51, 3]), tensor(0)),
(tensor([ 5, 5, 7, 7, 8, 24, 18, 4, 7, 7, 5, 5, 6, 10, 34, 3, 1, 1]), tensor(0)) ...]
即第2個樣本的末尾padding了兩個1。
2.4 構建DataLoader迭代器
在經過前面的一系列處理後，我們便可以通過如下程式碼來構建DataLoader迭代器：
def load_train_val_test_data(self, train_file_paths, val_file_paths, test_file_paths):
train_data, max_sen_len = self.data_process(train_file_paths) # 得到處理好的所有樣本
if self.max_sen_len == 'same':
self.max_sen_len = max_sen_len
val_data, _ = self.data_process(val_file_paths)
test_data, _ = self.data_process(test_file_paths)
train_iter = DataLoader(train_data, batch_size=self.batch_size, # 構造DataLoader
shuffle=True, collate_fn=self.generate_batch)
valid_iter = DataLoader(val_data, batch_size=self.batch_size,
shuffle=True, collate_fn=self.generate_batch)
test_iter = DataLoader(test_data, batch_size=self.batch_size,
shuffle=True, collate_fn=self.generate_batch)
return train_iter, valid_iter, test_iter
def generate_batch(self, data_batch):
batch_sentence, batch_label = [], []
for (sen, label) in data_batch: # 開始對一個batch中的每一個樣本進行處理。
batch_sentence.append(sen)
batch_label.append(label)
batch_sentence = pad_sequence(batch_sentence, # [batch_size,max_len]
padding_value=self.PAD_IDX,
batch_first=True,
max_len=self.max_sen_len)
batch_label = torch.tensor(batch_label, dtype=torch.long)
return batch_sentence, batch_label
在上述程式碼中，第1-13行便是用來構造最後需要返回的DataLoader迭代器；而第15-25行則是自定義一個函數來對每個batch中的樣本進行處理，該函數將作為一個參數傳入到類DataLoader中。同時，由於在DataLoader中是對每一個batch的數據進行處理，所以，當max_len=None時這就意味著上面的pad_sequence操作最終表現出來的結果就是不同的樣本，padding後在同一個batch中長度是一樣的，而在不同的batch之間可能是不一樣的。因為此時pad_sequence是以一個batch中最長的樣本為標準對其它樣本進行padding。當max_len = 'same'時，最終表現出來的結果就是，所有樣本在padding後的長度都等於訓練集中最長樣本的長度。
最終，在定義完成類LoadSentenceClassificationDataset後，便可以通過如下方式進行使用：
if __name__ == '__main__':
path = 'data_02.txt'
data_loader = LoadSentenceClassificationDataset(train_file_path=path,
tokenizer=tokenizer,
batch_size=2,
word=True,
max_sen_len='same')
train_iter, valid_iter, test_iter = data_loader.load_train_val_test_data(path, path, path)
for sen, label in train_iter:
print("batch:", sen)
# batch: tensor([[6, 14, 10, 25, 19, 21, 11, 4, 13, 8, 20, 22, 26, 12, 27, 3, 1, 1],
# [61, 36, 58, 45, 33, 37, 40, 4, 39, 18, 16, 23, 12, 49, 35, 30, 51, 3]])
print("batch size:", sen.shape)
# batch size: torch.Size([2, 18])
print("labels:", label)
# labels: tensor([1, 0])
當然，暫時不想理解程式碼的朋友，可以直接將原始數據整理成上述一樣的格式然後導入類LoadSentenceClassificationDataset使用即可（下載地址見文末[1]）。
3 構建類翻譯模型數據集
通常，在NLP中還有一類任務就是模型的輸入和輸出均為序列的形式，這就需要在訓練的過程中同時將這兩部分輸入到模型中。例如在翻譯模型的訓練過程中，需要同時將原始序列和目標序列一同輸入到網路模型里。對於構建這種類似翻譯模型的數據集總體上同樣也可以採用上面的4個步驟：①構建字典；②將文本中的每一個詞（字）轉換為Token序列；③對不同長度的樣本序列按照某個標準進行padding處理；④構建DataLoader類。只是在每一步中都需要分別對原始序列和目標序列進行處理。
現在假設我們有如下格式的平行語料數據：
# 原始序列 source sequence
Zwei junge weiße Männer sind im, Freien in der Nähe vieler Büsche.
Mehrere Männer mit Schutzhelmen bedienen ein Antriebsradsystem.
Ein kleines Mädchen klettert in ein Spielhaus aus Holz.
Ein Mann in einem blauen Hemd steht auf einer Leiter und putzt ein Fenster.
Zwei Männer stehen am Herd und bereiten Essen zu.
# 目標序列 target sequence
Two young, White males are outside near many bushes.
Several men in hard hats are operating a giant pulley system.
A little girl climbing into a wooden playhouse.
A man in a blue shirt is standing on a ladder cleaning a window.
Two men are at the stove preparing food.
從上面的語料可以看出，這是一個用於訓練翻譯模型的數據，原始序列為德語，目標序列為英語。下面我們開始來一步步的構建數據集。
3.1 構建字典
在構建字典的過程中，整體上與2.1節內容中的一樣，只是在這裡需要同時對原始序列和目標序列分別構建一個字典。具體程式碼如下所示：
def my_tokenizer(s):
s = s.replace(',', " ,").replace(".", " .")
return s.split()
def build_vocab(tokenizer, filepath, specials=None):
if specials is None:
specials = ['<unk>', '<pad>', '<bos>', '<eos>']
counter = Counter()
with open(filepath, encoding='utf8') as f:
for string_ in f:
counter.update(tokenizer(string_))
return Vocab(counter, specials=specials)
在上述程式碼中，第1-3行為自定義的一個tokenizer；雖然上述兩種語料可以直接通過空格來對每個詞進行分割，但是還需要做的就是在單詞和符號之間加上一個空格，以便把符號分割出來。第5-12行定義的build_vocab函數還是同之前的一樣，沒有發生改變。
在完成上述過程後，我們將得到兩個Vocab類的實例化對象。
一個為原始序列的字典：
{'<unk>': 0, '<pad>': 1, '<bos>': 2, '<eos>': 3, '.': 4, 'Männer': 5, 'ein': 6, 'in': 7, 'Ein': 8, 'Zwei': 9, 'und': 10, ',': 11, ......}
一個為目標序列的字典：
{'<unk>': 0, '<pad>': 1, '<bos>': 2, '<eos>': 3, '.': 4, 'a': 5, 'are': 6, 'A': 7, 'Two': 8, 'in': 9, 'men': 10, ',': 11, 'Several': 12,......}
此時，我們就需要定義一個類，並在類的初始化過程中根據訓練語料完成字典的構建，程式碼如下：
class LoadEnglishGermanDataset():
def __init__(self, train_file_paths=None, tokenizer=None, batch_size=2):
# 根據訓練預料建立英語和德語各自的字典
self.tokenizer = tokenizer
self.de_vocab = build_vocab(self.tokenizer, filepath=train_file_paths[0])
self.en_vocab = build_vocab(self.tokenizer, filepath=train_file_paths[1])
self.specials = ['<unk>', '<pad>', '<bos>', '<eos>']
self.PAD_IDX = self.de_vocab['<pad>']
self.BOS_IDX = self.de_vocab['<bos>']
self.EOS_IDX = self.de_vocab['<eos>']
self.batch_size = batch_size
3.2 轉換為Token序列
在得到構建的字典後，便可以通過如下函數來將訓練集、驗證集和測試集轉換成Token序列：
def data_process(self, filepaths):
"""
將每一句話中的每一個詞根據字典轉換成索引的形式
:param filepaths:
:return:
"""
raw_de_iter = iter(open(filepaths[0], encoding="utf8"))
raw_en_iter = iter(open(filepaths[1], encoding="utf8"))
data = []
for (raw_de, raw_en) in zip(raw_de_iter, raw_en_iter):
de_tensor_ = torch.tensor([self.de_vocab[token] for token in
self.tokenizer(raw_de.rstrip("\n"))], dtype=torch.long)
en_tensor_ = torch.tensor([self.en_vocab[token] for token in
self.tokenizer(raw_en.rstrip("\n"))], dtype=torch.long)
data.append((de_tensor_, en_tensor_))
return data
在上述程式碼中，第11-4行分別用來將原始序列和目標序列轉換為對應詞表中的Token形式。在處理完成後，就會得到類似如下的結果：
[(tensor([9,37, 46, 5, 42, 36, 11, 16,7, 33, 24, 45, 13,4]),tensor([8,45, 11, 13,28, 6, 34,31, 30,16, 4])),
(tensor([22, 5, 40, 25, 30, 6, 12, 4]), tensor([12, 10, 9, 22, 23, 6, 33, 5, 20, 37, 41, 4])),
(tensor([8, 38, 23, 39, 7, 6, 26, 29, 19, 4]), tensor([ 7, 27, 21, 18, 24, 5, 44, 35, 4])),
(tensor([ 9, 5, 43, 27, 18, 10, 31, 14, 47, 4]), tensor([ 8, 10, 6, 14, 42, 40, 36, 19, 4])) ]
其中左邊的一列就是原始序列的Token形式，右邊一列就是目標序列的Token形式，每一行構成一個樣本。
3.3 padding處理
同樣，從上面的輸出結果可以看到，無論是對於原始序列來說還是目標序列來說，在不同的樣本中其對應長度都不盡相同。但是在將數據輸入到相應模型時卻需要保持同樣的長度，因此在這裡我們就需要對Token序列化後的樣本進行padding處理。同時需要注意的是，一般在這種生成模型中，模型在訓練過程中只需要保證同一個batch中所有的原始序列等長，所有的目標序列等長，也就是說不需要在整個數據集中所有樣本都保證等長。
因此，在實際處理過程中無論是原始序列還是目標序列都會以每個batch中最長的樣本為標準對其它樣本進行padding，具體程式碼如下：
def generate_batch(self, data_batch):
de_batch, en_batch = [], []
for (de_item, en_item) in data_batch: # 開始對一個batch中的每一個樣本進行處理。
de_batch.append(de_item) # 編碼器輸入序列不需要加起止符
# 在每個idx序列的首位加上 起始token 和 結束 token
en = torch.cat([torch.tensor([self.BOS_IDX]), en_item, torch.tensor([self.EOS_IDX])], dim=0)
en_batch.append(en)
# 以最長的序列為標準進行填充
de_batch = pad_sequence(de_batch, padding_value=self.PAD_IDX) # [de_len,batch_size]
en_batch = pad_sequence(en_batch, padding_value=self.PAD_IDX) # [en_len,batch_size]
return de_batch, en_batch
在上述程式碼中，第6-7行用來在目標序列的首尾加上特定的起止符；第9-10行則是分別對一個batch中的原始序列和目標序列以各自當中最長的樣本為標準進行padding（這裡的pad_sequence導入自torch.nn.utils.rnn）。
3.4 構建DataLoader迭代器
在經過前面的一系列處理後，我們便可以通過如下程式碼來構建DataLoader迭代器：
def load_train_val_test_data(self, train_file_paths, val_file_paths, test_file_paths):
train_data = self.data_process(train_file_paths)
val_data = self.data_process(val_file_paths)
test_data = self.data_process(test_file_paths)
train_iter = DataLoader(train_data, batch_size=self.batch_size,
shuffle=True, collate_fn=self.generate_batch)
valid_iter = DataLoader(val_data, batch_size=self.batch_size,
shuffle=True, collate_fn=self.generate_batch)
test_iter = DataLoader(test_data, batch_size=self.batch_size,
shuffle=True, collate_fn=self.generate_batch)
return train_iter, valid_iter, test_iter
最終，在定義完成類LoadEnglishGermanDataset後，便可以通過如下方式進行使用：
if __name__ == '__main__':
train_filepath = ['train_.de',
'train_.en']
data_loader = LoadEnglishGermanCorpus(train_filepath, tokenizer=my_tokenizer, batch_size=2)
train_iter, valid_iter, test_iter = data_loader.load_train_val_test_data(train_filepath,
train_filepath,
train_filepath)
for src, tgt in train_iter:
print("src shape：", src.shape) # [de_tensor_len,batch_size]
print("tgt shape:", tgt.shape) # [de_tensor_len,batch_size]
print("===================")
運行結果為：
src shape：torch.Size([12, 2])
tgt shape: torch.Size([14, 2])
===================
src shape：torch.Size([16, 2])
tgt shape: torch.Size([13, 2])
===================
src shape：torch.Size([17, 1])
tgt shape: torch.Size([17, 1])
===================
從上述結果可以看出，對於同一個batch來說，原始序列的長度都相同，目標序列的長度也都相同。最後，如果暫時不想理解程式碼的朋友，可以直接將原始數據整理成上述一樣的格式，然後導入類LoadEnglishGermanDataset使用即可。
4 總結
在這篇文章中，筆者首先介紹了如何快速的構建類似文本分類模型中的DataLoader迭代器，包括詞表字典的構建、序列轉換為Token、對不同長度的樣本進行padding等；然後介紹了如何構建類似翻譯模型中的數據迭代器，並且也對其用法進行了示例。
本次內容就到此結束，感謝您的閱讀！如果你覺得上述內容對你有所幫助，歡迎分享至一位你的朋友！若有任何疑問與建議，請添加筆者微信nulls8或加群進行交流。青山不改，綠水長流，我們月來客棧見！
引用
[1] 完整程式碼：//github.com/moon-hotel/DeepLearningWithMe
