Merge pull request #2 from Tridu33/openmind

feat: add imdb distributed demo for mindnlp Trainer API

Author: Tridu33

examples/parallel/bert_imdb_finetune/bert_imdb_finetune_cpu_mindnlp_trainer_npus_same.py

@@ -0,0 +1,97 @@
+#!/usr/bin/env python
+# coding: utf-8
+"""
+python bert_imdb_finetune_cpu_mindnlp_trainer.py
+bash bert_imdb_finetune_npu_mindnlp_trainer.sh
+"""
+
+def main():
+    import mindspore
+    from mindspore.dataset import transforms
+    from mindnlp.engine import Trainer
+    from mindnlp.dataset import load_dataset
+
+    imdb_ds = load_dataset('imdb', split=['train', 'test'])
+    imdb_train = imdb_ds['train']
+    imdb_test = imdb_ds['test']
+    imdb_train.get_dataset_size()
+    from mindnlp.transformers import AutoTokenizer
+    # tokenizer
+    tokenizer = AutoTokenizer.from_pretrained('bert-base-cased')
+
+    def process_dataset(dataset, tokenizer, max_seq_len=256, batch_size=32, shuffle=False):
+        is_ascend = mindspore.get_context('device_target') == 'Ascend'
+        def tokenize(text):
+            if is_ascend:
+                tokenized = tokenizer(text, padding='max_length', truncation=True, max_length=max_seq_len)
+            else:
+                tokenized = tokenizer(text, truncation=True, max_length=max_seq_len)
+            return tokenized['input_ids'], tokenized['token_type_ids'], tokenized['attention_mask']
+
+        if shuffle:
+            dataset = dataset.shuffle(batch_size)
+
+        # map dataset
+        dataset = dataset.map(operations=[tokenize], input_columns="text", output_columns=['input_ids', 'token_type_ids', 'attention_mask'])
+        dataset = dataset.map(operations=transforms.TypeCast(mindspore.int32), input_columns="label", output_columns="labels")
+        # batch dataset
+        if is_ascend:
+            dataset = dataset.batch(batch_size)
+        else:
+            dataset = dataset.padded_batch(batch_size, pad_info={'input_ids': (None, tokenizer.pad_token_id),
+                                                                'token_type_ids': (None, 0),
+                                                                'attention_mask': (None, 0)})
+
+        return dataset
+
+    # split train dataset into train and valid datasets
+    imdb_train, imdb_val = imdb_train.split([0.7, 0.3])
+
+    dataset_train = process_dataset(imdb_train, tokenizer, shuffle=True)
+    dataset_val = process_dataset(imdb_val, tokenizer)
+    dataset_test = process_dataset(imdb_test, tokenizer)
+
+    next(dataset_train.create_tuple_iterator())
+
+    from mindnlp.transformers import AutoModelForSequenceClassification
+
+    # set bert config and define parameters for training
+    model = AutoModelForSequenceClassification.from_pretrained('bert-base-cased', num_labels=2)
+
+    from mindnlp.engine import TrainingArguments
+    
+    training_args = TrainingArguments(
+        output_dir="bert_imdb_finetune_cpu",
+        evaluation_strategy="epoch",
+        save_strategy="epoch",
+        logging_strategy="epoch",
+        load_best_model_at_end=True,
+        num_train_epochs=2.0,
+        learning_rate=2e-5
+    )
+    training_args = training_args.set_optimizer(name="adamw", beta1=0.8) # OptimizerNames.SGD
+
+    from mindnlp import evaluate
+    import numpy as np
+    metric = evaluate.load("accuracy")
+    def compute_metrics(eval_pred):
+        logits, labels = eval_pred
+        predictions = np.argmax(logits, axis=-1)
+        return metric.compute(predictions=predictions, references=labels)
+    
+    trainer = Trainer(
+        model=model,
+        args=training_args,
+        train_dataset=dataset_train,
+        eval_dataset=dataset_val,
+        compute_metrics=compute_metrics
+    )
+    print("Start training")
+    trainer.train()
+
+    print("Start checking the test set")
+    trainer.evaluate(dataset_test)
+
+if __name__ == '__main__':
+    main()
+

examples/parallel/bert_imdb_finetune/bert_imdb_finetune_npu_mindnlp_trainer.sh

@@ -0,0 +1,25 @@
+#!/bin/bash
+
+echo "=========================================="
+echo "Please run the script as: "
+echo "bash bert_imdb_finetune_npu_mindnlp_trainer.sh"
+echo "==========================================="
+
+EXEC_PATH=$(pwd)
+if [ ! -d "${EXEC_PATH}/data" ]; then
+    if [ ! -f "${EXEC_PATH}/emotion_detection.tar.gz" ]; then
+        wget wget https://baidu-nlp.bj.bcebos.com/emotion_detection-dataset-1.0.0.tar.gz -O emotion_detection.tar.gz
+    fi
+    tar xvf emotion_detection.tar.gz
+fi
+export DATA_PATH=${EXEC_PATH}/data/
+
+rm -rf bert_imdb_finetune_cpu_mindnlp_trainer_npus_same
+mkdir bert_imdb_finetune_cpu_mindnlp_trainer_npus_same
+echo "start training"
+
+export MULTI_NPU="true" 
+# unset MULTI_NPU
+msrun --worker_num=2 --local_worker_num=2 --master_port=8119 \
+--log_dir=bert_imdb_finetune_cpu_mindnlp_trainer_npus_same --join=True \
+--cluster_time_out=30 bert_imdb_finetune_cpu_mindnlp_trainer_npus_same.py 

mindnlp/accelerate/accelerator.py

@@ -105,20 +105,13 @@ def prepare(self, *args):
         """
         result = []
 
-        # Only support mindsormers and MULTI_NPU_DP now
+        # Only support mindsormers and MULTI_NPU now
         if self.distributed_type == DistributedType.MINDFORMERS:
             result = self._prepare_mindformers(*args)
-        elif self.distributed_type == DistributedType.MULTI_NPU_DP:
-            result = self._prepare_data_parallel_native_minspore(*args)
+        elif self.distributed_type == DistributedType.MULTI_NPU:
+            pass # nothing prepare for data parallel
         return result
 
-    def _prepare_data_parallel_native_minspore(self, *args):
-        # initialize data parallel for native mindspore
-        mindspore.set_context(mode=mindspore.GRAPH_MODE)
-        mindspore.set_auto_parallel_context(parallel_mode=mindspore.ParallelMode.DATA_PARALLEL, gradients_mean=True)
-        mindspore.communication.init()
-        mindspore.set_seed(numpy.random.seed())
-
     def _prepare_mindformers(self, *args):
         mindformers_plugin = self.state.mindformers_plugin
 

mindnlp/accelerate/state.py

@@ -347,9 +347,9 @@ def _prepare_backend(self):
         if accelerate_distributed_type == DistributedType.MINDFORMERS and is_mindformers_available():
             self.backend = "hccl"
             self.distributed_type = DistributedType.MINDFORMERS
-        elif accelerate_distributed_type == DistributedType.MULTI_NPU_DP:
+        elif accelerate_distributed_type == DistributedType.MULTI_NPU:
             self.backend = "hccl"
-            self.distributed_type = DistributedType.MULTI_NPU_DP
+            self.distributed_type = DistributedType.MULTI_NPU
 
     @num_processes.setter
     def num_processes(self, value):
@@ -372,8 +372,8 @@ def __init__(self, mindformers_plugin=None, **kwargs):
             PartialState(**kwargs)
         self.__dict__.update(PartialState._shared_state)
         # set distributed_type
-        if accelerate_distributed_type == DistributedType.MULTI_NPU_DP:
-            self.distributed_type = DistributedType.MULTI_NPU_DP
+        if accelerate_distributed_type == DistributedType.MULTI_NPU:
+            self.distributed_type = DistributedType.MULTI_NPU
         elif accelerate_distributed_type == DistributedType.MINDFORMERS:
             self.distributed_type = DistributedType.MINDFORMERS
             self.mindformers_plugin = mindformers_plugin

mindnlp/accelerate/utils/constants.py

@@ -1,7 +1,18 @@
 """constants"""
 import os
+import mindspore
+import numpy
 from .dataclasses import DistributedType
 
+
+def _prepare_data_parallel_native_minspore():
+    # initialize data parallel hcc backend for data_loader and Trainer API
+    mindspore.set_context(mode=mindspore.GRAPH_MODE)
+    mindspore.set_auto_parallel_context(parallel_mode=mindspore.ParallelMode.DATA_PARALLEL, gradients_mean=True)
+    mindspore.communication.init()
+    random_seed = numpy.random.randint(10000)
+    mindspore.set_seed(random_seed)
+    
 def detect_accelerate_distributed_type():
     """
     detect distributed_type
@@ -10,8 +21,9 @@ def detect_accelerate_distributed_type():
         _type_: According to the factors such as the available parallel software and hardware environment of the current system and the user-specified parallel scheme,
           the optimal parallel strategy is comprehensively decided in different situations.
     """
-    if os.environ.get("MULTI_NPU_DP", None) == "true": 
-        return DistributedType.MULTI_NPU_DP
+    if os.environ.get("MULTI_NPU", None) == "true": 
+        _prepare_data_parallel_native_minspore()
+        return DistributedType.MULTI_NPU
     if os.environ.get("ACCELERATE_USE_MINDFORMERS", "false") == "true": 
         return DistributedType.MINDFORMERS
     else:

mindnlp/accelerate/utils/dataclasses.py

@@ -18,10 +18,10 @@ class DistributedType(str, enum.Enum):
     Values:
         - **MINDFORMERS** -- Using mindformers
         - **NO** -- Not a distributed environment, just a single process.
-        - **MULTI_NPU_DP** -- Distributed data parallel on multiple NPUs.
+        - **MULTI_NPU** -- Distributed data parallel on multiple NPUs.
     """
 
-    MULTI_NPU_DP = "MULTI_NPU_DP"
+    MULTI_NPU = "MULTI_NPU"
     MINDFORMERS = "MINDFORMERS"
     NO = "NO"
 

mindnlp/dataset/load.py

@@ -335,7 +335,7 @@ def load_dataset(
         column_names = list(raw_ds.features.keys())
         source = TransferDataset(raw_ds, column_names) if isinstance(raw_ds, Dataset) \
             else TransferIterableDataset(raw_ds, column_names)
-        if accelerate_distributed_type == DistributedType.MULTI_NPU_DP:
+        if accelerate_distributed_type == DistributedType.MULTI_NPU:
             ms_ds = GeneratorDataset(source=source, 
             column_names=column_names, 
             shuffle=shuffle,

mindnlp/engine/trainer/base.py

@@ -90,7 +90,6 @@
     TrainerControl,
     TrainerState,
 )
-from ..utils import _get_learning_rate
 
 
 logger = logging.get_logger(__name__)
@@ -126,7 +125,7 @@ def _is_peft_model(model):
 class Trainer:
     """
     Trainer is a simple but feature-complete training and eval loop for MindSpore, optimized for 🤗 Transformers.
-    """
+    """ 
     def __init__(
         self,
         model: Union[PreTrainedModel, nn.Module] = None,
@@ -288,6 +287,29 @@ def __init__(
         self._created_lr_scheduler = False
         self.actual_distributed_type = accelerate_distributed_type
 
+
+    def _get_learning_rate(self):
+        r"""
+        This function retrieves the learning rate used by the optimizer.
+        
+        Args:
+            self: An instance of the class containing the optimizer and learning rate scheduler.
+        
+        Returns:
+            The learning rate value (float) used by the optimizer.
+        
+        Raises:
+            None.
+        """
+        if isinstance(self.lr_scheduler, optim.lr_scheduler.ReduceLROnPlateau):
+            last_lr = self.optimizer.param_groups[0]["lr"]
+        else:
+            last_lr = self.lr_scheduler.get_last_lr()[0]
+        if ops.is_tensor(last_lr):
+            last_lr = last_lr.item()
+        return last_lr
+
+
     def _activate_neftune(self, model):
         r"""
         Activates the neftune as presented in this code: https://github.com/neelsjain/NEFTune and paper:
@@ -1136,6 +1158,7 @@ def _inner_training_loop(
                             model.parameters(),
                             args.max_grad_norm,
                         )
+                    
                     # Optimizer step
                     self.optimizer.step()
 
@@ -1376,7 +1399,7 @@ def training_step(self, model: nn.Module, inputs: Dict[str, Union[mindspore.Tens
         inputs = self._prepare_inputs(inputs)
 
         def forward(inputs):
-            if accelerate_distributed_type == DistributedType.MULTI_NPU_DP:
+            if accelerate_distributed_type == DistributedType.MULTI_NPU:
                 from mindspore.communication import get_group_size
                 import mindspore.ops as msops
                 rank_size = get_group_size()

mindnlp/engine/utils.py

@@ -505,28 +505,6 @@ def speed_metrics(split, start_time, num_samples=None, num_steps=None, num_token
         result[f"{split}_tokens_per_second"] = round(tokens_per_second, 3)
     return result
 
-def _get_learning_rate(self):
-    r"""
-    This function retrieves the learning rate used by the optimizer.
-    
-    Args:
-        self: An instance of the class containing the optimizer and learning rate scheduler.
-    
-    Returns:
-        The learning rate value (float) used by the optimizer.
-    
-    Raises:
-        None.
-    """
-    if isinstance(self.lr_scheduler, optim.lr_scheduler.ReduceLROnPlateau):
-        last_lr = self.optimizer.param_groups[0]["lr"]
-    else:
-        last_lr = self.lr_scheduler.get_last_lr()[0]
-    if ops.is_tensor(last_lr):
-        last_lr = last_lr.item()
-    return last_lr
-
-
 def find_batch_size(tensors):
     """
     Find the first dimension of a tensor in a nested list/tuple/dict of tensors.