<commit_before>package embed.tmp

import java.util.Random

import embed.lda.LDAConfig
import embed.sampling.{AliasTable, Dice}
import org.apache.spark.rdd.RDD
import embed.lda.mh.Sampler
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by chris on 8/14/17.
  */
// 假设所有数据结构是Dense的。
// 本地数据是D*V（按照文章存储，每个文章用词包抽象）, server上的数据是该模型的model(word-topic count table,
// 相当于lr里面的参数) 以及一个global topic count。纪录所有文章中每一个topic有多少词
// Array的下标就是词的id，因此D*V 实际上存储的是每一个token所分配的topic
object PsWarpLda {
  def main(args: Array[String]): Unit = {

    //迭代次数
    val iterations = 1000

    //总词数
    val word = 100

    //总topic数
    val topic = 10

    //mcmc的步数
    val mhSteps = 10

    val r = new Random()

    //初始化模型word-topic matrix 根据DICE随机生成
    val model = Initialization.topicInitialization(word,topic)

    // lda configuration
    val config = LDAConfig()

    // 注册ps word-topic count table
    val pSMatrix = PSMatrix().createModel(model)

    // 计算出每一个topic 有多少个token，数据格式是dense array，以下标来表示topic
    val globalTopicCount = model.transpose.map{f=>
      f.sum
    }

    //注册 ps global vector
    val pSVector = PSVector().createModel(globalTopicCount)

    //data这里面就是D×V的topic
    val conf = new SparkConf()
    val sc = new SparkContext(conf)
    val data: RDD[Array[Int]] = sc.parallelize(model)

    //进行迭代
    (0 until iterations) foreach{i=>

      data.mapPartitions{documents =>
        //为了防止通信过多在本地建立alias table这样会产生多余的alias table，比如“我”这个词会在每个partition上出现建立一个aliastable，这些aliastable实际上是重复的，
        // 实际上是可以将建立aliastable放在ps上边的，这个需要后期优化
        val buf = new collection.mutable.ArrayBuffer[Array[Int]](documents.length)

        documents.copyToBuffer(buf)


        //检查该partition里面包含多少个word
        val wordsOfPartitions = buf.flatMap(f=>f).distinct.toArray

        //通过这些word 从ps里面将pSMatrix对应的行pull下来
        val wordsDistributions = pSMatrix.pullByRows(wordsOfPartitions).zip(wordsOfPartitions).map{

          //从psMatrix拉下来行，对每一个词建立aliastable，并记录下wordTopicDistribution
          case (topicDistribution, word) =>
            // sparse topic Distribution
            val sparseTopicDistribution = topicDistribution.zipWithIndex.filter(_._1 != 0).map(f =>(f._2,f._1))
            val aliasTable = new AliasTable(sparseTopicDistribution)
            (word,(aliasTable, topicDistribution))
        }.toMap[Int,(AliasTable,Array[Int])]

        buf.map{tokens => // tokens这里指带的是一个token的topic， token对应的词用下标表示（tokens是文章的topic集合）
          val documentTopicAssignments = tokens
          val documentSize: Int = tokens.length
          val sampler = new Sampler(config, mhSteps, r)
          sampler.documentTopicAssignments = documentTopicAssignments
          sampler.documentSize = documentSize
          (0 until tokens.length) map{i =>
            // 从每一个partition的wordsDistribution Count table 中取出来每一个词对应的alias table 和 topic 分布
            val (aliasTable,topicDistribution) = wordsDistributions.get(i).get
            var wordCounts = topicDistribution

            //将global topic vector pull下来
            var globalCounts = pSVector.pullByRows(0)

            //sampler 需要根据词来定制
            sampler.globalCounts = globalCounts
            sampler.aliasTable = aliasTable
            sampler.wordCounts = wordCounts

            //根据上一步的结果进行采样
            val z = sampler.sample(tokens(i))

            if (z != tokens(i)){
              globalCounts(z) += 1
              globalCounts(tokens(i)) -= 1
              wordCounts(z) += 1
              wordCounts(tokens(i)) -= 1
              //更新word - topic table
              pSMatrix.incrementAndFlush(i, wordCounts)
              //更新global topic count
              pSVector.incrementAndFlush(0,globalCounts)
            }
            z //更新本地的D*V 表
        }<commit_msg>modified pseudo code<commit_after>package embed.tmp

import java.util.Random

import embed.lda.LDAConfig
import embed.sampling.{AliasTable, Dice}
import org.apache.spark.rdd.RDD
import embed.lda.mh.Sampler
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by chris on 8/14/17.
  */
// 假设所有数据结构是Dense的。
// 本地数据是D*V（按照文章存储，每个文章用词包抽象）, server上的数据是该模型的model(word-topic count table,
// 相当于lr里面的参数) 以及一个global topic count。纪录所有文章中每一个topic有多少词
// Array的下标就是词的id，因此D*V 实际上存储的是每一个token所分配的topic
object PsWarpLda {
  def main(args: Array[String]): Unit = {

    //迭代次数
    val iterations = 1000

    //总词数
    val word = 100

    //总topic数
    val topic = 10

    //mcmc的步数
    val mhSteps = 10

    val r = new Random()

    //初始化模型word-topic matrix 根据DICE随机生成
    val model = Initialization.topicInitialization(word,topic)

    // lda configuration
    val config = LDAConfig()

    // 注册ps word-topic count table
    val pSMatrix = PSMatrix().createModel(model)

    // 计算出每一个topic 有多少个token，数据格式是dense array，以下标来表示topic
    val globalTopicCount = model.transpose.map{f=>
      f.sum
    }

    //注册 ps global vector
    val pSVector = PSVector().createModel(globalTopicCount)

    //data这里面就是D×V的topic
    val conf = new SparkConf()
    val sc = new SparkContext(conf)
    val data: RDD[Array[Int]] = sc.parallelize(model)

    //进行迭代
    (0 until iterations) foreach{i=>

      data.mapPartitions{documents =>
        //为了防止通信过多在本地建立alias table这样会产生多余的alias table，比如“我”这个词会在每个partition上出现建立一个aliastable，这些aliastable实际上是重复的，
        // 实际上是可以将建立aliastable放在ps上边的，这个需要后期优化
        val buf = new collection.mutable.ArrayBuffer[Array[Int]](documents.length)

        documents.copyToBuffer(buf)


        //检查该partition里面包含多少个word
        val wordsOfPartitions = buf.flatMap(f=>f).distinct.toArray

        //通过这些word 从ps里面将pSMatrix对应的行pull下来
        val wordsDistributions = pSMatrix.pullByRows(wordsOfPartitions).zip(wordsOfPartitions).map{

          //从psMatrix拉下来行，对每一个词建立aliastable，并记录下wordTopicDistribution
          case (topicDistribution, word) =>
            // sparse topic Distribution
            val sparseTopicDistribution = topicDistribution.zipWithIndex.filter(_._1 != 0).map(f =>(f._2,f._1))
            val aliasTable = new AliasTable(sparseTopicDistribution)
            (word,(aliasTable, topicDistribution))
        }.toMap[Int,(AliasTable,Array[Int])]

        buf.map{tokens => // tokens这里指带的是一个token的topic， token对应的词用下标表示（tokens是文章的topic集合）
          val documentTopicAssignments = tokens
          val documentSize: Int = tokens.length
          val sampler = new Sampler(config, mhSteps, r)

          //sampler需要根据每一个document定制
          sampler.documentTopicAssignments = documentTopicAssignments
          sampler.documentSize = documentSize
          (0 until tokens.length) map{i =>
            // 从每一个partition的wordsDistribution Count table 中取出来每一个词对应的alias table 和 topic 分布
            val (aliasTable,topicDistribution) = wordsDistributions.get(i).get
            var wordCounts = topicDistribution

            //将global topic vector pull下来
            var globalCounts = pSVector.pullByRows(0)

            //sampler 需要根据词来定制
            sampler.globalCounts = globalCounts
            sampler.aliasTable = aliasTable
            sampler.wordCounts = wordCounts

            //根据上一步的结果进行采样
            val z = sampler.sample(tokens(i))

            if (z != tokens(i)){
              globalCounts(z) += 1
              globalCounts(tokens(i)) -= 1
              wordCounts(z) += 1
              wordCounts(tokens(i)) -= 1
              //更新word - topic table
              pSMatrix.incrementAndFlush(i, wordCounts)
              //更新global topic count
              pSVector.incrementAndFlush(0,globalCounts)
            }
            z //更新本地的D*V 表
        }
{'fraction_non_alphanumeric': 0.09255586407510247, 'fraction_numerical': 0.007404469126008198, 'mean_word_length': 2.55618241654913, 'pattern_counts': {'":': 0, '<': 3, '<?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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7307649', 'n_tokens_mistral': 2768, 'n_tokens_neox': 2768, 'n_words': 455}