// Setup Wikipedia source
val lang = "sv"
val freeze_date = "20201120"
val wp_file = s"${lang}wiki-${freeze_date}-pages-articles.xml"
val wp_zip_remote = s"https://dumps.wikimedia.org/${lang}wiki/${freeze_date}/${wp_file}.bz2"

// Stopwords download soure
val stopwords_remote = "https://raw.githubusercontent.com/peterdalle/svensktext/master/stoppord/stoppord.csv"

// Setup base paths for storage
val group = "05_lda"  // directory name to separate file structure, allows to run different instances in parallel
val localpath=s"file:/databricks/driver/$group/"
val dir_local = s"$group/"
val dir = s"dbfs:/datasets/$group/"

// Some file names
val stopwords_file = "stopwords.csv"
val stopwords_store = s"${dir}${stopwords_file}"

val wp_store = s"${dir}${wp_file}"

val filtered_articles_store = s"${dir}filtered_articles_${lang}_${freeze_date}"
val lda_countVector_store = s"${dir}lda_countvector_${lang}_${freeze_date}"

val lda_model_store = s"${dir}lda_${lang}_${freeze_date}.model"

lang: String = sv
freeze_date: String = 20201120
wp_file: String = svwiki-20201120-pages-articles.xml
wp_zip_remote: String = https://dumps.wikimedia.org/svwiki/20201120/svwiki-20201120-pages-articles.xml.bz2
stopwords_remote: String = https://raw.githubusercontent.com/peterdalle/svensktext/master/stoppord/stoppord.csv
group: String = 05_lda
localpath: String = file:/databricks/driver/05_lda/
dir_local: String = 05_lda/
dir: String = dbfs:/datasets/05_lda/
stopwords_file: String = stopwords.csv
stopwords_store: String = dbfs:/datasets/05_lda/stopwords.csv
wp_store: String = dbfs:/datasets/05_lda/svwiki-20201120-pages-articles.xml
filtered_articles_store: String = dbfs:/datasets/05_lda/filtered_articles_sv_20201120
lda_countVector_store: String = dbfs:/datasets/05_lda/lda_countvector_sv_20201120
lda_model_store: String = dbfs:/datasets/05_lda/lda_sv_20201120.model

# Just check where local path is
pwd

/databricks/driver

// Run this only to force a clean start
/*
// Delete DBFS store
dbutils.fs.rm(dir,recurse=true)
// Delete local files
import org.apache.commons.io.FileUtils;
import java.io.File;
FileUtils.deleteDirectory(new File(dir_local))
*/

import org.apache.commons.io.FileUtils
import java.io.File

// Prepare DBFS store
dbutils.fs.mkdirs(dir)
display(dbutils.fs.ls(dir))
// Prepare local folder
import org.apache.commons.io.FileUtils;
import java.io.File;
FileUtils.forceMkdir(new File(dir_local));

// Check local files
import sys.process._  // For spawning shell commands
s"ls -l ./  ${dir_local}" !!

warning: there was one feature warning; for details, enable `:setting -feature' or `:replay -feature'
import sys.process._
res19: String =
"./:
total 24
drwxr-xr-x 2 root root 4096 Jan 26 04:22 05_lda
drwxr-xr-x 2 root root 4096 Jan  1  1970 conf
-rw-r--r-- 1 root root  733 Jan 26 03:34 derby.log
drwxr-xr-x 3 root root 4096 Jan 26 03:33 eventlogs
drwxr-xr-x 2 root root 4096 Jan 26 04:15 ganglia
drwxr-xr-x 2 root root 4096 Jan 26 04:03 logs

05_lda/:
total 0
"

// Download stopwords to store.
val local = s"${dir_local}${stopwords_file}"
val remote = stopwords_remote
try
{
  // -nc prevents download if already exist. !! converts wget output to string.
  s"wget -nc -O${local} ${remote}" !!
}
catch
{
  case e: Throwable => println("Exception: " + e)
}

--2021-01-26 04:23:45--  https://raw.githubusercontent.com/peterdalle/svensktext/master/stoppord/stoppord.csv
Resolving raw.githubusercontent.com (raw.githubusercontent.com)... 151.101.52.133
Connecting to raw.githubusercontent.com (raw.githubusercontent.com)|151.101.52.133|:443... connected.
HTTP request sent, awaiting response... 200 OK
Length: 1936 (1.9K) [text/plain]
Saving to: ‘05_lda/stopwords.csv’

     0K .                                                     100% 15.9M=0s

2021-01-26 04:23:45 (15.9 MB/s) - ‘05_lda/stopwords.csv’ saved [1936/1936]

warning: there was one feature warning; for details, enable `:setting -feature' or `:replay -feature'
local: String = 05_lda/stopwords.csv
remote: String = https://raw.githubusercontent.com/peterdalle/svensktext/master/stoppord/stoppord.csv
res20: Any = ""

// Check local stopwords file
s"ls -l ${dir_local}" !!

warning: there was one feature warning; for details, enable `:setting -feature' or `:replay -feature'
res21: String =
"total 4
-rw-r--r-- 1 root root 1936 Jan 26 04:23 stopwords.csv
"

dbutils.fs.cp(localpath + stopwords_file, dir)

res22: Boolean = true

display(dbutils.fs.ls(stopwords_store))

// Download Wikipedia xml to store. -nc prevents overwrite if already exist. !! converts wget output to string.
val local = s"${dir_local}${wp_file}.bz2"
val remote = wp_zip_remote
try
{
  s"wget -nc -O${local} ${remote}" !!
}
catch
{
  case e: Throwable => println("Exception: " + e)
}

// Unpack bz2 to xml.
val local = s"${dir_local}${wp_file}.bz2"
try
{
  // -d unpack, -k keep source file, -v activate some output. !! converts output to string.
  s"bzip2 -dk -f -v ${local}" !!
}
catch
{
  case e: Throwable => println("Exception: " + e)
}

  05_lda/svwiki-20201120-pages-articles.xml.bz2: done
warning: there was one feature warning; for details, enable `:setting -feature' or `:replay -feature'
local: String = 05_lda/svwiki-20201120-pages-articles.xml.bz2
res27: Any = ""

s"ls ${dir_local}" !!

warning: there was one feature warning; for details, enable `:setting -feature' or `:replay -feature'
res28: String =
"stopwords.csv
svwiki-20201120-pages-articles.xml
svwiki-20201120-pages-articles.xml.bz2
"

val local = s"${localpath}${wp_file}"
dbutils.fs.cp(local, dir)

local: String = file:/databricks/driver/05_lda/svwiki-20201120-pages-articles.xml
res29: Boolean = true

display(dbutils.fs.ls(wp_store))

// Discard the local files no longer needed
import org.apache.commons.io.FileUtils;
import java.io.File;
FileUtils.cleanDirectory(new File(dir_local))

import org.apache.commons.io.FileUtils
import java.io.File

// ---------------spark_import
import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
import org.apache.spark.sql.SQLContext

// ----------------xml_loader_import
import org.apache.hadoop.io.LongWritable
import org.apache.hadoop.io.Text
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.io.{ LongWritable, Text }
import com.databricks.spark.xml._

import org.apache.hadoop.conf.Configuration

// ---- Matching
import scala.util.matching.UnanchoredRegex

import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
import org.apache.spark.sql.SQLContext
import org.apache.hadoop.io.LongWritable
import org.apache.hadoop.io.Text
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.io.{LongWritable, Text}
import com.databricks.spark.xml._
import org.apache.hadoop.conf.Configuration
import scala.util.matching.UnanchoredRegex

def readWikiDump(sc: SparkContext, file: String) : RDD[String] = {
  val conf = new Configuration()
  conf.set(XmlInputFormat.START_TAG_KEY, "<page>")
  conf.set(XmlInputFormat.END_TAG_KEY, "</page>")
  val rdd = sc.newAPIHadoopFile(file,
                                classOf[XmlInputFormat],
                                classOf[LongWritable],
                                classOf[Text],
                                conf)
  rdd.map{case (k,v) => (new String(v.copyBytes()))}
}

readWikiDump: (sc: org.apache.spark.SparkContext, file: String)org.apache.spark.rdd.RDD[String]

// Read the wiki dump
val dump = readWikiDump(sc, wp_store)

val n_wp_pages = dump.count()
dump.take(5)

val redirection = "<redirect title=".r
def is_not_redirect(source : String) = { redirection.findFirstIn(source) == None }

val ns0 = "<ns>0</ns>".r
def is_article(source : String) = { ns0.findFirstIn(source) != None }

val autogenerated = "robotskapad|Robotskapad".r
def is_not_autogenerated(source : String) = { autogenerated.findFirstIn(source) == None }

val articles = dump.filter(is_not_redirect).filter(is_article).filter(is_not_autogenerated)

val n_articles = articles.count()
articles.take(5)

val id_regexp = raw"<id>([0-9]*)</id>".r
def extract_id(source : String) = {
  val m = id_regexp.findFirstMatchIn(source).get
  m.group(1).toLong
}

val title_regexp = raw"<title>([^\<\>]*)</title>".r
def extract_title(source : String) = {
  val m = title_regexp.findFirstMatchIn(source).get
  m.group(1)
}

val labelled_articles = articles.map(source => (extract_id(source), extract_title(source), source))

labelled_articles.take(5).map({case(id, title, contents) => title})

id_regexp: scala.util.matching.Regex = <id>([0-9]*)</id>
extract_id: (source: String)Long
title_regexp: scala.util.matching.Regex = <title>([^\<\>]*)</title>
extract_title: (source: String)String
labelled_articles: org.apache.spark.rdd.RDD[(Long, String, String)] = MapPartitionsRDD[1021] at map at command-2629358840275853:13
res11: Array[String] = Array(Amager, Afrika, Amerika, Abbekås, Alingsås)

def filter_xml(source : String) = {
  val noAmp = raw"&".r.replaceAllIn(source, "&")
  val noNbsp = raw" ".r.replaceAllIn(noAmp, " ")
  val noQuot = raw""".r.replaceAllIn(noNbsp, "")
  val noLt = raw"&lt;".r.replaceAllIn(noQuot, "<")
  val noGt = raw"&gt;".r.replaceAllIn(noLt, ">")



  val tags = Seq("id", "ns", "parentid", "timestamp", "contributor", "comment", "model", "format", "sha1")
  var current = noGt
  for (tag <- tags) {
    val start = s"<$tag>".r
    val end = s"</$tag>".r
    while (start.findAllIn(current).hasNext) {
      val a = start.findAllIn(current).start
      val b = end.findAllIn(current).start
      val len = current.length
      current = current.substring(0, a) + current.substring(b+tag.length+3, current.length)
      assert (current.length < len)
    }
  }
  current = raw"<[^<>]*>".r.replaceAllIn(current, "")

  // A loop to remove innermost curly braces, then outer such things.
  val curly = raw"\{[^\{\}]*\}".r
  while (curly.findAllIn(current).hasNext)
  {
     current = curly.replaceAllIn(current, "")
  }

  current
  }


val filtered_xml = labelled_articles.map({ case(id, title, source) => (id, title, filter_xml(source)) })

filtered_xml.take(5)

def sanitize(source: String) : String = {
  val noDollar = source.replaceAllLiterally("$", "")
  val noBackSlash = noDollar.replaceAllLiterally(raw"\", "")
  noBackSlash
}

val sanitized_xml = filtered_xml.map({case(id, title, source) => (id, title, sanitize(source))})

sanitize: (source: String)String
sanitized_xml: org.apache.spark.rdd.RDD[(Long, String, String)] = MapPartitionsRDD[1026] at map at command-2629358840275860:7

def filterMarkup(source: String) : String = {

  // Loop to remove some specific sections
  val sections = Seq("Se även", "Referenser", "Noter", "Källor", "Externa länkar")
  var current = source
  for (section <- sections) {
    val regexp = s"[=]+ *$section *[=]+".r
    if (regexp.findAllIn(current).hasNext)
    {
      val a = regexp.findAllIn(current).start
      val pre = current.substring(0, a)
      var post = current.substring(a, current.length)
      post = regexp.replaceAllIn(post, "")
      if ("^=".r.findAllIn(post).hasNext) {
        val b = "^=".r.findAllIn(post).start
        post = post.substring(b, post.length)
      }
      else {
        post = ""        
      }
      current = pre + post
    }
  }

  // Unnamed internal links appear in figure captions, confusing later matching
  val noUnnamedInternalLinks = raw"\[\[([^\|\]]*)\]\]".r.replaceAllIn(current, m => m.group(1))

  // Loop to remove internal links and their properties
  current = noUnnamedInternalLinks
  val regexp = raw"\[\[[^\|\[]*\|([^\[\]]*)\]\]".r
  while (regexp.findAllIn(current).hasNext) {
    val len = current.length
    current = regexp.replaceAllIn(current, m => "[[" + m.group(1) + "]]")
    assert (current.length < len)
  }
  val noNamedInternalLinks = current

  val noInternalLinks = raw"\[\[([^\]]*)\]\]".r.replaceAllIn(noNamedInternalLinks, m => m.group(1))
  val noNamedExternalLinks = raw"\[[^ \[]*\ ([^\]]*)\]".r.replaceAllIn(noInternalLinks, m => m.group(1))
  val noExternalLinks = raw"\[([^\]]*)\]".r.replaceAllIn(noNamedExternalLinks, m => m.group(1))


 val lowerCase = noExternalLinks.toLowerCase  

  val noQuoteChars = raw"'".r.replaceAllIn(lowerCase, "")
  // Loop to remove double whitespace characters
  current = noQuoteChars
  val doubleWhitespace = raw"\s(\s)".r
  while (doubleWhitespace.findAllIn(current).hasNext) {
    current = doubleWhitespace.replaceAllIn(current, m => m.group(1))
  }
  val noDoubleWhitespace = current
  noDoubleWhitespace
}


val filtered = sanitized_xml.map({case (id, title, source) => (id, title, filterMarkup(source))})

filtered.take(2)

// dbutils.fs.rm(filtered_articles_store, recurse=true)

res19: Boolean = true

val outputs = filtered.map({case (id, title, contents) => (id, title.replaceAllLiterally(",", "**COMMA**"), contents.replaceAllLiterally("\n", "**NEWLINE**"))})
outputs.saveAsTextFile(filtered_articles_store);

outputs: org.apache.spark.rdd.RDD[(Long, String, String)] = MapPartitionsRDD[1038] at map at command-4443336225071544:1

val format = "^\\(([^,]*),([^,]*),(.*)\\)$".r
def parse_line(s : String) = {
  assert (s(0) == '(')
  val firstComma = s.indexOf(",")
  val id = s.substring(1, firstComma).toLong
  val afterId = s.substring(firstComma+1, s.length)
  val nextComma = afterId.indexOf(",")
  val title = afterId.substring(0, nextComma)
  val contents = afterId.substring(nextComma+1, afterId.length-1)

  assert (afterId(afterId.length-1) == ')')
  (id, title.replaceAllLiterally("**COMMA**", ","), contents.replaceAllLiterally("**NEWLINE**", "\n"))
}

val filtered = spark.sparkContext.textFile(filtered_articles_store).map(parse_line)
val n_articles = filtered.count()

format: scala.util.matching.Regex = ^\(([^,]*),([^,]*),(.*)\)$
parse_line: (s: String)(Long, String, String)
filtered: org.apache.spark.rdd.RDD[(Long, String, String)] = MapPartitionsRDD[1042] at map at command-4443336225071727:15
n_articles: Long = 813245

val corpus_df = filtered.toDF("id", "title", "contents")

corpus_df: org.apache.spark.sql.DataFrame = [id: bigint, title: string ... 1 more field]

display(corpus_df.sample(0.0001))

import org.apache.spark.ml.feature.RegexTokenizer

// Set params for RegexTokenizer
val tokenizer = new RegexTokenizer()
  .setGaps(true) // pattern below identifies non-word alphabet
  .setPattern("[^a-zåäö]+") // break word and remove any detected non-word character(s).
  .setMinTokenLength(4) // Filter away tokens with length < 4
  .setInputCol("contents") // name of the input column
  .setOutputCol("tokens") // name of the output column

// Tokenize document
val tokenized_df = tokenizer.transform(corpus_df)

import org.apache.spark.ml.feature.RegexTokenizer
tokenizer: org.apache.spark.ml.feature.RegexTokenizer = RegexTokenizer: uid=regexTok_5d41547188e0, minTokenLength=4, gaps=true, pattern=[^a-zåäö]+, toLowercase=true
tokenized_df: org.apache.spark.sql.DataFrame = [id: bigint, title: string ... 2 more fields]

display(tokenized_df.sample(0.0001))

val swedish_stopwords = sc.textFile(stopwords_store).collect()
val wikipedia_stopwords = Array("label", "note", "area", "unit", "type", "mark", "long", "right", "kartposition", "quot", "text", "title", "page", "timestamp", "revision", "name", "username", "sha1", "format", "coord", "left", "center", "align", "region", "nasa", "source", "mouth", "species", "highest", "style", "kategori", "http", "wikipedia", "referenser", "källor", "noter")
// Combine newly identified stopwords to our exising list of stopwords
val stopwords = swedish_stopwords.union(wikipedia_stopwords)

import org.apache.spark.ml.feature.StopWordsRemover

val remover = new StopWordsRemover()
  .setStopWords(stopwords)
  .setInputCol("tokens")
  .setOutputCol("filtered")

val filtered_df = remover.transform(tokenized_df)

import org.apache.spark.ml.feature.StopWordsRemover
remover: org.apache.spark.ml.feature.StopWordsRemover = StopWordsRemover: uid=stopWords_ecf00fa02fd5, numStopWords=366, locale=en, caseSensitive=false
filtered_df: org.apache.spark.sql.DataFrame = [id: bigint, title: string ... 3 more fields]

import org.apache.spark.ml.feature.CountVectorizer

val vectorizer = new CountVectorizer()
  .setInputCol("filtered")
  .setOutputCol("features")
  .setVocabSize(10000)
  .setMinDF(5) // the minimum number of different documents a term must appear in to be included in the vocabulary.
  .fit(filtered_df)

import org.apache.spark.ml.feature.CountVectorizer
vectorizer: org.apache.spark.ml.feature.CountVectorizerModel = CountVectorizerModel: uid=cntVec_ad38aa4e1641, vocabularySize=10000

import org.apache.spark.ml.linalg.Vector

val countVectors = vectorizer.transform(filtered_df).select("id", "features")
val lda_countVector = countVectors.map { case Row(id: Long, countVector: Vector) => (id, countVector) }
val lda_countVector_mllib = lda_countVector.map { case (id, vector) => (id, org.apache.spark.mllib.linalg.Vectors.fromML(vector)) }.rdd

lda_countVector_mllib.take(1)

import org.apache.spark.ml.linalg.Vector
countVectors: org.apache.spark.sql.DataFrame = [id: bigint, features: vector]
lda_countVector: org.apache.spark.sql.Dataset[(Long, org.apache.spark.ml.linalg.Vector)] = [_1: bigint, _2: vector]
lda_countVector_mllib: org.apache.spark.rdd.RDD[(Long, org.apache.spark.mllib.linalg.Vector)] = MapPartitionsRDD[1088] at rdd at command-4443336225071537:5
res41: Array[(Long, org.apache.spark.mllib.linalg.Vector)] = Array((1,(10000,[1,2,8,9,15,17,20,26,37,38,39,42,48,50,51,55,64,67,68,71,75,80,81,85,86,94,95,96,106,114,117,118,128,129,140,149,150,156,161,189,195,208,216,297,299,332,336,399,409,427,452,457,459,461,465,535,540,572,603,612,686,739,747,806,849,904,922,983,1002,1013,1037,1081,1128,1167,1170,1175,1272,1274,1295,1334,1374,1387,1474,1578,1593,1602,1707,1750,1764,1792,1835,1852,1927,2022,2027,2114,2182,2268,2358,2612,2726,2766,2929,3316,3386,3415,3468,3486,3588,3593,3634,3693,3735,4156,4184,4335,4339,4470,4579,4675,4691,4851,5046,5083,5118,5186,5355,5653,5712,5801,6051,6080,6140,6751,7381,7491,7602,8758,8892,8944,8990,9148,9185,9375,9416,9737,9779,9870],[7.0,2.0,1.0,3.0,3.0,1.0,1.0,4.0,4.0,1.0,1.0,1.0,2.0,1.0,1.0,1.0,1.0,3.0,1.0,1.0,1.0,1.0,1.0,5.0,2.0,2.0,1.0,4.0,2.0,1.0,4.0,1.0,2.0,1.0,1.0,1.0,1.0,1.0,2.0,1.0,2.0,2.0,1.0,1.0,1.0,1.0,3.0,1.0,2.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,4.0,1.0,3.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,2.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,2.0,1.0,1.0,1.0,1.0,1.0,1.0,2.0,1.0,1.0,2.0,1.0,1.0,1.0,4.0,1.0,1.0,1.0,1.0,2.0,1.0,2.0,1.0,2.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,3.0,1.0,1.0,1.0,1.0,2.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,2.0,1.0,1.0,1.0,1.0,1.0,2.0,1.0,1.0,2.0,1.0,1.0,1.0,1.0,1.0,1.0])))

println(lda_countVector_store)

// lda_countVector_mllib.saveAsTextFile(lda_countVector_store);

dbfs:/datasets/05_lda/lda_countvector_sv_20201120

val numTopics = 50
val maxIterations = 20

numTopics: Int = 50
maxIterations: Int = 20

import org.apache.spark.mllib.clustering.{LDA, OnlineLDAOptimizer}

val lda = new LDA()
  .setOptimizer(new OnlineLDAOptimizer().setMiniBatchFraction(0.8))
  .setK(numTopics)
  .setMaxIterations(maxIterations)
  .setDocConcentration(-1) // use default values
  .setTopicConcentration(-1) // use default values

import org.apache.spark.mllib.clustering.{LDA, OnlineLDAOptimizer}
lda: org.apache.spark.mllib.clustering.LDA = org.apache.spark.mllib.clustering.LDA@54e767e2

val ldaModel = lda.run(lda_countVector_mllib)

ldaModel: org.apache.spark.mllib.clustering.LDAModel = org.apache.spark.mllib.clustering.LocalLDAModel@5c324bb0

// Save the model to file for later use.
dbutils.fs.rm(lda_model_store, recurse=true)
ldaModel.save(sc, lda_model_store)

val maxTermsPerTopic = 8
val topicIndices = ldaModel.describeTopics(maxTermsPerTopic)
val vocabList = vectorizer.vocabulary
val topics = topicIndices.map { case (terms, termWeights) =>
  terms.map(vocabList(_)).zip(termWeights)
}

println(s"$numTopics topics:")
topics.zipWithIndex.foreach { case (topic, i) =>
  println(s"TOPIC $i")
  topic.foreach { case (term, weight) => println(s"$term\t$weight") }
  println(s"==========")
}

Zip topic terms with topic IDs
val termArray = topics.zipWithIndex

// Transform data into the form (term, probability, topicId)
val termRDD = sc.parallelize(termArray)
val termRDD2 = termRDD.flatMap( (x: (Array[(String, Double)], Int) ) => {
  val arrayOfTuple = x._1
  val topicId = x._2
  arrayOfTuple.map(el => (el._1, el._2, topicId))
})

termRDD: org.apache.spark.rdd.RDD[(Array[(String, Double)], Int)] = ParallelCollectionRDD[1881] at parallelize at command-4443336225072171:2
termRDD2: org.apache.spark.rdd.RDD[(String, Double, Int)] = MapPartitionsRDD[1882] at flatMap at command-4443336225072171:3

// Create DF with proper column names
val termDF = termRDD2.toDF.withColumnRenamed("_1", "term").withColumnRenamed("_2", "probability").withColumnRenamed("_3", "topicId")

termDF: org.apache.spark.sql.DataFrame = [term: string, probability: double ... 1 more field]

 display(termDF)

Create JSON data
val rawJson = termDF.toJSON.collect().mkString(",\n")

displayHTML(s"""
<!DOCTYPE html>
<meta charset="utf-8">
<style>

circle {
  fill: rgb(31, 119, 180);
  fill-opacity: 0.5;
  stroke: rgb(31, 119, 180);
  stroke-width: 1px;
}

.leaf circle {
  fill: #ff7f0e;
  fill-opacity: 1;
}

text {
  font: 14px sans-serif;
}

</style>
<body>
<script src="https://cdnjs.cloudflare.com/ajax/libs/d3/3.5.5/d3.min.js"></script>
<script>

var json = {
 "name": "data",
 "children": [
  {
     "name": "topics",
     "children": [
      ${rawJson}
     ]
    }
   ]
};

var r = 1000,
    format = d3.format(",d"),
    fill = d3.scale.category20c();

var bubble = d3.layout.pack()
    .sort(null)
    .size([r, r])
    .padding(1.5);

var vis = d3.select("body").append("svg")
    .attr("width", r)
    .attr("height", r)
    .attr("class", "bubble");


var node = vis.selectAll("g.node")
    .data(bubble.nodes(classes(json))
    .filter(function(d) { return !d.children; }))
    .enter().append("g")
    .attr("class", "node")
    .attr("transform", function(d) { return "translate(" + d.x + "," + d.y + ")"; })
    color = d3.scale.category20();

  node.append("title")
      .text(function(d) { return d.className + ": " + format(d.value); });

  node.append("circle")
      .attr("r", function(d) { return d.r; })
      .style("fill", function(d) {return color(d.topicName);});

var text = node.append("text")
    .attr("text-anchor", "middle")
    .attr("dy", ".3em")
    .text(function(d) { return d.className.substring(0, d.r / 3)});

  text.append("tspan")
      .attr("dy", "1.2em")
      .attr("x", 0)
      .text(function(d) {return Math.ceil(d.value * 10000) /10000; });

// Returns a flattened hierarchy containing all leaf nodes under the root.
function classes(root) {
  var classes = [];

  function recurse(term, node) {
    if (node.children) node.children.forEach(function(child) { recurse(node.term, child); });
    else classes.push({topicName: node.topicId, className: node.term, value: node.probability});
  }

  recurse(null, root);
  return {children: classes};
}
</script>
""")

// Load the model
import org.apache.spark.mllib.clustering.LocalLDAModel
val model = LocalLDAModel.load(sc, lda_model_store)

import org.apache.spark.mllib.clustering.LocalLDAModel
model: org.apache.spark.mllib.clustering.LocalLDAModel = org.apache.spark.mllib.clustering.LocalLDAModel@6067f4d5

// Load the countvector
// val lda_countVector_mllib_loaded = sc.textFile("lda_countvector_sv_20201208") // This doesn't work, need to figure out how to parse the loaded rdd
// print(lda_countVector_mllib_loaded)

val numsample = 20
val sample = lda_countVector_mllib.takeSample(false, numsample)

val sampleRdd = sc.makeRDD(sample)
val topicDistributions = model.topicDistributions(sampleRdd)
val MLTopics = topicDistributions.map({case(id, vs) => (id, vs.argmax)})
val ids = MLTopics.map({case(id, vs)=>id}).collect()
val selection = filtered.filter({case(id, title, contents) => ids.contains(id)}).map({case(id, title, contents) => (id, title)}).collect()
val allTopicProbs = model.topicDistributions(lda_countVector_mllib)

sampleRdd: org.apache.spark.rdd.RDD[(Long, org.apache.spark.mllib.linalg.Vector)] = ParallelCollectionRDD[2329] at makeRDD at command-4443336225072182:1
topicDistributions: org.apache.spark.rdd.RDD[(Long, org.apache.spark.mllib.linalg.Vector)] = MapPartitionsRDD[2330] at map at LDAModel.scala:373
MLTopics: org.apache.spark.rdd.RDD[(Long, Int)] = MapPartitionsRDD[2331] at map at command-4443336225072182:3
ids: Array[Long] = Array(2779114, 226060, 1613204, 4970370, 700470, 636341, 4987605, 766457, 4965484, 82392, 7982574, 7973374, 367933, 1204225, 1225253, 503655, 821159, 8299631, 4684122, 1695311)
selection: Array[(Long, String)] = Array((82392,Senmedeltiden), (226060,Río Verde, San Luis Potosí), (367933,Kontrollsumma), (503655,Adam Kårsnäs), (636341,Live at McCabe's Guitar Shop), (700470,R2-0), (766457,Pakistanska muslimska förbundet - L), (821159,Fisher Athletic FC), (1204225,Financial Institutions Reform, Recovery and Enforcement Act of 1989), (1225253,Gardar), (1613204,Bessan), (1695311,Borneol), (2779114,Sara Branham Matthews), (4684122,Cecil Blachford), (4965484,Riddarsynerätt), (4970370,Kenyaskratthärfågel), (4987605,Rihannas diskografi), (7973374,Ahtisaariplanen), (7982574,James S. Brown), (8299631,Mount Roper))
allTopicProbs: org.apache.spark.rdd.RDD[(Long, org.apache.spark.mllib.linalg.Vector)] = MapPartitionsRDD[2335] at map at LDAModel.scala:373

for ( (article_id, topic_id) <- MLTopics.collect() ) {
  val title = selection.filter({case(id, title) => id == article_id})(0)._2
  val topicProbs = allTopicProbs.map({case(id, vs) => (id, vs(topic_id))})
  val top5articles = topicProbs.takeOrdered(5)(Ordering[Double].reverse.on(_._2))
  val top5ids = top5articles.map(_._1)
  val top5titles = filtered.filter({case(id, title, contents) => top5ids.contains(id) }).map({case(id, title, contents) => (title)}).collect
  println(s"Top 5 articles on the same topic as ${title}:")

  for (t <- top5titles) println(s" * ${t}")
  println("")
}