docs: public Pair<VocabCache<VocabWord>, GloveWeightLookupTable> train(JavaRDD<String> rdd) throws Exception {
          // Each `train()` can use different parameters
          final JavaSparkContext sc = new JavaSparkContext(rdd.context());
          final SparkConf conf = sc.getConf();
          final int vectorLength = assignVar(VECTOR_LENGTH, conf, Integer.class);
          final boolean useAdaGrad = assignVar(ADAGRAD, conf, Boolean.class);
          final double negative = assignVar(NEGATIVE, conf, Double.class);
          final int numWords = assignVar(NUM_WORDS, conf, Integer.class);
          final int window = assignVar(WINDOW, conf, Integer.class);
          final double alpha = assignVar(ALPHA, conf, Double.class);
          final double minAlpha = assignVar(MIN_ALPHA, conf, Double.class);
          final int iterations = assignVar(ITERATIONS, conf, Integer.class);
          final int nGrams = assignVar(N_GRAMS, conf, Integer.class);
          final String tokenizer = assignVar(TOKENIZER, conf, String.class);
          final String tokenPreprocessor = assignVar(TOKEN_PREPROCESSOR, conf, String.class);
          final boolean removeStop = assignVar(REMOVE_STOPWORDS, conf, Boolean.class);
  
          Map<String, Object> tokenizerVarMap = new HashMap<String, Object>() {
              {
                  put("numWords", numWords);
                  put("nGrams", nGrams);
                  put("tokenizer", tokenizer);
                  put("tokenPreprocessor", tokenPreprocessor);
                  put("removeStop", removeStop);
              }
          };
          Broadcast<Map<String, Object>> broadcastTokenizerVarMap = sc.broadcast(tokenizerVarMap);
  
  
          TextPipeline pipeline = new TextPipeline(rdd, broadcastTokenizerVarMap);
          pipeline.buildVocabCache();
          pipeline.buildVocabWordListRDD();
  
  
          // Get total word count
          Long totalWordCount = pipeline.getTotalWordCount();
          VocabCache<VocabWord> vocabCache = pipeline.getVocabCache();
          JavaRDD<Pair<List<String>, AtomicLong>> sentenceWordsCountRDD = pipeline.getSentenceWordsCountRDD();
          final Pair<VocabCache<VocabWord>, Long> vocabAndNumWords = new Pair<>(vocabCache, totalWordCount);
  
          vocabCacheBroadcast = sc.broadcast(vocabAndNumWords.getFirst());
  
          final GloveWeightLookupTable gloveWeightLookupTable = new GloveWeightLookupTable.Builder()
                          .cache(vocabAndNumWords.getFirst()).lr(conf.getDouble(GlovePerformer.ALPHA, 0.01))
                          .maxCount(conf.getDouble(GlovePerformer.MAX_COUNT, 100))
                          .vectorLength(conf.getInt(GlovePerformer.VECTOR_LENGTH, 300))
                          .xMax(conf.getDouble(GlovePerformer.X_MAX, 0.75)).build();
          gloveWeightLookupTable.resetWeights();
  
          gloveWeightLookupTable.getBiasAdaGrad().historicalGradient = Nd4j.ones(gloveWeightLookupTable.getSyn0().rows());
          gloveWeightLookupTable.getWeightAdaGrad().historicalGradient =
                          Nd4j.ones(gloveWeightLookupTable.getSyn0().shape());
  
  
          log.info("Created lookup table of size " + Arrays.toString(gloveWeightLookupTable.getSyn0().shape()));
          CounterMap<String, String> coOccurrenceCounts = sentenceWordsCountRDD
                          .map(new CoOccurrenceCalculator(symmetric, vocabCacheBroadcast, windowSize))
                          .fold(new CounterMap<String, String>(), new CoOccurrenceCounts());
          Iterator<Pair<String, String>> pair2 = coOccurrenceCounts.getIterator();
          List<Triple<String, String, Float>> counts = new ArrayList<>();
  
          while (pair2.hasNext()) {
              Pair<String, String> next = pair2.next();
              if (coOccurrenceCounts.getCount(next.getFirst(), next.getSecond()) > gloveWeightLookupTable.getMaxCount()) {
                  coOccurrenceCounts.setCount(next.getFirst(), next.getSecond(),
                                  (float) gloveWeightLookupTable.getMaxCount());
              }
              counts.add(new Triple<>(next.getFirst(), next.getSecond(),
                      (float) coOccurrenceCounts.getCount(next.getFirst(), next.getSecond())));
  
          }
  
          log.info("Calculated co occurrences");
  
          JavaRDD<Triple<String, String, Float>> parallel = sc.parallelize(counts);
          JavaPairRDD<String, Tuple2<String, Float>> pairs = parallel
                          .mapToPair(new PairFunction<Triple<String, String, Float>, String, Tuple2<String, Float>>() {
                              @Override
                              public Tuple2<String, Tuple2<String, Float>> call(
                                              Triple<String, String, Float> stringStringDoubleTriple) throws Exception {
                                  return new Tuple2<>(stringStringDoubleTriple.getFirst(),
                                                  new Tuple2<>(stringStringDoubleTriple.getSecond(),
                                                                  stringStringDoubleTriple.getThird()));
                              }
                          });
  
          JavaPairRDD<VocabWord, Tuple2<VocabWord, Float>> pairsVocab = pairs.mapToPair(
                          new PairFunction<Tuple2<String, Tuple2<String, Float>>, VocabWord, Tuple2<VocabWord, Float>>() {
                              @Override
                              public Tuple2<VocabWord, Tuple2<VocabWord, Float>> call(
                                              Tuple2<String, Tuple2<String, Float>> stringTuple2Tuple2) throws Exception {
                                  VocabWord w1 = vocabCacheBroadcast.getValue().wordFor(stringTuple2Tuple2._1());
                                  VocabWord w2 = vocabCacheBroadcast.getValue().wordFor(stringTuple2Tuple2._2()._1());
                                  return new Tuple2<>(w1, new Tuple2<>(w2, stringTuple2Tuple2._2()._2()));
                              }
                          });
  
  
          for (int i = 0; i < iterations; i++) {
              JavaRDD<GloveChange> change =
                              pairsVocab.map(new Function<Tuple2<VocabWord, Tuple2<VocabWord, Float>>, GloveChange>() {
                                  @Override
                                  public GloveChange call(
                                                  Tuple2<VocabWord, Tuple2<VocabWord, Float>> vocabWordTuple2Tuple2)
                                                  throws Exception {
                                      VocabWord w1 = vocabWordTuple2Tuple2._1();
                                      VocabWord w2 = vocabWordTuple2Tuple2._2()._1();
                                      INDArray w1Vector = gloveWeightLookupTable.getSyn0().slice(w1.getIndex());
                                      INDArray w2Vector = gloveWeightLookupTable.getSyn0().slice(w2.getIndex());
                                      INDArray bias = gloveWeightLookupTable.getBias();
                                      double score = vocabWordTuple2Tuple2._2()._2();
                                      double xMax = gloveWeightLookupTable.getxMax();
                                      double maxCount = gloveWeightLookupTable.getMaxCount();
                                      //w1 * w2 + bias
                                      double prediction = Nd4j.getBlasWrapper().dot(w1Vector, w2Vector);
                                      prediction += bias.getDouble(w1.getIndex()) + bias.getDouble(w2.getIndex());
  
                                      double weight = FastMath.pow(Math.min(1.0, (score / maxCount)), xMax);
  
                                      double fDiff = score > xMax ? prediction : weight * (prediction - Math.log(score));
                                      if (Double.isNaN(fDiff))
                                          fDiff = Nd4j.EPS_THRESHOLD;
                                      //amount of change
                                      double gradient = fDiff;
  
                                      Pair<INDArray, Float> w1Update = update(gloveWeightLookupTable.getWeightAdaGrad(),
                                                      gloveWeightLookupTable.getBiasAdaGrad(),
                                                      gloveWeightLookupTable.getSyn0(), gloveWeightLookupTable.getBias(),
                                                      w1, w1Vector, w2Vector, gradient);
                                      Pair<INDArray, Float> w2Update = update(gloveWeightLookupTable.getWeightAdaGrad(),
                                                      gloveWeightLookupTable.getBiasAdaGrad(),
                                                      gloveWeightLookupTable.getSyn0(), gloveWeightLookupTable.getBias(),
                                                      w2, w2Vector, w1Vector, gradient);
                                      return new GloveChange(w1, w2, w1Update.getFirst(), w2Update.getFirst(),
                                                      w1Update.getSecond(), w2Update.getSecond(), fDiff,
                                                      gloveWeightLookupTable.getWeightAdaGrad().getHistoricalGradient()
                                                                      .slice(w1.getIndex()),
                                                      gloveWeightLookupTable.getWeightAdaGrad().getHistoricalGradient()
                                                                      .slice(w2.getIndex()),
                                                      gloveWeightLookupTable.getBiasAdaGrad().getHistoricalGradient()
                                                                      .getDouble(w2.getIndex()),
                                                      gloveWeightLookupTable.getBiasAdaGrad().getHistoricalGradient()
                                                                      .getDouble(w1.getIndex()));
  
                                  }
                              });
  
  
  
              List<GloveChange> gloveChanges = change.collect();
              double error = 0.0;
              for (GloveChange change2 : gloveChanges) {
                  change2.apply(gloveWeightLookupTable);
                  error += change2.getError();
              }
  
  
              List l = pairsVocab.collect();
              Collections.shuffle(l);
              pairsVocab = sc.parallelizePairs(l);
  
              log.info("Error at iteration " + i + " was " + error);
  
  
  
          }
  
          return new Pair<>(vocabAndNumWords.getFirst(), gloveWeightLookupTable);
      }