code: def test_distributed_textset_integration(self):
      texts_rdd = self.sc.parallelize(self.texts)
      labels_rdd = self.sc.parallelize(self.labels)
      distributed_set = DistributedTextSet(texts_rdd, labels_rdd)
      assert distributed_set.is_distributed()
      assert not distributed_set.is_local()
      assert distributed_set.get_texts().collect() == self.texts
      assert distributed_set.get_labels().collect() == self.labels
  
      sets = distributed_set.random_split([0.5, 0.5])
      train_texts = sets[0].get_texts().collect()
      test_texts = sets[1].get_texts().collect()
      assert set(train_texts + test_texts) == set(self.texts)
  
      tokenized = Tokenizer()(distributed_set)
      transformed = (
          tokenized.normalize()
          .word2idx()
          .shape_sequence(5)
          .generate_sample()
      )
      word_index = transformed.get_word_index()
      assert len(word_index) == 14
      samples = transformed.get_samples().collect()
      assert len(samples) == 3
      for sample in samples:
          assert sample.feature.shape[0] == 5
  
      vocab_file = create_tmp_path() + ".txt"
      transformed.save_word_index(vocab_file)
      distributed_set2 = DistributedTextSet(texts_rdd, labels_rdd)
      distributed_set2.load_word_index(vocab_file)
      transformed2 = (
          distributed_set2.tokenize()
          .normalize()
          .word2idx()
          .shape_sequence(5)
          .generate_sample()
      )
      samples2 = transformed2.get_samples().collect()
      for s1, s2 in zip(samples, samples2):
          assert np.allclose(
              s1.feature.to_ndarray(), s2.feature.to_ndarray()
          )
      os.remove(vocab_file)
  
      model = TextClassifier(
          5, self.glove_path, word_index, 5, encoder="lstm"
      )
      model.compile(SGD(), SparseCategoricalCrossEntropy())
      model.fit(transformed, batch_size=2, nb_epoch=2)
      res_set = model.predict(transformed, batch_per_thread=2)
      predicts = res_set.get_predicts().collect()
      for predict in predicts:  # (uri, prediction)
          assert not predict[0]  # uri is not recorded and thus None
          assert len(predict[1]) == 1
          assert predict[1][0].shape == (5,)
  
      tmp_path = create_tmp_path() + ".bigdl"
      model.save_model(tmp_path, over_write=True)
      loaded_model = TextClassifier.load_model(tmp_path)
      loaded_res_set = loaded_model.predict(
          transformed, batch_per_thread=2
      )
      loaded_predicts = loaded_res_set.get_predicts().collect()
      assert len(loaded_predicts) == len(predicts)
      os.remove(tmp_path)
  
imports: import pytest
  import shutil
  from bigdl.optim.optimizer import SGD
  from zoo.feature.common import ChainedPreprocessing, Relations
  from zoo.feature.text import *
  from zoo.common.nncontext import *
  from zoo.models.textclassification import TextClassifier
  from zoo.models.textmatching import KNRM
  from zoo.pipeline.api.keras.models import Sequential
  from zoo.pipeline.api.keras.layers import TimeDistributed
  from zoo.pipeline.api.keras.objectives import SparseCategoricalCrossEntropy
label_desc: Use image node_id 6 for calling the TestTextSet obj's underlying member method code with example usage: obj.test_distributed_textset_integration() without return types