@Override
    protected void _predict(Dataframe newData) {
        //This method uses similar approach to the training but the most important
        //difference is that we do not wish to modify the original training params.
        //as a result we need to modify the code to use additional temporary
        //counts for the testing data and merge them with the parameters from the
        //training data in order to make a decision
        ModelParameters modelParameters = knowledgeBase.getModelParameters();
        TrainingParameters trainingParameters = knowledgeBase.getTrainingParameters();

        //get model parameters
        int d = modelParameters.getD();
        int k = trainingParameters.getK(); //number of topics
        
        
        Map<List<Object>, Integer> topicWordCounts = modelParameters.getTopicWordCounts();
        Map<Integer, Integer> topicCounts = modelParameters.getTopicCounts();
        
        
        StorageEngine storageEngine = knowledgeBase.getStorageEngine();
        
        //we create temporary maps for the prediction sets to avoid modifing the maps that we already learned
        Map<List<Object>, Integer> tmp_topicAssignmentOfDocumentWord = storageEngine.getBigMap("tmp_topicAssignmentOfDocumentWord", (Class<List<Object>>)(Class<?>)List.class, Integer.class, MapType.HASHMAP, StorageHint.IN_CACHE, false, true);
        Map<List<Integer>, Integer> tmp_documentTopicCounts = storageEngine.getBigMap("tmp_documentTopicCounts", (Class<List<Integer>>)(Class<?>)List.class, Integer.class, MapType.HASHMAP, StorageHint.IN_MEMORY, false, true);
        Map<List<Object>, Integer> tmp_topicWordCounts = storageEngine.getBigMap("tmp_topicWordCounts", (Class<List<Object>>)(Class<?>)List.class, Integer.class, MapType.HASHMAP, StorageHint.IN_CACHE, false, true);
        Map<Integer, Integer> tmp_topicCounts = storageEngine.getBigMap("tmp_topicCounts", Integer.class, Integer.class, MapType.HASHMAP, StorageHint.IN_MEMORY, false, true);
        
        //initialize topic assignments of each word randomly and update the counters
        for(Map.Entry<Integer, Record> e : newData.entries()) {
            Integer rId = e.getKey();
            Record r = e.getValue();
            Integer documentId = rId;
            
            for(Map.Entry<Object, Object> entry : r.getX().entrySet()) {
                Object wordPosition = entry.getKey();
                Object word = entry.getValue();

                //sample a topic
                Integer topic = PHPMethods.mt_rand(0,k-1);
                
                increase(tmp_topicCounts, topic);
                tmp_topicAssignmentOfDocumentWord.put(Arrays.asList(documentId, wordPosition), topic);
                increase(tmp_documentTopicCounts, Arrays.asList(documentId, topic));
                increase(tmp_topicWordCounts, Arrays.asList(topic, word));
            }
        }
        
        
        double alpha = trainingParameters.getAlpha();
        double beta = trainingParameters.getBeta();
        
        int maxIterations = trainingParameters.getMaxIterations();

        for(int iteration=0;iteration<maxIterations;++iteration) {
            
            logger.debug("Iteration {}", iteration);
            
            
            //collapsed gibbs sampler
            int changedCounter = 0;
            double perplexity = 0.0;
            double totalDatasetWords = 0.0;
            for(Map.Entry<Integer, Record> e : newData.entries()) {
                Integer rId = e.getKey();
                Record r = e.getValue();
                Integer documentId = rId;
                
                
                AssociativeArray topicAssignments = new AssociativeArray();
                for(int j=0;j<k;++j) {
                    topicAssignments.put(j, 0.0);
                }
                
                int totalDocumentWords = r.getX().size();
                totalDatasetWords+=totalDocumentWords;
                for(Map.Entry<Object, Object> entry : r.getX().entrySet()) {
                    Object wordPosition = entry.getKey();
                    Object word = entry.getValue();
            
                    
                    //remove the word from the dataset
                    Integer topic = tmp_topicAssignmentOfDocumentWord.get(Arrays.asList(documentId, wordPosition));
                    decrease(tmp_topicCounts, topic);
                    decrease(tmp_documentTopicCounts, Arrays.asList(documentId, topic));
                    decrease(tmp_topicWordCounts, Arrays.asList(topic, word));
                        
                    int numberOfDocumentWords = r.getX().size()-1;
                    
                    //compute the posteriors of the topics and sample from it
                    AssociativeArray topicProbabilities = new AssociativeArray();
                    for(int j=0;j<k;++j) {
                        //get the counts from the current testing data
                        List<Object> topicWordKey = Arrays.asList(j,word);
                        Integer njw = tmp_topicWordCounts.get(topicWordKey);
                        
                        double enumerator;
                        if(njw !=null) {
                            enumerator = njw + beta;
                        }
                        else {
                            enumerator = beta;
                        }
                        
                        //get also the counts from the training data
                        Integer njw_original = topicWordCounts.get(topicWordKey);
                        if(njw_original!=null) {
                            enumerator+=njw_original;
                        }
                        
                        Integer njd = tmp_documentTopicCounts.get(Arrays.asList(documentId, j));
                        if(njd != null) {
                            enumerator *= (njd + alpha);
                        }
                        else {
                            enumerator *= alpha;
                        }
                        
                        //add the counts from testing data
                        double denominator = tmp_topicCounts.get((Integer)j)+beta*d -1;
                        //and the ones from training data
                        denominator+=topicCounts.get((Integer)j);
                        denominator *= numberOfDocumentWords+alpha*k;
                        
                        topicProbabilities.put(j, enumerator/denominator);
                    }
                    
                    perplexity += Math.log(Descriptives.sum(topicProbabilities.toFlatDataCollection()));
                    
                    //normalize probabilities
                    //Descriptives.normalize(topicProbabilities);
                    
                    //sample from these probabilieis
                    Integer newTopic = (Integer)SimpleRandomSampling.weightedSampling(topicProbabilities, 1, true).iterator().next();
                    topic = newTopic; //new topic assignment
                    
                    
                    //add back the word in the dataset
                    tmp_topicAssignmentOfDocumentWord.put(Arrays.asList(documentId, wordPosition), topic);
                    increase(tmp_topicCounts, topic);
                    increase(tmp_documentTopicCounts, Arrays.asList(documentId, topic));
                    increase(tmp_topicWordCounts, Arrays.asList(topic, word));
                    
                    topicAssignments.put(topic, TypeInference.toDouble(topicAssignments.get(topic))+1.0/totalDocumentWords);
                }
                
                Object mainTopic=MapMethods.selectMaxKeyValue(topicAssignments).getKey();
                
                if(!mainTopic.equals(r.getYPredicted())) {
                    ++changedCounter;
                }                
                newData._unsafe_set(rId, new Record(r.getX(), r.getY(), mainTopic, topicAssignments));
            }

            perplexity=Math.exp(-perplexity/totalDatasetWords);
            
            logger.debug("Reassigned Records {} - Perplexity: {}", changedCounter, perplexity);
            
            if(changedCounter==0) {
                break;
            }            
        }
        
        //Drop the temporary Collection
        storageEngine.dropBigMap("tmp_topicAssignmentOfDocumentWord", tmp_topicAssignmentOfDocumentWord);
        storageEngine.dropBigMap("tmp_documentTopicCounts", tmp_documentTopicCounts);
        storageEngine.dropBigMap("tmp_topicWordCounts", tmp_topicWordCounts);
        storageEngine.dropBigMap("tmp_topicCounts", tmp_topicCounts);
    }
@ Override protected void _predict ( Dataframe newData ) { ModelParameters modelParameters = knowledgeBase . getModelParameters ( ) ; TrainingParameters trainingParameters = knowledgeBase . getTrainingParameters ( ) ; int d = modelParameters . getD ( ) ; int k = trainingParameters . getK ( ) ; Map < List < Object > , Integer > topicWordCounts = modelParameters . getTopicWordCounts ( ) ; Map < Integer , Integer > topicCounts = modelParameters . getTopicCounts ( ) ; StorageEngine storageEngine = knowledgeBase . getStorageEngine ( ) ; Map < List < Object > , Integer > tmp_topicAssignmentOfDocumentWord = storageEngine . getBigMap ( "tmp_topicAssignmentOfDocumentWord" , ( Class < List < Object > > ) ( Class < ? > ) List . class , Integer . class , MapType . HASHMAP , StorageHint . IN_CACHE , false , true ) ; Map < List < Integer > , Integer > tmp_documentTopicCounts = storageEngine . getBigMap ( "tmp_documentTopicCounts" , ( Class < List < Integer > > ) ( Class < ? > ) List . class , Integer . class , MapType . HASHMAP , StorageHint . IN_MEMORY , false , true ) ; Map < List < Object > , Integer > tmp_topicWordCounts = storageEngine . getBigMap ( "tmp_topicWordCounts" , ( Class < List < Object > > ) ( Class < ? > ) List . class , Integer . class , MapType . HASHMAP , StorageHint . IN_CACHE , false , true ) ; Map < Integer , Integer > tmp_topicCounts = storageEngine . getBigMap ( "tmp_topicCounts" , Integer . class , Integer . class , MapType . HASHMAP , StorageHint . IN_MEMORY , false , true ) ; for ( Map . Entry < Integer , Record > e : newData . entries ( ) ) { Integer rId = e . getKey ( ) ; Record r = e . getValue ( ) ; Integer documentId = rId ; for ( Map . Entry < Object , Object > entry : r . getX ( ) . entrySet ( ) ) { Object wordPosition = entry . getKey ( ) ; Object word = entry . getValue ( ) ; Integer topic = PHPMethods . mt_rand ( 0 , k - 1 ) ; increase ( tmp_topicCounts , topic ) ; tmp_topicAssignmentOfDocumentWord . put ( Arrays . asList ( documentId , wordPosition ) , topic ) ; increase ( tmp_documentTopicCounts , Arrays . asList ( documentId , topic ) ) ; increase ( tmp_topicWordCounts , Arrays . asList ( topic , word ) ) ; } } double alpha = trainingParameters . getAlpha ( ) ; double beta = trainingParameters . getBeta ( ) ; int maxIterations = trainingParameters . getMaxIterations ( ) ; for ( int iteration = 0 ; iteration < maxIterations ; ++ iteration ) { logger . debug ( "Iteration {}" , iteration ) ; int changedCounter = 0 ; double perplexity = 0.0 ; double totalDatasetWords = 0.0 ; for ( Map . Entry < Integer , Record > e : newData . entries ( ) ) { Integer rId = e . getKey ( ) ; Record r = e . getValue ( ) ; Integer documentId = rId ; AssociativeArray topicAssignments = new AssociativeArray ( ) ; for ( int j = 0 ; j < k ; ++ j ) { topicAssignments . put ( j , 0.0 ) ; } int totalDocumentWords = r . getX ( ) . size ( ) ; totalDatasetWords += totalDocumentWords ; for ( Map . Entry < Object , Object > entry : r . getX ( ) . entrySet ( ) ) { Object wordPosition = entry . getKey ( ) ; Object word = entry . getValue ( ) ; Integer topic = tmp_topicAssignmentOfDocumentWord . get ( Arrays . asList ( documentId , wordPosition ) ) ; decrease ( tmp_topicCounts , topic ) ; decrease ( tmp_documentTopicCounts , Arrays . asList ( documentId , topic ) ) ; decrease ( tmp_topicWordCounts , Arrays . asList ( topic , word ) ) ; int numberOfDocumentWords = r . getX ( ) . size ( ) - 1 ; AssociativeArray topicProbabilities = new AssociativeArray ( ) ; for ( int j = 0 ; j < k ; ++ j ) { List < Object > topicWordKey = Arrays . asList ( j , word ) ; Integer njw = tmp_topicWordCounts . get ( topicWordKey ) ; double enumerator ; if ( njw != null ) { enumerator = njw + beta ; } else { enumerator = beta ; } Integer njw_original = topicWordCounts . get ( topicWordKey ) ; if ( njw_original != null ) { enumerator += njw_original ; } Integer njd = tmp_documentTopicCounts . get ( Arrays . asList ( documentId , j ) ) ; if ( njd != null ) { enumerator *= ( njd + alpha ) ; } else { enumerator *= alpha ; } double denominator = tmp_topicCounts . get ( ( Integer ) j ) + beta * d - 1 ; denominator += topicCounts . get ( ( Integer ) j ) ; denominator *= numberOfDocumentWords + alpha * k ; topicProbabilities . put ( j , enumerator / denominator ) ; } perplexity += Math . log ( Descriptives . sum ( topicProbabilities . toFlatDataCollection ( ) ) ) ; Integer newTopic = ( Integer ) SimpleRandomSampling . weightedSampling ( topicProbabilities , 1 , true ) . iterator ( ) . next ( ) ; topic = newTopic ; tmp_topicAssignmentOfDocumentWord . put ( Arrays . asList ( documentId , wordPosition ) , topic ) ; increase ( tmp_topicCounts , topic ) ; increase ( tmp_documentTopicCounts , Arrays . asList ( documentId , topic ) ) ; increase ( tmp_topicWordCounts , Arrays . asList ( topic , word ) ) ; topicAssignments . put ( topic , TypeInference . toDouble ( topicAssignments . get ( topic ) ) + 1.0 / totalDocumentWords ) ; } Object mainTopic = MapMethods . selectMaxKeyValue ( topicAssignments ) . getKey ( ) ; if ( ! mainTopic . equals ( r . getYPredicted ( ) ) ) { ++ changedCounter ; } newData . _unsafe_set ( rId , new Record ( r . getX ( ) , r . getY ( ) , mainTopic , topicAssignments ) ) ; } perplexity = Math . exp ( - perplexity / totalDatasetWords ) ; logger . debug ( "Reassigned Records {} - Perplexity: {}" , changedCounter , perplexity ) ; if ( changedCounter == 0 ) { break ; } } storageEngine . dropBigMap ( "tmp_topicAssignmentOfDocumentWord" , tmp_topicAssignmentOfDocumentWord ) ; storageEngine . dropBigMap ( "tmp_documentTopicCounts" , tmp_documentTopicCounts ) ; storageEngine . dropBigMap ( "tmp_topicWordCounts" , tmp_topicWordCounts ) ; storageEngine . dropBigMap ( "tmp_topicCounts" , tmp_topicCounts ) ; }
{@inheritDoc}