Adopt new data pipeline for GPT. (PaddlePaddle#930)

Author: ZHUI

examples/language_model/data_tools/README.md

@@ -30,8 +30,10 @@
  - tqdm
  - numpy
  - pybind11
+ - lac (可选)
+ - zstandard (可选)
 
-安装命令`pip install tqdm numpy pybind11`。另，部分功能需要`g++>=4.8`编译支持
+安装命令`pip install tqdm numpy pybind11 lac zstandard`。另，部分功能需要`g++>=4.8`编译支持
 
 
 ## 训练全流程数据Pipeline
@@ -179,6 +181,12 @@ sh run_static.sh
 可以自定义的选项有do_whole_word_mask, favor_longer_ngram, do_permutation, geometric_dist等，
 可以参考[Megatron](https://github.com/NVIDIA/Megatron-LM)使用这些lm_mask策略。
 
+### FAQ
+
+#### C++代码编译失败怎么办？
+- 请先检查pybind11包是否安装，g++、make工具是否正常。
+- 编译失败可能是本文件夹下的Makefile命令出现了一些问题。可以将Makefile中的python3、python3-config设置成完全的路径，如/usr/bin/python3.7。
+
 ## 参考内容
 
 注: 大部分数据流程，参考自[Megatron](https://github.com/NVIDIA/Megatron-LM)，特此表达感谢。

examples/language_model/data_tools/create_pretraining_data.py

@@ -26,6 +26,12 @@
 
 import paddlenlp.transformers as tfs
 
+try:
+    import nltk
+    nltk_available = True
+except ImportError:
+    nltk_available = False
+
 
 def get_args():
     parser = argparse.ArgumentParser()
@@ -137,7 +143,6 @@ def process(line):
 
 def jieba_segmentation_fn():
     import jieba
-    jieba.initialize()  # 手动初始化（可选）
 
     def process(line):
         words = jieba.cut(line)
@@ -280,12 +285,11 @@ def encode(self, json_line):
         if len(doc_ids) > 0 and self.args.append_eos:
             doc_ids[-1].append(Converter.tokenizer.eos_token_id)
 
-        return doc_ids, len(json_line.encode("utf-8"))
+        return doc_ids, len(text.encode("utf-8"))
 
 
 def main():
     args = get_args()
-    startup_start = time.time()
 
     file_paths = []
     if os.path.isfile(args.input_path):
@@ -318,16 +322,28 @@ def main():
 
     sent_count = 0
     token_count = 0
+
+    file_paths.sort()
+
+    step = 0
+    total_bytes_processed = 0
+    startup_start = time.time()
     for file_path in tqdm(file_paths):
-        total_bytes_processed = 0
-        text = open(file_path, 'r', encoding='utf-8')
-        encoded_docs = pool.imap(convert.encode, text, 256)
+        if file_path.endswith(".zst"):
+            import zstandard
+            cctx = zstandard.ZstdDecompressor()
+            fh = open(file_path, 'rb')
+            text = io.BufferedReader(cctx.stream_reader(fh))
+        elif file_path.endswith(".jsonl"):
+            text = open(file_path, 'r', encoding='utf-8')
+        else:
+            print("Unexpected data format, skiped %s" % file_path)
+            continue
 
-        startup_end = time.time()
-        proc_start = time.time()
-        print("Time to startup:", startup_end - startup_start)
+        encoded_docs = pool.imap(convert.encode, text, 256)
         print("Processing %s" % file_path)
         for i, (doc, bytes_processed) in enumerate(encoded_docs, start=1):
+            step += 1
             total_bytes_processed += bytes_processed
             if len(doc) == 0:
                 continue
@@ -352,13 +368,13 @@ def main():
                 sent_count.to_bytes(
                     8, byteorder='little', signed=True))
 
-            if i % args.log_interval == 0:
+            if step % args.log_interval == 0:
                 current = time.time()
-                elapsed = current - proc_start
+                elapsed = current - startup_start
                 mbs = total_bytes_processed / elapsed / 1024 / 1024
                 print(
-                    f"Processed {i} documents",
-                    f"({i/elapsed:.2f} docs/s, {mbs:.4f} MB/s).",
+                    f"Processed {step} documents",
+                    f"({step/elapsed:.2f} docs/s, {mbs:.4f} MB/s).",
                     file=sys.stderr)
 
     pool.close()

examples/language_model/ernie-1.0/run_pretrain_static.py

@@ -644,7 +644,7 @@ def do_train(args):
             if global_step >= args.max_steps:
                 eval_fetch = []
                 if topo.is_last:
-                    eval_fetch = [loss]
+                    eval_fetch = [loss, lm_loss, sop_loss]
 
                 run_evaluate(test_data_loader, exe, test_program,
                              args.test_iters, log_writer, global_step, args,

examples/language_model/gpt-3/static/run_pretrain_static.py

@@ -19,6 +19,7 @@
 import os
 import random
 import time
+import sys
 
 os.path.expandvars('$HOME')
 os.path.expanduser('~')
@@ -34,6 +35,8 @@
 import paddlenlp.ops as ops
 from visualdl import LogWriter
 
+# Used to load the data_tools path, should import before dataset
+sys.path.insert(0, "../../")
 from dataset import create_pretrained_dataset
 from args import parse_args
 import lr
@@ -121,12 +124,25 @@ def dist_optimizer(args, topo):
 def get_train_data_file(args):
     files = [
         os.path.join(args.input_dir, f) for f in os.listdir(args.input_dir)
-        if (os.path.isfile(os.path.join(args.input_dir, f)) and "npz_" not in
-            str(f))
+        if (os.path.isfile(os.path.join(args.input_dir, f)) and str(f).endswith(
+            "_idx.npz"))
+    ]
+    files = [x.replace("_idx.npz", "") for x in files]
+    if len(files) == 0:
+        logger.warning(
+            "Not found dataset with name of xxx_ids.npy and xxx_idx.npz! Try to found old compatible xxx_ids.npz file."
+        )
+    else:
+        return files
+
+    files = [
+        os.path.join(args.input_dir, f) for f in os.listdir(args.input_dir)
+        if (os.path.isfile(os.path.join(args.input_dir, f)) and str(f).endswith(
+            "_ids.npz"))
     ]
 
-    data_file = files[0]
-    return data_file
+    files = [x.replace("_ids.npz", "") for x in files]
+    return files
 
 
 def init_static_with_params(model, dygraph_params, topo, prog=None):
@@ -189,6 +205,7 @@ def do_train(args):
 
     worker_num = fleet.worker_num()
     worker_index = fleet.worker_index()
+    local_rank = 0 if fleet.local_rank() is None else int(fleet.local_rank())
 
     topo = Topology(
         device_rank=worker_index,
@@ -237,6 +254,7 @@ def do_train(args):
                 train_data_loader, valid_data_loader, test_data_loader = create_pretrained_dataset(
                     args,
                     data_file,
+                    local_rank=local_rank,
                     data_world_size=topo.data_info.size,
                     data_world_rank=topo.data_info.rank,
                     eos_id=eos_id,

examples/language_model/gpt/README.md

@@ -8,10 +8,8 @@ GPT-[2](https://cdn.openai.com/better-language-models/language_models_are_unsupe
 ```text
 .
 ├── args.py                 # 训练参数配置
-├── create_pretraining_data.py         # 数据预处理脚本
 ├── converter.py            # 权重转化脚本
 ├── dataset.py              # 数据处理
-├── decompress.sh           # 数据集解压脚本
 ├── deploy/                 # 模型部署的inference脚本
 ├── export_model.py         # 导出预测部署的模型脚本
 ├── faster_gpt/             # 使用 FasterGPT 高性能预测 sample  
@@ -39,43 +37,45 @@ GPT-[2](https://cdn.openai.com/better-language-models/language_models_are_unsupe
 
 ### 数据准备
 
-#### 原始数据获取
+#### 数据获取与制作
 
 [OpenWebTextCorpus](https://skylion007.github.io/OpenWebTextCorpus/)是一个开源的英文网页文本数据集，数据来源于Reddit，经过去重、清洗、提取，最终包含800多万个文档。
+本示例采用EleutherAI清洗好的[OpenWebText2数据](https://openwebtext2.readthedocs.io/en/latest/index.html#download-plug-and-play-version)
 
 下载以后通过以下命令解压：
 
 ```shell
-xz -d openwebtext.tar.xz
-tar xf openwebtext.tar
-mkdir raw_data
-bash decompress.sh
+wget https://the-eye.eu/public/AI/pile_preliminary_components/openwebtext2.jsonl.zst.tar ./
+tar -xvf openwebtext2.json.zst.tar -C  /path/to/openwebtext
 ```
 
-解压以后得到的`raw_data`目录大小约为54GB。
+然后使用[data_tools](../data_tools)工具下的`create_pretraining_data.py`脚本进行数据集制作：
 
-#### 数据预处理
-
-为了提升训练速度，我们在训练前将文本数据转成相应的id，并保存为npz格式：
-
-```shell
-python create_pretraining_data.py --input_path raw_data \
- --model_name gpt2-en \
- --append_eod \
- --workers 8
 ```
+python -u  create_pretraining_data.py \
+    --model_name gpt2-en \
+    --tokenizer_name GPTTokenizer \
+    --data_format JSON \
+    --input_path /path/to/openwebtext/ \
+    --append_eos \
+    --output_prefix gpt_openwebtext  \
+    --workers 40 \
+    --log_interval 10000
+```
+处理时间约一个小时左右，就可以得到我们需要的`gpt_openwebtext_ids.npy`, `gpt_openwebtext_idx.npz`数据集文件。
 
-运行命令后，产出`raw_data_ids.npz`文件。为了方便用户运行测试本模型，本项目提供了处理好的300M的训练样本：
-
+为了方便用户运行测试本模型，本项目提供了处理好的300M的训练样本：
 ```shell
-wget https://paddlenlp.bj.bcebos.com/models/transformers/gpt/train.data.json_ids.npz
+wget https://paddlenlp.bj.bcebos.com/models/transformers/gpt/data/gpt_en_dataset_300m_ids.npy
+wget https://paddlenlp.bj.bcebos.com/models/transformers/gpt/data/gpt_en_dataset_300m_idx.npz
 ```
 
-将所有预处理得到的npz文件统一放入一个文件夹中，以备训练使用：
+将所有预处理得到的文件统一放入一个文件夹中，以备训练使用：
 
 ```
 mkdir data
-mv train.data.json_ids.npz data
+mv gpt_en_dataset_300m_ids.npy ./data
+mv gpt_en_dataset_300m_idx.npz ./data
 ```
 
 ### 模型训练

examples/language_model/gpt/args.py

@@ -129,17 +129,11 @@ def parse_args(MODEL_CLASSES):
         help="Epsilon for Adam optimizer.")
 
     # Training steps config
-    parser.add_argument(
-        "--num_train_epochs",
-        default=1,
-        type=int,
-        help="Total number of training epochs to perform.", )
     parser.add_argument(
         "--max_steps",
         default=500000,
         type=int,
-        help="If > 0: set total number of training steps to perform. Override num_train_epochs."
-    )
+        help="set total number of training steps to perform.")
     parser.add_argument(
         "--save_steps",
         type=int,