class OnlineCorpusStep extends ComputeFunction {

    @Override
    public void calc(ComContext context) {
        Tuple2<Long, Integer> tuple2 = ((List<Tuple2<Long, Integer>>) context.getObj(LdaVariable.shape)).get(0);
        int vocabularySize = tuple2.f1;
        List<Vector> data = context.getObj(LdaVariable.data);
        DenseMatrix lambda;
        DenseMatrix alpha;
        //the first iteration
        if (context.getStepNo() == 1) {
            Tuple2<DenseMatrix, DenseMatrix> initGammaAndAlpha = ((List<Tuple2<DenseMatrix, DenseMatrix>>)
                context.getObj(LdaVariable.initModel)).get(0);
            lambda = initGammaAndAlpha.f0;
            alpha = initGammaAndAlpha.f1;
        } else {
            lambda = context.getObj(LdaVariable.lambda);
            alpha = context.getObj(LdaVariable.alpha);
        }
        if (data == null) {
            context.putObj(LdaVariable.wordTopicStat, new double[numTopic * vocabularySize]);
            context.putObj(LdaVariable.logPhatPart, new double[numTopic]);
            context.putObj(LdaVariable.nonEmptyWordCount, new double[]{0});
            context.putObj(LdaVariable.nonEmptyDocCount, new double[]{0});
            return;
        }
        DenseMatrix gammad = null;
        Tuple4<DenseMatrix, DenseMatrix, Long, Long> corpusUpdatedData =
            onlineCorpusUpdate(data, lambda, alpha, gammad,
                vocabularySize, numTopic, subSamplingRate);

        context.putObj(LdaVariable.wordTopicStat, corpusUpdatedData.f0.getData().clone());
        context.putObj(LdaVariable.logPhatPart, corpusUpdatedData.f1.getData().clone());
        context.putObj(LdaVariable.nonEmptyWordCount, new double[]{corpusUpdatedData.f2});
        context.putObj(LdaVariable.nonEmptyDocCount, new double[]{corpusUpdatedData.f3});
    }

    public  OnlineCorpusStep(int numTopic, double subSamplingRate);

}