<commit_before>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.github.cloudml.zen.ml.clustering

import java.util.concurrent.Executors
import java.util.concurrent.atomic.AtomicInteger
import scala.concurrent._
import scala.concurrent.duration.Duration

import LDADefines._

import breeze.linalg.{DenseVector => BDV, SparseVector => BSV}
import org.apache.spark.graphx2.impl.GraphImpl


trait LDAMetrics {
  def measure(): Unit
}

class LDAPerplexity(lda: LDA) extends LDAMetrics {
  var pplx = 0D
  var wpplx = 0D
  var dpplx = 0D

  /**
   * the multiplcation between word distribution among all topics and the corresponding doc
   * distribution among all topics:
   * p(w)=\sum_{k}{p(k|d)*p(w|k)}=
   * \sum_{k}{\frac{{n}_{kw}+{\beta }_{w}} {{n}_{k}+\bar{\beta }} \frac{{n}_{kd}+{\alpha }_{k}}{\sum{{n}_{k}}+
   * \bar{\alpha }}}=
   * \sum_{k} \frac{{\alpha }_{k}{\beta }_{w}  + {n}_{kw}{\alpha }_{k} + {n}_{kd}{\beta }_{w} + {n}_{kw}{n}_{kd}}
   * {{n}_{k}+\bar{\beta }} \frac{1}{\sum{{n}_{k}}+\bar{\alpha }}}
   * \exp^{-(\sum{\log(p(w))})/N}
   * N is the number of tokens in corpus
   */
  def measure(): Unit = {
    val topicCounters = lda.topicCounters
    val numTopics = lda.numTopics
    val numTokens = lda.numTokens
    val alphaAS = lda.alphaAS
    val alphaSum = numTopics * lda.alpha
    val alphaRatio = alphaSum / (numTokens + alphaAS * numTopics)
    val beta = lda.beta
    val betaSum = lda.numTerms * beta

    val graph = lda.corpus.asInstanceOf[GraphImpl[TC, TA]]
    val vertices = graph.vertices
    val edges = graph.edges
    val numThreads = edges.context.getConf.getInt(cs_numThreads, 1)

    // \frac{{\alpha }_{k}{\beta }_{w}}{{n}_{k}+\bar{\beta }}
    val tDenseSum = topicCounters.valuesIterator.map(cnt =>
      beta * alphaRatio * (cnt + alphaAS) / (cnt + betaSum)
    ).sum

    val partRDD = vertices.partitionsRDD.mapPartitions(_.map(svp => {
      val index = svp.index
      val values= svp.values
      val results = new Array[(TC, Double, Int)](svp.capacity)
      implicit val ec = ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(numThreads))
      val all = Future.traverse(svp.mask.iterator)(i => Future {
        val vid = index.getValue(i)
        val counter = values(i)
        val pSum = counter.activeIterator.filter(_._2 > 0).map {
          case (topic, cnt) =>
            if (isDocId(vid)) {
              cnt * beta / (topicCounters(topic) + betaSum)
            } else {
              cnt * alphaRatio * (topicCounters(topic) + alphaAS) / (topicCounters(topic) + betaSum)
            }
        }.sum
        val cSum = if (isDocId(vid)) counter.activeValuesIterator.sum else 0
        results(i) = (counter, pSum, cSum)
      })
      Await.ready(all, Duration.Inf)
      ec.shutdown()
      svp.withValues(results)
    }), preservesPartitioning=true)
    val cachedGraph = GraphImpl.fromExistingRDDs(vertices.withPartitionsRDD(partRDD), edges)

    val refrGraph = refreshEdgeAssociations(cachedGraph)
    val sumPart = refrGraph.edges.partitionsRDD.mapPartitions(_.map(t => {
      val ep = t._2
      val totalSize = ep.size
      val lcSrcIds = ep.localSrcIds
      val lcDstIds = ep.localDstIds
      val vattrs = ep.vertexAttrs
      val data = ep.data
      val sums = new Array[Double](numThreads)
      val wSums = new Array[Double](numThreads)
      val dSums = new Array[Double](numThreads)
      val indicator = new AtomicInteger

      implicit val ec = ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(numThreads))
      val all = Future.traverse(ep.index.iterator)(t => Future {
        val thid = indicator.getAndIncrement() % numThreads
        var pos = t._2
        val si = lcSrcIds(pos)
        val (orgTermTopics, wSparseSum, _) = vattrs(si)
        val termTopics = orgTermTopics match {
          case v: BDV[Count] => v
          case v: BSV[Count] => toBDV(v)
        }
        var lcSum = 0D
        var lcWSum = 0D
        var lcDSum = 0D
        while (pos < totalSize && lcSrcIds(pos) == si) {
          val (docTopics, dSparseSum, docSize) = vattrs(lcDstIds(pos))
          val topics = data(pos)
          // \frac{{n}_{kw}{n}_{kd}}{{n}_{k}+\bar{\beta}}
          val dwSparseSum = docTopics.activeIterator.map {
            case (topic, cnt) =>
              val ntw = termTopics(topic)
              if (ntw > 0 ) {
                cnt * ntw / (topicCounters(topic) + betaSum)
              } else {
                0D
              }
          }.sum
          val prob = (tDenseSum + wSparseSum + dSparseSum + dwSparseSum) / (docSize + alphaSum)
          lcSum += Math.log(prob) * topics.length
          for (topic <- topics) {
            val wProb = (termTopics(topic) + beta) / (topicCounters(topic) + betaSum)
            val dProb = (docTopics(topic) + alphaRatio * (topicCounters(topic) + alphaAS)) /
              (docSize + alphaSum)
            lcWSum += Math.log(wProb)
            lcDSum += Math.log(dProb)
          }
          pos += 1
        }
        sums(thid) += lcSum<commit_msg>final release<commit_after>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.github.cloudml.zen.ml.clustering

import java.util.concurrent.Executors
import java.util.concurrent.atomic.AtomicInteger
import scala.concurrent._
import scala.concurrent.duration.Duration

import LDADefines._

import breeze.linalg.{DenseVector => BDV, SparseVector => BSV}
import org.apache.spark.graphx2.impl.GraphImpl


trait LDAMetrics {
  def measure(): Unit
}

class LDAPerplexity(lda: LDA) extends LDAMetrics {
  var pplx = 0D
  var wpplx = 0D
  var dpplx = 0D

  /**
   * the multiplcation between word distribution among all topics and the corresponding doc
   * distribution among all topics:
   * p(w)=\sum_{k}{p(k|d)*p(w|k)}=
   * \sum_{k}{\frac{{n}_{kw}+{\beta }_{w}} {{n}_{k}+\bar{\beta }} \frac{{n}_{kd}+{\alpha }_{k}}{\sum{{n}_{k}}+
   * \bar{\alpha }}}=
   * \sum_{k} \frac{{\alpha }_{k}{\beta }_{w}  + {n}_{kw}{\alpha }_{k} + {n}_{kd}{\beta }_{w} + {n}_{kw}{n}_{kd}}
   * {{n}_{k}+\bar{\beta }} \frac{1}{\sum{{n}_{k}}+\bar{\alpha }}}
   * \exp^{-(\sum{\log(p(w))})/N}
   * N is the number of tokens in corpus
   */
  def measure(): Unit = {
    val topicCounters = lda.topicCounters
    val numTopics = lda.numTopics
    val numTokens = lda.numTokens
    val alphaAS = lda.alphaAS
    val alphaSum = numTopics * lda.alpha
    val alphaRatio = alphaSum / (numTokens + alphaAS * numTopics)
    val beta = lda.beta
    val betaSum = lda.numTerms * beta

    val graph = lda.corpus.asInstanceOf[GraphImpl[TC, TA]]
    val vertices = graph.vertices
    val edges = graph.edges
    val numThreads = edges.context.getConf.getInt(cs_numThreads, 1)

    // \frac{{\alpha }_{k}{\beta }_{w}}{{n}_{k}+\bar{\beta }}
    val tDenseSum = topicCounters.valuesIterator.map(cnt =>
      beta * alphaRatio * (cnt + alphaAS) / (cnt + betaSum)
    ).sum

    val partRDD = vertices.partitionsRDD.mapPartitions(_.map(svp => {
      val index = svp.index
      val values= svp.values
      val results = new Array[(TC, Double, Int)](svp.capacity)
      implicit val ec = ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(numThreads))
      val all = Future.traverse(svp.mask.iterator)(i => Future {
        val vid = index.getValue(i)
        val counter = values(i)
        val pSum = counter.activeIterator.filter(_._2 > 0).map {
          case (topic, cnt) =>
            if (isDocId(vid)) {
              cnt * beta / (topicCounters(topic) + betaSum)
            } else {
              cnt * alphaRatio * (topicCounters(topic) + alphaAS) / (topicCounters(topic) + betaSum)
            }
        }.sum
        val cSum = if (isDocId(vid)) counter.activeValuesIterator.sum else 0
        results(i) = (counter, pSum, cSum)
      })
      Await.ready(all, Duration.Inf)
      ec.shutdown()
      svp.withValues(results)
    }), preservesPartitioning=true)
    val cachedGraph = GraphImpl.fromExistingRDDs(vertices.withPartitionsRDD(partRDD), edges)

    val refrGraph = refreshEdgeAssociations(cachedGraph)
    val sumPart = refrGraph.edges.partitionsRDD.mapPartitions(_.map(t => {
      val ep = t._2
      val totalSize = ep.size
      val lcSrcIds = ep.localSrcIds
      val lcDstIds = ep.localDstIds
      val vattrs = ep.vertexAttrs
      val data = ep.data
      val sums = new Array[Double](numThreads)
      val wSums = new Array[Double](numThreads)
      val dSums = new Array[Double](numThreads)
      val indicator = new AtomicInteger

      implicit val ec = ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(numThreads))
      val all = Future.traverse(ep.index.iterator)(t => Future {
        val thid = indicator.getAndIncrement() % numThreads
        var pos = t._2
        val si = lcSrcIds(pos)
        val (orgTermTopics, wSparseSum, _) = vattrs(si)
        val termTopics = orgTermTopics match {
          case v: BDV[Count] => v
          case v: BSV[Count] => toBDV(v)
        }
        var lcSum = 0D
        var lcWSum = 0D
        var lcDSum = 0D
        while (pos < totalSize && lcSrcIds(pos) == si) {
          val (docTopics, dSparseSum, docSize) = vattrs(lcDstIds(pos))
          val topics = data(pos)
          // \frac{{n}_{kw}{n}_{kd}}{{n}_{k}+\bar{\beta}}
          val dwSparseSum = docTopics.activeIterator.map {
            case (topic, cnt) =>
              cnt * termTopics(topic) / (topicCounters(topic) + betaSum)
          }.sum
          val prob = (tDenseSum + wSparseSum + dSparseSum + dwSparseSum) / (docSize + alphaSum)
          lcSum += Math.log(prob) * topics.length
          for (topic <- topics) {
            val wProb = (termTopics(topic) + beta) / (topicCounters(topic) + betaSum)
            val dProb = (docTopics(topic) + alphaRatio * (topicCounters(topic) + alphaAS)) /
              (docSize + alphaSum)
            lcWSum += Math.log(wProb)
            lcDSum += Math.log(dProb)
          }
          pos += 1
        }
        sums(thid) += lcSum
{'fraction_non_alphanumeric': 0.12582781456953643, 'fraction_numerical': 0.003485535029627048, 'mean_word_length': 2.689167470266795, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 28, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7060942', 'n_tokens_mistral': 3746, 'n_tokens_neox': 3645, 'n_words': 942}