class_docstr: Automatically trains tokenizers and preprocesses machine translation data based on the MTEncDecModelConfig.
  For training NMT models with datasets larger than 5M sentence pairs, 
  it can be inefficient to train them without first creating a tarred dataset. 
  If the user wants to change the tokenizer, vocab size, or batch size, for example, 
  they must reprocess the data with the correct configuration. 
  With MTDataPreproc users can sweep through data configurations and the tarred dataset will 
  be automatically created according to the model configuration.
  To train tokenizer model and create tarred dataset specify in configuration:
      model.preproc_out_dir=/path/to/preproc_out
      model.encoder_tokenizer.vocab_size=32000
      model.decoder_tokenizer.vocab_size=32000 
      model.train_ds.use_tarred_dataset=True 
      model.train_ds.src_file_name=/path/to/src.txt
      model.train_ds.tgt_file_name=/path/to/tgt.txt
      model.train_ds.tokens_in_batch=16000 
  Once a dataset has been constructed based on this configuration, MTDataPreproc will not process it again.
  If a previously trained tokenizer model or tarred dataset is found, MTDataPreproc will not preprocess the data.
  
  Note: the only tokenizer currently supported is YouTokenToMe.
docstr: Create tarred dataset from a large monolingual corpus.
  
  Args:
      clean (str): Cleans sentences to get rid of very long or short sentences.
      fname (str): Path to source text data
      out_dir (str): Path to write tarred dataset
      tokenizer (Any): Path to tokenizer model
      max_seq_length (int): maximum sequence length 
      min_seq_length (int): minimum sequence length 
      tokens_in_batch (int): tokens per batch per GPU, effectively batch size 
      lines_per_dataset_fragment (int): number of lines to consider for bucketing and padding
      num_batches_per_tarfile (int): number of batches (pickle files) within each tarfile
      global_rank (int): if set to zero, data will be processed on this node
      world_size (int): total number of processes being run (for training only, set to 1 when preproc only)
docstr_tok: ["Create","tarred","dataset","from","a","large","monolingual","corpus",".","Args",":","clean","(","str",")",":","Cleans","sentences","to","get","rid","of","very","long","or","short","sentences",".","fname","(","str",")",":","Path","to","source","text","data","out_dir","(","str",")",":","Path","to","write","tarred","dataset","tokenizer","(","Any",")",":","Path","to","tokenizer","model","max_seq_length","(","int",")",":","maximum","sequence","length","min_seq_length","(","int",")",":","minimum","sequence","length","tokens_in_batch","(","int",")",":","tokens","per","batch","per","GPU",",","effectively","batch","size","lines_per_dataset_fragment","(","int",")",":","number","of","lines","to","consider","for","bucketing","and","padding","num_batches_per_tarfile","(","int",")",":","number","of","batches","(","pickle","files",")","within","each","tarfile","global_rank","(","int",")",":","if","set","to","zero",",","data","will","be","processed","on","this","node","world_size","(","int",")",":","total","number","of","processes","being","run","(","for","training","only",",","set","to","1","when","preproc","only",")"]
code: def preprocess_monolingual_dataset(
      clean,
      fname,
      out_dir,
      tokenizer,
      max_seq_length,
      min_seq_length,
      tokens_in_batch,
      lines_per_dataset_fragment,
      num_batches_per_tarfile,
      pkl_file_prefix,
      global_rank,
      world_size,
  ):
      """Create tarred dataset from a large monolingual corpus.
  
      Args:
          clean (str): Cleans sentences to get rid of very long or short sentences.
          fname (str): Path to source text data
          out_dir (str): Path to write tarred dataset
          tokenizer (Any): Path to tokenizer model
          max_seq_length (int): maximum sequence length
          min_seq_length (int): minimum sequence length
          tokens_in_batch (int): tokens per batch per GPU, effectively batch size
          lines_per_dataset_fragment (int): number of lines to consider for bucketing and padding
          num_batches_per_tarfile (int): number of batches (pickle files) within each tarfile
          global_rank (int): if set to zero, data will be processed on this node
          world_size (int): total number of processes being run (for training only, set to 1 when preproc only)
      """
      os.makedirs(out_dir, exist_ok=True)
  
      tar_file_ctr = 1
      num_files_in_tar = 0
      num_lines = 0
      shard_num = 0
      global_batch_ctr = 0
      tmp_f = tempfile.NamedTemporaryFile(delete=False, mode="w")
      tar_file_ptr = tarfile.open(
          os.path.join(
              out_dir,
              "%s-batches.tokens.%d.%d.tar"
              % (pkl_file_prefix, tokens_in_batch, 1),
          ),
          "w",
      )
      metadata_path = os.path.join(
          out_dir, f"metadata.tokens.{tokens_in_batch}.json"
      )
      with open(fname, "r") as f:
          for line in f:
              tmp_f.write(line)
              num_lines += 1
  
              if num_lines == lines_per_dataset_fragment:
                  tmp_f.close()
                  (
                      tar_file_ptr,
                      global_batch_ctr,
                      num_files_in_tar,
                      tar_file_ctr,
                  ) = MTDataPreproc.write_monolingual_batches_to_tarfiles(
                      out_dir,
                      num_batches_per_tarfile,
                      clean,
                      max_seq_length,
                      min_seq_length,
                      tmp_f.name,
                      tokens_in_batch,
                      tokenizer,
                      num_files_in_tar=num_files_in_tar,
                      tar_file_ptr=tar_file_ptr,
                      tar_file_ctr=tar_file_ctr,
                      global_batch_ctr=global_batch_ctr,
                      pkl_file_prefix=pkl_file_prefix,
                  )
  
                  num_lines = 0
                  shard_num += 1
                  os.remove(tmp_f.name)
                  tmp_f = tempfile.NamedTemporaryFile(
                      delete=False, mode="w"
                  )
  
      tmp_f.close()
      (
          tar_file_ptr,
          global_batch_ctr,
          num_files_in_tar,
          tar_file_ctr,
      ) = MTDataPreproc.write_monolingual_batches_to_tarfiles(
          out_dir,
          num_batches_per_tarfile,
          clean,
          max_seq_length,
          min_seq_length,
          tmp_f.name,
          tokens_in_batch,
          tokenizer,
          num_files_in_tar=num_files_in_tar,
          tar_file_ptr=tar_file_ptr,
          tar_file_ctr=tar_file_ctr,
          global_batch_ctr=global_batch_ctr,
          pkl_file_prefix=pkl_file_prefix,
      )
      tar_file_ptr.close()
      os.remove(tmp_f.name)
  
      if num_files_in_tar != num_batches_per_tarfile:
          os.remove(
              os.path.join(
                  out_dir,
                  "%s-batches.tokens.%d.%d.tar"
                  % (pkl_file_prefix, tokens_in_batch, tar_file_ctr),
              )
          )
          global_batch_ctr -= num_files_in_tar
          print(
              "Dropping %d batches because of overflow"
              % (num_files_in_tar)
          )
  
      json.dump(
          {"num_batches": global_batch_ctr},
          open(os.path.join(out_dir, "metadata.json"), "w"),
      )
  
      tar_file_paths = glob.glob(
          f"{out_dir}/{pkl_file_prefix}-batches.tokens.{tokens_in_batch}.*.tar"
      )
  
      num_tar_files = len(tar_file_paths)
      if num_tar_files < world_size:
          raise ValueError(
              (
                  f"Number of tar files found: {num_tar_files} is less than world size: {world_size}. "
                  f"There should be at least one tar file per GPU (ideally many tar files per GPU). "
                  f"This may be due to dataset size, it is advisable to use at least 5M sentence pairs for tarred datasets. "
                  f"Decrease num_batches_per_tarfile or num_tokens_per_batch to increase the number of tarfiles. "
                  f"Also using shard_strategy=replicate will use all available tarfiles for every GPU. "
              )
          )
  
      return tar_file_paths, metadata_path
  
imports: import glob
  import json
  import os
  import pickle
  import tarfile
  import tempfile
  import youtokentome
  from joblib import Parallel, delayed
  from omegaconf import ListConfig, OmegaConf
  from pytorch_lightning import Trainer
  from nemo.collections.common.tokenizers.sentencepiece_tokenizer import SentencePieceTokenizer, create_spt_model
  from nemo.collections.nlp.data.language_modeling.sentence_dataset import SentenceDataset
  from nemo.collections.nlp.data.machine_translation.machine_translation_dataset import TranslationDataset
  from nemo.collections.nlp.models.machine_translation.mt_enc_dec_config import MTEncDecModelConfig
  from nemo.collections.nlp.modules.common.tokenizer_utils import get_nmt_tokenizer, get_tokenizer
  from nemo.utils import logging
label_desc: Use image node_id 9 for calling the MTDataPreproc obj's underlying member method code with example usage: obj.preprocess_monolingual_dataset(clean, fname, out_dir, tokenizer, max_seq_length, min_seq_length, tokens_in_batch, lines_per_dataset_fragment, num_batches_per_tarfile, pkl_file_prefix, global_rank, world_size) and returns: tar_file_paths,  metadata_path