<commit_before>package edu.uci.eecs.spectralLDA.datamoments

/**
 * Data Cumulants Calculation.
 * Created by <NAME> on 11/2/15.
 */

import edu.uci.eecs.spectralLDA.utils.AlgebraUtil
import breeze.linalg._
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.SparkContext
import scala.collection.mutable

import scalaxy.loops._
import scala.language.postfixOps

case class DataCumulant(thirdOrderMoments: DenseMatrix[Double], unwhiteningMatrix: DenseMatrix[Double])
  extends Serializable


object DataCumulant {
  def getDataCumulant(dimK: Int,
                      alpha0: Double,
                      tolerance: Double,
                      documents: RDD[(Long, Double, SparseVector[Double])])
        : DataCumulant = {
    val sc: SparkContext = documents.sparkContext
    val dimVocab = documents.take(1)(0)._3.length
    val numDocs = documents.count()

    println("Start calculating first order moments...")
    val M1: DenseVector[Double] = documents map {
      case (_, length, vec) => update_firstOrderMoments(dimVocab, vec.toDenseVector, length)
    } reduce (_ + _)

    val firstOrderMoments: DenseVector[Double] = M1 / numDocs.toDouble
    println("Finished calculating first order moments.")

    val (thirdOrderMoments: DenseMatrix[Double], unwhiteningMatrix: DenseMatrix[Double]) = computeThirdOrderMoments(
      sc, alpha0, dimVocab, dimK,
      numDocs, firstOrderMoments, documents,
      tolerance
    )

    new DataCumulant(thirdOrderMoments, unwhiteningMatrix)
  }

  private def computeThirdOrderMoments(sc: SparkContext,
                                       alpha0: Double,
                                       dimVocab: Int, dimK: Int,
                                       numDocs: Long,
                                       firstOrderMoments: DenseVector[Double],
                                       documents: RDD[(Long, Double, SparseVector[Double])],
                                       tolerance: Double)
  : (DenseMatrix[Double], DenseMatrix[Double]) = {
    println("Start calculating second order moments...")
    val (eigenVectors: DenseMatrix[Double], eigenValues: DenseVector[Double]) = whiten(sc, alpha0,
      dimVocab, dimK, numDocs, firstOrderMoments, documents)
    println("Finished calculating second order moments and whitening matrix.")

    println("Start whitening data with dimensionality reduction...")
    val whitenedData: RDD[(DenseVector[Double], Double)] = documents.map {
      case (_, length, vec) => (project(dimVocab, dimK, alpha0, eigenValues, eigenVectors, vec)(tolerance), length)
    }
    val firstOrderMoments_whitened: DenseVector[Double] = whitenedData
      .map(x => x._1 / x._2)
      .reduce((a, b) => a :+ b)
      .map(x => x / numDocs.toDouble)
    println("Finished whitening data.")

    println("Start calculating third order moments...")
    val Ta: DenseMatrix[Double] = whitenedData map {
      case (vec, len) => update_thirdOrderMoments(
        dimK, alpha0,
        firstOrderMoments_whitened,
        vec, len)
    } reduce(_ + _)

    val alpha0sq: Double = alpha0 * alpha0
    val Ta_shift = DenseMatrix.zeros[Double](dimK, dimK * dimK)
    for (id_i <- 0 until dimK optimized) {
      for (id_j <- 0 until dimK optimized) {
        for (id_l <- 0 until dimK optimized) {
          Ta_shift(id_i, id_j * dimK + id_l) += (alpha0sq * firstOrderMoments_whitened(id_i)
              * firstOrderMoments_whitened(id_j) * firstOrderMoments_whitened(id_l))
        }
      }
    }
    println("Finished calculating third order moments.")
    val unwhiteningMatrix: breeze.linalg.DenseMatrix[Double] = eigenVectors * breeze.linalg.diag(eigenValues.map(x => scala.math.sqrt(x)))
    (Ta / numDocs.toDouble + Ta_shift, unwhiteningMatrix)
  }

  private def whiten(sc: SparkContext,
                     alpha0: Double,
                     vocabSize: Int, dimK: Int,
                     numDocs: Long,
                     firstOrderMoments: DenseVector[Double],
                     documents: RDD[(Long, Double, SparseVector[Double])])
  : (DenseMatrix[Double], DenseVector[Double]) = {
    val para_main: Double = (alpha0 + 1.0) / numDocs.toDouble
    val para_shift: Double = alpha0

    val SEED_random: Long = System.currentTimeMillis
    val gaussianRandomMatrix: DenseMatrix[Double] = AlgebraUtil.gaussian(vocabSize, dimK * 2, SEED_random)
    val gaussianRandomMatrix_broadcasted: Broadcast[breeze.linalg.DenseMatrix[Double]] = sc.broadcast(gaussianRandomMatrix)
    val firstOrderMoments_broadcasted: Broadcast[breeze.linalg.DenseVector[Double]] = sc.broadcast(firstOrderMoments.toDenseVector)<commit_msg>put computing and whitening M1, M2, M3 in one function<commit_after>package edu.uci.eecs.spectralLDA.datamoments

/**
 * Data Cumulants Calculation.
 * Created by <NAME> on 11/2/15.
 */

import edu.uci.eecs.spectralLDA.utils.AlgebraUtil
import breeze.linalg._
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.SparkContext
import scala.collection.mutable

import scalaxy.loops._
import scala.language.postfixOps

case class DataCumulant(thirdOrderMoments: DenseMatrix[Double], unwhiteningMatrix: DenseMatrix[Double])
  extends Serializable


object DataCumulant {
  def getDataCumulant(dimK: Int,
                      alpha0: Double,
                      tolerance: Double,
                      documents: RDD[(Long, Double, SparseVector[Double])])
        : DataCumulant = {
    val sc: SparkContext = documents.sparkContext
    val dimVocab = documents.take(1)(0)._3.length
    val numDocs = documents.count()

    println("Start calculating first order moments...")
    val M1: DenseVector[Double] = documents map {
      case (_, length, vec) => update_firstOrderMoments(dimVocab, vec.toDenseVector, length)
    } reduce (_ + _)

    val firstOrderMoments: DenseVector[Double] = M1 / numDocs.toDouble
    println("Finished calculating first order moments.")

    println("Start calculating second order moments...")
    val (eigenVectors: DenseMatrix[Double], eigenValues: DenseVector[Double]) = whiten(sc, alpha0,
      dimVocab, dimK, numDocs, firstOrderMoments, documents)
    println("Finished calculating second order moments and whitening matrix.")

    println("Start whitening data with dimensionality reduction...")
    val whitenedData: RDD[(DenseVector[Double], Double)] = documents.map {
      case (_, length, vec) => (project(dimVocab, dimK, alpha0, eigenValues, eigenVectors, vec)(tolerance), length)
    }
    val firstOrderMoments_whitened: DenseVector[Double] = whitenedData
      .map(x => x._1 / x._2)
      .reduce((a, b) => a :+ b)
      .map(x => x / numDocs.toDouble)
    println("Finished whitening data.")

    println("Start calculating third order moments...")
    val Ta: DenseMatrix[Double] = whitenedData map {
      case (vec, len) => update_thirdOrderMoments(
        dimK, alpha0,
        firstOrderMoments_whitened,
        vec, len)
    } reduce(_ + _)

    val alpha0sq: Double = alpha0 * alpha0
    val Ta_shift = DenseMatrix.zeros[Double](dimK, dimK * dimK)
    for (id_i <- 0 until dimK optimized) {
      for (id_j <- 0 until dimK optimized) {
        for (id_l <- 0 until dimK optimized) {
          Ta_shift(id_i, id_j * dimK + id_l) += (alpha0sq * firstOrderMoments_whitened(id_i)
            * firstOrderMoments_whitened(id_j) * firstOrderMoments_whitened(id_l))
        }
      }
    }
    println("Finished calculating third order moments.")

    val thirdOrderMoments = Ta / numDocs.toDouble + Ta_shift
    val unwhiteningMatrix: breeze.linalg.DenseMatrix[Double] = eigenVectors * breeze.linalg.diag(eigenValues.map(x => scala.math.sqrt(x)))

    new DataCumulant(thirdOrderMoments, unwhiteningMatrix)
  }

  private def whiten(sc: SparkContext,
                     alpha0: Double,
                     vocabSize: Int, dimK: Int,
                     numDocs: Long,
                     firstOrderMoments: DenseVector[Double],
                     documents: RDD[(Long, Double, SparseVector[Double])])
  : (DenseMatrix[Double], DenseVector[Double]) = {
    val para_main: Double = (alpha0 + 1.0) / numDocs.toDouble
    val para_shift: Double = alpha0

    val SEED_random: Long = System.currentTimeMillis
    val gaussianRandomMatrix: DenseMatrix[Double] = AlgebraUtil.gaussian(vocabSize, dimK * 2, SEED_random)
    val gaussianRandomMatrix_broadcasted: Broadcast[breeze.linalg.DenseMatrix[Double]] = sc.broadcast(gaussianRandomMatrix)
    val firstOrderMoments_broadcasted: Broadcast[breeze.linalg.DenseVector[Double]] = sc.broadcast(firstOrderMoments.toDenseVector)
{'fraction_non_alphanumeric': 0.10338041213243807, 'fraction_numerical': 0.007293354943273906, 'mean_word_length': 3.090435606060606, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1374391', 'n_tokens_mistral': 2849, 'n_tokens_neox': 2688, 'n_words': 587}