docs: public void train(JavaRDD<String> corpusRDD) throws Exception {
          log.info("Start training ...");
  
          if (workers > 0)
              corpusRDD.repartition(workers);
  
          // SparkContext
          final JavaSparkContext sc = new JavaSparkContext(corpusRDD.context());
  
          // Pre-defined variables
          Map<String, Object> tokenizerVarMap = getTokenizerVarMap();
          Map<String, Object> word2vecVarMap = getWord2vecVarMap();
  
          // Variables to fill in train
          final JavaRDD<AtomicLong> sentenceWordsCountRDD;
          final JavaRDD<List<VocabWord>> vocabWordListRDD;
          final JavaPairRDD<List<VocabWord>, Long> vocabWordListSentenceCumSumRDD;
          final VocabCache<VocabWord> vocabCache;
          final JavaRDD<Long> sentenceCumSumCountRDD;
          int maxRep = 1;
  
          // Start Training //
          //////////////////////////////////////
          log.info("Tokenization and building VocabCache ...");
          // Processing every sentence and make a VocabCache which gets fed into a LookupCache
          Broadcast<Map<String, Object>> broadcastTokenizerVarMap = sc.broadcast(tokenizerVarMap);
          TextPipeline pipeline = new TextPipeline(corpusRDD, broadcastTokenizerVarMap);
          pipeline.buildVocabCache();
          pipeline.buildVocabWordListRDD();
  
          // Get total word count and put into word2vec variable map
          word2vecVarMap.put("totalWordCount", pipeline.getTotalWordCount());
  
          // 2 RDDs: (vocab words list) and (sentence Count).Already cached
          sentenceWordsCountRDD = pipeline.getSentenceCountRDD();
          vocabWordListRDD = pipeline.getVocabWordListRDD();
  
          // Get vocabCache and broad-casted vocabCache
          Broadcast<VocabCache<VocabWord>> vocabCacheBroadcast = pipeline.getBroadCastVocabCache();
          vocabCache = vocabCacheBroadcast.getValue();
  
          log.info("Vocab size: {}", vocabCache.numWords());
  
          //////////////////////////////////////
          log.info("Building Huffman Tree ...");
          // Building Huffman Tree would update the code and point in each of the vocabWord in vocabCache
          /*
          We don't need to build tree here, since it was built earlier, at TextPipeline.buildVocabCache() call.
          
          Huffman huffman = new Huffman(vocabCache.vocabWords());
          huffman.build();
          huffman.applyIndexes(vocabCache);
          */
          //////////////////////////////////////
          log.info("Calculating cumulative sum of sentence counts ...");
          sentenceCumSumCountRDD = new CountCumSum(sentenceWordsCountRDD).buildCumSum();
  
          //////////////////////////////////////
          log.info("Mapping to RDD(vocabWordList, cumulative sentence count) ...");
          vocabWordListSentenceCumSumRDD =
                          vocabWordListRDD.zip(sentenceCumSumCountRDD).setName("vocabWordListSentenceCumSumRDD");
  
          /////////////////////////////////////
          log.info("Broadcasting word2vec variables to workers ...");
          Broadcast<Map<String, Object>> word2vecVarMapBroadcast = sc.broadcast(word2vecVarMap);
          Broadcast<double[]> expTableBroadcast = sc.broadcast(expTable);
  
  
  
          /////////////////////////////////////
          log.info("Training word2vec sentences ...");
          FlatMapFunction firstIterFunc =
                          new FirstIterationFunction(word2vecVarMapBroadcast, expTableBroadcast, vocabCacheBroadcast);
          @SuppressWarnings("unchecked")
          JavaRDD<Pair<VocabWord, INDArray>> indexSyn0UpdateEntryRDD =
                          vocabWordListSentenceCumSumRDD.mapPartitions(firstIterFunc).map(new MapToPairFunction());
  
          // Get all the syn0 updates into a list in driver
          List<Pair<VocabWord, INDArray>> syn0UpdateEntries = indexSyn0UpdateEntryRDD.collect();
  
          // Instantiate syn0
          INDArray syn0 = Nd4j.zeros(vocabCache.numWords(), layerSize);
  
          // Updating syn0 first pass: just add vectors obtained from different nodes
          log.info("Averaging results...");
          Map<VocabWord, AtomicInteger> updates = new HashMap<>();
          Map<Long, Long> updaters = new HashMap<>();
          for (Pair<VocabWord, INDArray> syn0UpdateEntry : syn0UpdateEntries) {
              syn0.getRow(syn0UpdateEntry.getFirst().getIndex()).addi(syn0UpdateEntry.getSecond());
  
              // for proper averaging we need to divide resulting sums later, by the number of additions
              if (updates.containsKey(syn0UpdateEntry.getFirst())) {
                  updates.get(syn0UpdateEntry.getFirst()).incrementAndGet();
              } else
                  updates.put(syn0UpdateEntry.getFirst(), new AtomicInteger(1));
  
              if (!updaters.containsKey(syn0UpdateEntry.getFirst().getVocabId())) {
                  updaters.put(syn0UpdateEntry.getFirst().getVocabId(), syn0UpdateEntry.getFirst().getAffinityId());
              }
          }
  
          // Updating syn0 second pass: average obtained vectors
          for (Map.Entry<VocabWord, AtomicInteger> entry : updates.entrySet()) {
              if (entry.getValue().get() > 1) {
                  if (entry.getValue().get() > maxRep)
                      maxRep = entry.getValue().get();
                  syn0.getRow(entry.getKey().getIndex()).divi(entry.getValue().get());
              }
          }
  
          long totals = 0;
  
          log.info("Finished calculations...");
  
  
          vocab = vocabCache;
          InMemoryLookupTable<VocabWord> inMemoryLookupTable = new InMemoryLookupTable<VocabWord>();
          Environment env = EnvironmentUtils.buildEnvironment();
          env.setNumCores(maxRep);
          env.setAvailableMemory(totals);
          update(env, Event.SPARK);
          inMemoryLookupTable.setVocab(vocabCache);
          inMemoryLookupTable.setVectorLength(layerSize);
          inMemoryLookupTable.setSyn0(syn0);
          lookupTable = inMemoryLookupTable;
          modelUtils.init(lookupTable);
      }