@Override
    protected void _fit(Dataframe trainingData) {
        ModelParameters modelParameters = knowledgeBase.getModelParameters();
        int n = trainingData.size();
        int d = trainingData.xColumnSize();
        
        knowledgeBase.getTrainingParameters().setMultiProbabilityWeighted(false);
        
        
        Map<List<Object>, Double> likelihoods = modelParameters.getLogLikelihoods();
        Map<Object, Double> logPriors = modelParameters.getLogPriors();
        Set<Object> classesSet = modelParameters.getClasses();
        Map<Object, Double> sumOfLog1minusProb = modelParameters.getSumOfLog1minusProb();
        
        
        //calculate first statistics about the classes
        Map<Object, Integer> totalFeatureOccurrencesForEachClass = new HashMap<>();
        for(Record r : trainingData) {
            Object theClass=r.getY();
            
            if(classesSet.add(theClass)) { //is it new class? add it
                logPriors.put(theClass, 1.0);  
                totalFeatureOccurrencesForEachClass.put(theClass, 0);
                sumOfLog1minusProb.put(theClass, 0.0);
            }
            else { //already exists? increase counter
                logPriors.put(theClass,logPriors.get(theClass)+1.0);
            }
        }
        
        //Loop through all the classes to ensure that the feature-class combination is initialized for ALL the classes
        //The math REQUIRE us to have scores for all classes to make the probabilities comparable.
        /*
            Implementation note:
            The code below uses the metadata from the Dataframe to avoid looping through all the data. 
            This means that if the metadata are stale (contain more columns than the actual data due to 
            updates/removes) we will initialize more parameters here. Nevertheless this should not have 
            any effects on the results of the algorithm since the scores will be the same in all classes
            and it will be taken care by the normalization.
        */
        streamExecutor.forEach(StreamMethods.stream(trainingData.getXDataTypes().keySet().stream(), isParallelized()), feature -> {
            for(Object theClass : classesSet) {
                List<Object> featureClassTuple = Arrays.asList(feature, theClass);
                likelihoods.put(featureClassTuple, 0.0); //the key unique across threads and the map is concurrent
            }
        });
        
        
        //now calculate the statistics of features
        streamExecutor.forEach(StreamMethods.stream(trainingData.stream(), isParallelized()), r -> {
            Object theClass = r.getY();
            
            //store the occurrances of the features
            int sumOfOccurrences = 0;
            for(Map.Entry<Object, Object> entry : r.getX().entrySet()) {
                Object feature = entry.getKey();
                Double occurrences=TypeInference.toDouble(entry.getValue());
                
                if(occurrences!= null && occurrences>0.0) {
                    //The below block of code clips occurrences to 1
                    List<Object> featureClassTuple = Arrays.asList(feature, theClass);
                    likelihoods.put(featureClassTuple, likelihoods.get(featureClassTuple)+1.0); //each thread updates a unique key and the map is cuncurrent
                    
                    sumOfOccurrences++;
                }
            }
            
            synchronized(totalFeatureOccurrencesForEachClass) {
                totalFeatureOccurrencesForEachClass.put(theClass, totalFeatureOccurrencesForEachClass.get(theClass)+sumOfOccurrences);
            }
            
        });
        
        //calculate prior log probabilities
        for(Map.Entry<Object, Double> entry : logPriors.entrySet()) {
            Object theClass = entry.getKey();
            Double count = entry.getValue();
            
            //updated log priors
            logPriors.put(theClass, Math.log(count/n));
        }
        
        //update log likelihood
        for(Object theClass : classesSet) {
            double sumLog1minusP = streamExecutor.sum(StreamMethods.stream(trainingData.getXDataTypes().keySet().stream(), isParallelized()).mapToDouble(feature -> {
                List<Object> featureClassTuple = Arrays.asList(feature, theClass);
                Double occurrences = likelihoods.get(featureClassTuple);

                //We perform laplace smoothing (also known as add-1)
                Double smoothedProbability = (occurrences+1.0)/(totalFeatureOccurrencesForEachClass.get(theClass)+d); // the d is also known in NLP problems as the Vocabulary size. 

                likelihoods.put(featureClassTuple, smoothedProbability);

                double log1minusP = Math.log( 1.0-smoothedProbability );
                
                return log1minusP;
            }));
            
            sumOfLog1minusProb.put(theClass, sumOfLog1minusProb.get(theClass) + sumLog1minusP); 
        }
        
        //Drop the temporary Collection
    }
@ Override protected void _fit ( Dataframe trainingData ) { ModelParameters modelParameters = knowledgeBase . getModelParameters ( ) ; int n = trainingData . size ( ) ; int d = trainingData . xColumnSize ( ) ; knowledgeBase . getTrainingParameters ( ) . setMultiProbabilityWeighted ( false ) ; Map < List < Object > , Double > likelihoods = modelParameters . getLogLikelihoods ( ) ; Map < Object , Double > logPriors = modelParameters . getLogPriors ( ) ; Set < Object > classesSet = modelParameters . getClasses ( ) ; Map < Object , Double > sumOfLog1minusProb = modelParameters . getSumOfLog1minusProb ( ) ; Map < Object , Integer > totalFeatureOccurrencesForEachClass = new HashMap < > ( ) ; for ( Record r : trainingData ) { Object theClass = r . getY ( ) ; if ( classesSet . add ( theClass ) ) { logPriors . put ( theClass , 1.0 ) ; totalFeatureOccurrencesForEachClass . put ( theClass , 0 ) ; sumOfLog1minusProb . put ( theClass , 0.0 ) ; } else { logPriors . put ( theClass , logPriors . get ( theClass ) + 1.0 ) ; } } streamExecutor . forEach ( StreamMethods . stream ( trainingData . getXDataTypes ( ) . keySet ( ) . stream ( ) , isParallelized ( ) ) , feature -> { for ( Object theClass : classesSet ) { List < Object > featureClassTuple = Arrays . asList ( feature , theClass ) ; likelihoods . put ( featureClassTuple , 0.0 ) ; } } ) ; streamExecutor . forEach ( StreamMethods . stream ( trainingData . stream ( ) , isParallelized ( ) ) , r -> { Object theClass = r . getY ( ) ; int sumOfOccurrences = 0 ; for ( Map . Entry < Object , Object > entry : r . getX ( ) . entrySet ( ) ) { Object feature = entry . getKey ( ) ; Double occurrences = TypeInference . toDouble ( entry . getValue ( ) ) ; if ( occurrences != null && occurrences > 0.0 ) { List < Object > featureClassTuple = Arrays . asList ( feature , theClass ) ; likelihoods . put ( featureClassTuple , likelihoods . get ( featureClassTuple ) + 1.0 ) ; sumOfOccurrences ++ ; } } synchronized ( totalFeatureOccurrencesForEachClass ) { totalFeatureOccurrencesForEachClass . put ( theClass , totalFeatureOccurrencesForEachClass . get ( theClass ) + sumOfOccurrences ) ; } } ) ; for ( Map . Entry < Object , Double > entry : logPriors . entrySet ( ) ) { Object theClass = entry . getKey ( ) ; Double count = entry . getValue ( ) ; logPriors . put ( theClass , Math . log ( count / n ) ) ; } for ( Object theClass : classesSet ) { double sumLog1minusP = streamExecutor . sum ( StreamMethods . stream ( trainingData . getXDataTypes ( ) . keySet ( ) . stream ( ) , isParallelized ( ) ) . mapToDouble ( feature -> { List < Object > featureClassTuple = Arrays . asList ( feature , theClass ) ; Double occurrences = likelihoods . get ( featureClassTuple ) ; Double smoothedProbability = ( occurrences + 1.0 ) / ( totalFeatureOccurrencesForEachClass . get ( theClass ) + d ) ; likelihoods . put ( featureClassTuple , smoothedProbability ) ; double log1minusP = Math . log ( 1.0 - smoothedProbability ) ; return log1minusP ; } ) ) ; sumOfLog1minusProb . put ( theClass , sumOfLog1minusProb . get ( theClass ) + sumLog1minusP ) ; } }
{@inheritDoc}