code: def prepare_data(sc, folder, vocabsize, training_split):
      if not folder.startswith("hdfs://"):
          file = download_data(folder)
      else:
          file = folder
      sentences_rdd = sc.textFile(file).map(
          lambda line: sentence.sentences_split(line)
      )
      pad_sent = sentences_rdd.flatMap(lambda x: x).map(
          lambda sent: sentence.sentences_bipadding(sent)
      )
      tokens = pad_sent.map(
          lambda pad: sentence.sentence_tokenizer(pad)
      )
      train_tokens, val_tokens = tokens.randomSplit(
          [training_split, 1 - training_split]
      )
      train_tokens.cache()
      val_tokens.cache()
  
      train_max_len = train_tokens.map(lambda x: len(x)).max()
      print("max length %s" % train_max_len)
  
      words = train_tokens.flatMap(lambda x: x)
      print(
          "%s words and %s sentences processed in train data"
          % (words.count(), train_tokens.count())
      )
  
      val_max_len = val_tokens.map(lambda x: len(x)).max()
      print("val max length %s" % val_max_len)
  
      val_words = val_tokens.flatMap(lambda x: x)
      print(
          "%s words and %s sentences processed in validation data"
          % (val_words.count(), val_tokens.count())
      )
  
      sort_words = (
          words.map(lambda w: (w, 1))
          .reduceByKey(lambda a, b: a + b)
          .sortBy(lambda w_c: w_c[1])
      )
      vocabulary = np.array(sort_words.map(lambda w: w[0]).collect())
  
      fre_len = vocabulary.size
      if vocabsize > fre_len:
          length = fre_len
      else:
          length = vocabsize
      discard_vocab = vocabulary[: fre_len - length]
      used_vocab = vocabulary[fre_len - length : fre_len]
      used_vocab_size = used_vocab.size
      index = np.arange(used_vocab_size)
      index2word = dict(enumerate(used_vocab))
      word2index = dict(zip(used_vocab, index))
      total_vocab_len = used_vocab_size + 1
      startIdx = word2index.get("SENTENCESTART")
      endIdx = word2index.get("SENTENCEEND")
  
      def text2labeled(sent):
          indexes = [word2index.get(x, used_vocab_size) for x in sent]
          data = indexes[0:-1]
          label = indexes[1 : len(indexes)]
          return data, label
  
      def labeled2onehotformat(labeled_sent):
          label = [x + 1 for x in labeled_sent[1]]
          size = len(labeled_sent[0])
          feature_onehot = np.zeros(
              size * total_vocab_len, dtype="int"
          ).reshape([size, total_vocab_len])
          for i, el in enumerate(labeled_sent[0]):
              feature_onehot[i, el] = 1
          return feature_onehot, label
  
      def padding(features, label, length):
          pad_len = length - len(label)
          padded_label = (label + [startIdx] * length)[:length]
          feature_padding = np.zeros(
              (pad_len, total_vocab_len), dtype=np.int32
          )
          feature_padding[:, endIdx + 1] = np.ones(pad_len)
          padded_feautres = np.concatenate(
              (features, feature_padding), axis=0
          )
          return padded_feautres, padded_label
  
      sample_rdd = (
          train_tokens.map(
              lambda sentence_te: text2labeled(sentence_te)
          )
          .map(lambda labeled_sent: labeled2onehotformat(labeled_sent))
          .map(lambda x: padding(x[0], x[1], train_max_len))
          .map(
              lambda vectors_label: Sample.from_ndarray(
                  vectors_label[0], np.array(vectors_label[1])
              )
          )
          .cache()
      )
  
      val_sample_rdd = (
          val_tokens.map(lambda sentence_t: text2labeled(sentence_t))
          .map(lambda labeled_sent: labeled2onehotformat(labeled_sent))
          .map(lambda x: padding(x[0], x[1], val_max_len))
          .map(
              lambda vectors_label: Sample.from_ndarray(
                  vectors_label[0], np.array(vectors_label[1])
              )
          )
          .cache()
      )
  
      return sample_rdd, val_sample_rdd, total_vocab_len
  
file_path: BigDL/python/dllib/src/bigdl/dllib/models/rnn/rnnexample.py
imports: import itertools
  import re
  from optparse import OptionParser
  from bigdl.dllib.feature.dataset import base
  from bigdl.dllib.feature.dataset import sentence
  from bigdl.dllib.nn.layer import *
  from bigdl.dllib.nn.criterion import *
  from bigdl.dllib.optim.optimizer import *
  from bigdl.dllib.utils.common import *
  from bigdl.dllib.utils.common import Sample
label_desc: Use image node_id 2 for calling a global function with example usage: prepare_data(sc, folder, vocabsize, training_split) and returns: sample_rdd,  val_sample_rdd,  total_vocab_len, data,  label, feature_onehot,  label, padded_feautres,  padded_label