@Test public void testConcepts() throws Exception { SparkConf sparkConf = new SparkConf().setMaster("local[8]") .set("spark.driver.host", "localhost") .setAppName("sparktest"); JavaSparkContext sc = new JavaSparkContext(sparkConf); String dataPath = new ClassPathResource("big/raw_sentences.txt").getFile().getAbsolutePath(); JavaRDD<String> corpus = sc.textFile(dataPath); TokenizerFactory t = new DefaultTokenizerFactory(); t.setTokenPreProcessor(new CommonPreprocessor()); Word2Vec word2Vec = new Word2Vec.Builder().setNGrams(1) .tokenizerFactory(t).seed(42L).negative(10).useAdaGrad(false).layerSize(150).windowSize(5) .learningRate(0.025).minLearningRate(0.0001).iterations(1).batchSize(100).minWordFrequency(5) .stopWords(Arrays.asList("three")).useUnknown(true).build(); word2Vec.train(corpus); System.out.println("UNK: " + word2Vec.getWordVectorMatrix("UNK")); InMemoryLookupTable<VocabWord> table = (InMemoryLookupTable<VocabWord>) word2Vec.lookupTable(); double sim = word2Vec.similarity("day", "night"); System.out.println("day/night similarity: " + sim); Collection<String> words = word2Vec.wordsNearest("day", 10); printWords("day", words, word2Vec); assertTrue(words.contains("night")); assertTrue(words.contains("week")); assertTrue(words.contains("year")); sim = word2Vec.similarity("two", "four"); System.out.println("two/four similarity: " + sim); words = word2Vec.wordsNearest("two", 10); printWords("two", words, word2Vec); assertFalse(words.contains("three")); assertTrue(words.contains("five")); assertTrue(words.contains("four")); sc.stop(); File tempFile = testDir.newFile("temp" + System.currentTimeMillis() + ".tmp"); int idx1 = word2Vec.vocab().wordFor("day").getIndex(); INDArray array1 = word2Vec.getWordVectorMatrix("day").dup(); VocabWord word1 = word2Vec.vocab().elementAtIndex(0); WordVectorSerializer.writeWordVectors(word2Vec.getLookupTable(), tempFile); WordVectors vectors = WordVectorSerializer.loadTxtVectors(tempFile); VocabWord word2 = ((VocabCache<VocabWord>) vectors.vocab()).elementAtIndex(0); VocabWord wordIT = ((VocabCache<VocabWord>) vectors.vocab()).wordFor("it"); int idx2 = vectors.vocab().wordFor("day").getIndex(); INDArray array2 = vectors.getWordVectorMatrix("day").dup(); System.out.println("word 'i': " + word2); System.out.println("word 'it': " + wordIT); assertEquals(idx1, idx2); assertEquals(word1, word2); assertEquals(array1, array2); }
public void train(JavaRDD<String> corpusRDD) throws Exception { log.info("Start training ..."); if (workers > 0) corpusRDD.repartition(workers); final JavaSparkContext sc = new JavaSparkContext(corpusRDD.context()); Map<String, Object> tokenizerVarMap = getTokenizerVarMap(); Map<String, Object> word2vecVarMap = getWord2vecVarMap(); 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; log.info("Tokenization and building VocabCache ..."); Broadcast<Map<String, Object>> broadcastTokenizerVarMap = sc.broadcast(tokenizerVarMap); TextPipeline pipeline = new TextPipeline(corpusRDD, broadcastTokenizerVarMap); pipeline.buildVocabCache(); pipeline.buildVocabWordListRDD(); word2vecVarMap.put("totalWordCount", pipeline.getTotalWordCount()); sentenceWordsCountRDD = pipeline.getSentenceCountRDD(); vocabWordListRDD = pipeline.getVocabWordListRDD(); Broadcast<VocabCache<VocabWord>> vocabCacheBroadcast = pipeline.getBroadCastVocabCache(); vocabCache = vocabCacheBroadcast.getValue(); log.info("Vocab size: {}", vocabCache.numWords()); log.info("Building Huffman Tree ..."); 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()); List<Pair<VocabWord, INDArray>> syn0UpdateEntries = indexSyn0UpdateEntryRDD.collect(); INDArray syn0 = Nd4j.zeros(vocabCache.numWords(), layerSize); 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()); 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()); } } 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); }
Word2Vec extends WordVectorsImpl<VocabWord> implements Serializable { public void train(JavaRDD<String> corpusRDD) throws Exception { log.info("Start training ..."); if (workers > 0) corpusRDD.repartition(workers); final JavaSparkContext sc = new JavaSparkContext(corpusRDD.context()); Map<String, Object> tokenizerVarMap = getTokenizerVarMap(); Map<String, Object> word2vecVarMap = getWord2vecVarMap(); 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; log.info("Tokenization and building VocabCache ..."); Broadcast<Map<String, Object>> broadcastTokenizerVarMap = sc.broadcast(tokenizerVarMap); TextPipeline pipeline = new TextPipeline(corpusRDD, broadcastTokenizerVarMap); pipeline.buildVocabCache(); pipeline.buildVocabWordListRDD(); word2vecVarMap.put("totalWordCount", pipeline.getTotalWordCount()); sentenceWordsCountRDD = pipeline.getSentenceCountRDD(); vocabWordListRDD = pipeline.getVocabWordListRDD(); Broadcast<VocabCache<VocabWord>> vocabCacheBroadcast = pipeline.getBroadCastVocabCache(); vocabCache = vocabCacheBroadcast.getValue(); log.info("Vocab size: {}", vocabCache.numWords()); log.info("Building Huffman Tree ..."); 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()); List<Pair<VocabWord, INDArray>> syn0UpdateEntries = indexSyn0UpdateEntryRDD.collect(); INDArray syn0 = Nd4j.zeros(vocabCache.numWords(), layerSize); 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()); 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()); } } 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); } }
Word2Vec extends WordVectorsImpl<VocabWord> implements Serializable { public void train(JavaRDD<String> corpusRDD) throws Exception { log.info("Start training ..."); if (workers > 0) corpusRDD.repartition(workers); final JavaSparkContext sc = new JavaSparkContext(corpusRDD.context()); Map<String, Object> tokenizerVarMap = getTokenizerVarMap(); Map<String, Object> word2vecVarMap = getWord2vecVarMap(); 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; log.info("Tokenization and building VocabCache ..."); Broadcast<Map<String, Object>> broadcastTokenizerVarMap = sc.broadcast(tokenizerVarMap); TextPipeline pipeline = new TextPipeline(corpusRDD, broadcastTokenizerVarMap); pipeline.buildVocabCache(); pipeline.buildVocabWordListRDD(); word2vecVarMap.put("totalWordCount", pipeline.getTotalWordCount()); sentenceWordsCountRDD = pipeline.getSentenceCountRDD(); vocabWordListRDD = pipeline.getVocabWordListRDD(); Broadcast<VocabCache<VocabWord>> vocabCacheBroadcast = pipeline.getBroadCastVocabCache(); vocabCache = vocabCacheBroadcast.getValue(); log.info("Vocab size: {}", vocabCache.numWords()); log.info("Building Huffman Tree ..."); 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()); List<Pair<VocabWord, INDArray>> syn0UpdateEntries = indexSyn0UpdateEntryRDD.collect(); INDArray syn0 = Nd4j.zeros(vocabCache.numWords(), layerSize); 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()); 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()); } } 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); } protected  Word2Vec(INDArray trainedSyn1); protected  Word2Vec(); }
Word2Vec extends WordVectorsImpl<VocabWord> implements Serializable { public void train(JavaRDD<String> corpusRDD) throws Exception { log.info("Start training ..."); if (workers > 0) corpusRDD.repartition(workers); final JavaSparkContext sc = new JavaSparkContext(corpusRDD.context()); Map<String, Object> tokenizerVarMap = getTokenizerVarMap(); Map<String, Object> word2vecVarMap = getWord2vecVarMap(); 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; log.info("Tokenization and building VocabCache ..."); Broadcast<Map<String, Object>> broadcastTokenizerVarMap = sc.broadcast(tokenizerVarMap); TextPipeline pipeline = new TextPipeline(corpusRDD, broadcastTokenizerVarMap); pipeline.buildVocabCache(); pipeline.buildVocabWordListRDD(); word2vecVarMap.put("totalWordCount", pipeline.getTotalWordCount()); sentenceWordsCountRDD = pipeline.getSentenceCountRDD(); vocabWordListRDD = pipeline.getVocabWordListRDD(); Broadcast<VocabCache<VocabWord>> vocabCacheBroadcast = pipeline.getBroadCastVocabCache(); vocabCache = vocabCacheBroadcast.getValue(); log.info("Vocab size: {}", vocabCache.numWords()); log.info("Building Huffman Tree ..."); 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()); List<Pair<VocabWord, INDArray>> syn0UpdateEntries = indexSyn0UpdateEntryRDD.collect(); INDArray syn0 = Nd4j.zeros(vocabCache.numWords(), layerSize); 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()); 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()); } } 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); } protected  Word2Vec(INDArray trainedSyn1); protected  Word2Vec(); Map<String, Object> getTokenizerVarMap(); Map<String, Object> getWord2vecVarMap(); void train(JavaRDD<String> corpusRDD); }
Word2Vec extends WordVectorsImpl<VocabWord> implements Serializable { public void train(JavaRDD<String> corpusRDD) throws Exception { log.info("Start training ..."); if (workers > 0) corpusRDD.repartition(workers); final JavaSparkContext sc = new JavaSparkContext(corpusRDD.context()); Map<String, Object> tokenizerVarMap = getTokenizerVarMap(); Map<String, Object> word2vecVarMap = getWord2vecVarMap(); 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; log.info("Tokenization and building VocabCache ..."); Broadcast<Map<String, Object>> broadcastTokenizerVarMap = sc.broadcast(tokenizerVarMap); TextPipeline pipeline = new TextPipeline(corpusRDD, broadcastTokenizerVarMap); pipeline.buildVocabCache(); pipeline.buildVocabWordListRDD(); word2vecVarMap.put("totalWordCount", pipeline.getTotalWordCount()); sentenceWordsCountRDD = pipeline.getSentenceCountRDD(); vocabWordListRDD = pipeline.getVocabWordListRDD(); Broadcast<VocabCache<VocabWord>> vocabCacheBroadcast = pipeline.getBroadCastVocabCache(); vocabCache = vocabCacheBroadcast.getValue(); log.info("Vocab size: {}", vocabCache.numWords()); log.info("Building Huffman Tree ..."); 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()); List<Pair<VocabWord, INDArray>> syn0UpdateEntries = indexSyn0UpdateEntryRDD.collect(); INDArray syn0 = Nd4j.zeros(vocabCache.numWords(), layerSize); 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()); 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()); } } 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); } protected  Word2Vec(INDArray trainedSyn1); protected  Word2Vec(); Map<String, Object> getTokenizerVarMap(); Map<String, Object> getWord2vecVarMap(); void train(JavaRDD<String> corpusRDD);  }