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 large paired translation data.
  
  Args:
      clean (str): Cleans source and target sentences to get rid of noisy data.
      src_fname (str): path to source text data
      tgt_fname (str): path to target text data
      out_dir (str): path to write tarred dataset
      encoder_tokenizer (Any): tokenizer for encoder 
      decoder_tokenizer (Any): tokenizer for decoder
      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
      tar_file_prefix (str) : add string prefix to tar files 
      n_jobs (int): number of processes to use for data processing (-2 to use all but 2)
docstr_tok: ["Create","tarred","dataset","from","large","paired","translation","data",".","Args",":","clean","(","str",")",":","Cleans","source","and","target","sentences","to","get","rid","of","noisy","data",".","src_fname","(","str",")",":","path","to","source","text","data","tgt_fname","(","str",")",":","path","to","target","text","data","out_dir","(","str",")",":","path","to","write","tarred","dataset","encoder_tokenizer","(","Any",")",":","tokenizer","for","encoder","decoder_tokenizer","(","Any",")",":","tokenizer","for","decoder","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","tar_file_prefix","(","str",")",":","add","string","prefix","to","tar","files","n_jobs","(","int",")",":","number","of","processes","to","use","for","data","processing","(","-2","to","use","all","but","2",")"]
code: def preprocess_parallel_dataset(
      clean,
      src_fname,
      tgt_fname,
      out_dir,
      encoder_tokenizer_name,
      encoder_tokenizer_model,
      encoder_tokenizer_r2l,
      encoder_bpe_dropout,
      encoder_model_name,
      decoder_tokenizer_name,
      decoder_tokenizer_model,
      decoder_bpe_dropout,
      decoder_model_name,
      decoder_tokenizer_r2l,
      max_seq_length,
      min_seq_length,
      tokens_in_batch,
      lines_per_dataset_fragment,
      num_batches_per_tarfile,
      global_rank,
      world_size,
      n_jobs=-2,
      tar_file_prefix="parallel",
      encoder_tokenizer_legacy=False,
      decoder_tokenizer_legacy=False,
  ):
      """Create tarred dataset from large paired translation data.
  
      Args:
          clean (str): Cleans source and target sentences to get rid of noisy data.
          src_fname (str): path to source text data
          tgt_fname (str): path to target text data
          out_dir (str): path to write tarred dataset
          encoder_tokenizer (Any): tokenizer for encoder
          decoder_tokenizer (Any): tokenizer for decoder
          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
          tar_file_prefix (str) : add string prefix to tar files
          n_jobs (int): number of processes to use for data processing (-2 to use all but 2)
      """
  
      os.makedirs(out_dir, exist_ok=True)
  
      metadata_path = os.path.join(
          out_dir, f"metadata.tokens.{tokens_in_batch}.json"
      )
  
      if global_rank == 0:
          tar_files_in_out_dir = glob.glob(f"{out_dir}/*.tar")
          if tar_files_in_out_dir:
              logging.info(
                  f"Tarred dataset detected: {tar_files_in_out_dir} and will be used. Remove if reprocessing."
              )
          else:
              filenames = [src_fname, tgt_fname]
  
              # get number of lines so that we can create a partition of the lines of the text file
              num_src_lines, num_tgt_lines = Parallel(n_jobs=2)(
                  delayed(MTDataPreproc._get_num_lines)(filename)
                  for filename in filenames
              )
              logging.info(
                  f"Found {num_src_lines} source lines and  {num_tgt_lines} target lines."
              )
              assert (
                  num_src_lines == num_tgt_lines
              ), "Number of source lines should equal number of target lines."
  
              # create a partition of lines that we can parallelize over
              lines_partition = MTDataPreproc._get_lines_partition(
                  num_src_lines, lines_per_dataset_fragment
              )
              logging.info(
                  f"Found {len(lines_partition)} fragments to parallelize over."
              )
  
              # create tarfiles for each fragment in parallel
              results_list = Parallel(n_jobs=n_jobs)(
                  delayed(MTDataPreproc._process_fragment)(
                      src_filename=src_fname,
                      tgt_filename=tgt_fname,
                      lines_indices=lines_indices,
                      out_dir=out_dir,
                      num_batches_per_tarfile=num_batches_per_tarfile,
                      clean=clean,
                      max_seq_length=max_seq_length,
                      min_seq_length=min_seq_length,
                      tokens_in_batch=tokens_in_batch,
                      encoder_tokenizer_name=encoder_tokenizer_name,
                      encoder_tokenizer_model=encoder_tokenizer_model,
                      encoder_bpe_dropout=encoder_bpe_dropout,
                      encoder_model_name=encoder_model_name,
                      decoder_tokenizer_name=decoder_tokenizer_name,
                      decoder_tokenizer_model=decoder_tokenizer_model,
                      decoder_bpe_dropout=decoder_bpe_dropout,
                      decoder_model_name=decoder_model_name,
                      fragment_index=fragment_index,
                      encoder_tokenizer_r2l=encoder_tokenizer_r2l,
                      decoder_tokenizer_r2l=decoder_tokenizer_r2l,
                      encoder_tokenizer_legacy=encoder_tokenizer_legacy,
                      decoder_tokenizer_legacy=decoder_tokenizer_legacy,
                  )
                  for fragment_index, lines_indices in enumerate(
                      lines_partition
                  )
              )
  
              # compute total batches so far
              total_batches = sum(
                  [batch_count for batch_count, _ in results_list]
              )
  
              # save batches from tar files containing the left over batches (if there's enough batches)
              remainder_tar_file_ctr = 0
              remainder_tar_file_path = os.path.join(
                  out_dir,
                  f"remainder-batches.tokens.{tokens_in_batch}.tar_file_{remainder_tar_file_ctr}.tar",
              )
              remainder_tar_file_ptr = tarfile.open(
                  remainder_tar_file_path, "w"
              )
              batch_in_tar_ctr = 0
              for _, tar_file_path in results_list:
                  tar_file_ptr = tarfile.open(tar_file_path, "r")
                  for member in tar_file_ptr.getmembers():
                      remainder_tar_file_ptr.addfile(
                          member, tar_file_ptr.extractfile(member.name)
                      )
                      batch_in_tar_ctr += 1
                      if batch_in_tar_ctr == num_batches_per_tarfile:
                          remainder_tar_file_ctr += 1
                          remainder_tar_file_ptr.close()
                          remainder_tar_file_path = os.path.join(
                              out_dir,
                              f"remainder-batches.tokens.{tokens_in_batch}.tar_file_{remainder_tar_file_ctr}.tar",
                          )
                          remainder_tar_file_ptr = tarfile.open(
                              remainder_tar_file_path,
                              "w",
                          )
                          batch_in_tar_ctr = 0
                  tar_file_ptr.close()
                  os.remove(tar_file_path)
  
              # log the number of batches remaining as they will be discarded
              num_batches_discarded = len(
                  remainder_tar_file_ptr.getmembers()
              )
              total_batches -= num_batches_discarded
              logging.info(
                  f"Number of batches discarded: {num_batches_discarded}, total batches kept: {total_batches}"
              )
              remainder_tar_file_ptr.close()
              os.remove(remainder_tar_file_path)
  
              # dump metadata to json
              metadata = {}
              metadata["num_batches"] = total_batches
  
              # rename tar files so they can be more easily used with CLI and YAML
              tar_file_paths = glob.glob(f"{out_dir}/*.tar")
              for index, path in enumerate(tar_file_paths):
                  os.rename(
                      path,
                      os.path.join(
                          out_dir,
                          f"{tar_file_prefix}.batches.tokens.{tokens_in_batch}.{index}.tar",
                      ),
                  )
  
              # add tar files to manifest
              tar_file_paths = glob.glob(f"{out_dir}/*.tar")
              metadata["tar_files"] = tar_file_paths
              json.dump(metadata, open(metadata_path, "w"))
  
      tar_file_paths = glob.glob(f"{out_dir}/*.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 5 for calling the MTDataPreproc obj's underlying member method code with example usage: obj.preprocess_parallel_dataset(clean, src_fname, tgt_fname, out_dir, encoder_tokenizer_name, encoder_tokenizer_model, encoder_tokenizer_r2l, encoder_bpe_dropout, encoder_model_name, decoder_tokenizer_name, decoder_tokenizer_model, decoder_bpe_dropout, decoder_model_name, decoder_tokenizer_r2l, max_seq_length, min_seq_length, tokens_in_batch, lines_per_dataset_fragment, num_batches_per_tarfile, global_rank, world_size, n_jobs, tar_file_prefix, encoder_tokenizer_legacy, decoder_tokenizer_legacy) and returns: tar_file_paths,  metadata_path