<jupyter_start><jupyter_text>Lab 2 - Clustering with GraphXIn this lab we use GraphX to cluster music songs according to the tags attached to each song. The datasets that we are working is available under the folder `data`, and also can be downloaded from [here](http://www.cs.cornell.edu/~shuochen/lme/data_page.html). The datasets consist of the following files:* `song_hash.txt`: this file maps a song ID to its title and artist.  + Each line corresponds to one song, and has the format called `Integer_ID\tTitle\tArtist\n`.* `tag_hash.txt`: this one maps a tag ID to its name.  + Each line corresponds to one song, and has the format called `Integer_ID, Name\n`.* `tags.txt`: this file includes the social tags by using the integer ID to represent songs.  + The tag data file has the same number of lines as the total number of songs. Each line consists of the IDs of the tags for a song, represented by integers, and separated by space. If a song does not have a tag, its line is just a . Note that for the tag file, there is no space at the end of each line. Load the data into a Spark graph propertyLet's start by loading the data into a Spark graph property. To do this, wee define a case class `Song`, with three attributes: `title`, `artist`, and a set of `tags`.<jupyter_code>import scala.io.Source
import org.apache.spark.rdd.RDD
import org.apache.spark.graphx._
import org.apache.spark.mllib.clustering.PowerIterationClustering

case class Song(title: String, artist: String, tags: Set[String])<jupyter_output><empty_output><jupyter_text>Now, import the songs from the dataset file `song_hash.txt` into `RDD[(VertexId, Song)]`, called `songs`, and initialize each song with an empty set of tags. You should see the following lines, if you run `songs.take(5).foreach(println)`:```(0,Song(Gucci Time (w\/ Swizz Beatz),Gucci Mane,Set()))(1,Song(Aston Martin Music (w\/ Drake & Chrisette Michelle),Rick Ross,Set()))(2,Song(Get Back Up (w\/ <NAME>),T.I.,Set()))(3,Song(Hot Toddy (w\/ Jay-Z & Ester Dean),Usher,Set()))(4,Song(Whip My Hair,Willow,Set()))```<jupyter_code>val fileName = "data/song_hash.txt"

// var songs: RDD[(VertexId, Song)] = < FILL IN >
var songs: RDD[(VertexId, Song)] = sc
                                    .textFile(fileName)
                                    .map(row => row.split('\t'))
                                    .map(a => (a(0).toInt, Song(a(1).toString, a(2).toString, Set())))
songs.take(5).foreach(println)<jupyter_output>(0,Song(Gucci Time (w\/ Swizz Beatz),Gucci Mane,Set()))
(1,Song(Aston Martin Music (w\/ Drake & Chrisette Michelle),<NAME>,Set()))
(2,Song(Get Back Up (w\/ <NAME>),T.I.,Set()))
(3,Song(Hot Toddy (w\/ Jay-Z & Ester Dean),Usher,Set()))
(4,Song(Whip My Hair,Willow,Set()))<jupyter_text>Then, create a graph property called `graphFromSongs`, whose nodes are the songs, i.e., `Graph[Song, Int]`. It will not add any edges into the graph at first. If you print the vertices of the `graphFromSongs` by calling `graphFromSongs.vertices.take(5).foreach(println)`, you should see something as below:```(1084,Song(Tequila Sunrise,Fiji,Set()))(1410,Song(The Sweetest Taboo,Sade,Set()))(3066,Song(Bow Chicka Wow Wow,<NAME>ner,Set()))(1894,Song(Love Your Love The Most,Eric Church,Set()))(466,Song(Stupify,Disturbed,Set()))```<jupyter_code>// val graphFromSongs: Graph[Song, Int] = < FILL IN>
val graphFromSongs: Graph[Song, Int] = Graph(songs, sc.emptyRDD)

graphFromSongs.vertices.take(5).foreach(println)<jupyter_output>(1084,Song(Tequila Sunrise,Fiji,Set()))
(1410,Song(The Sweetest Taboo,Sade,Set()))
(3066,Song(Bow Chicka Wow Wow,<NAME>,Set()))
(1894,Song(Love Your Love The Most,<NAME>,Set()))
(466,Song(Stupify,Disturbed,Set()))<jupyter_text>Extract the features of nodesNow, let's join the tags from the dataset called `tags.txt` into the nodes. To do this, we first need to create `tagRDD`, which is in form of `RDD[(VertexId, Set[String])]`. We hen join it with `graphFromSong`. For now, we have only the mapping between the song ID and the set of tag IDs in our `tagRDD`. If we print the first five items of the `tagRDD` we should see the following lines:```(0,Set(115, 173))(1,Set(62, 88, 110, 90, 123, 155, 173, 14, 190, 214, 115, 27))(2,Set(115, 173))(3,Set(2, 72, 173))(4,Set(62, 88, 141, 107, 24, 155, 72, 6, 126, 173, 190, 115, 2, 52))```<jupyter_code>val tagFile = "data/tags.txt"
// val tagIter: Iterator[(VertexId, Set[String])] = Source.fromFile(tagFile).getLines.zipWithIndex.map { <FILL IN> }
val tagIter: Iterator[(VertexId, Set[String])] = Source
                                                    .fromFile(tagFile)
                                                    .getLines
                                                    .zipWithIndex
                                                    .map {line => (line._2, line._1.split(" ").toSet)}                        

val tagRDD = sc.parallelize(tagIter.toSeq)

tagRDD.take(5).foreach(println)<jupyter_output>(0,Set(115, 173))
(1,Set(62, 88, 110, 90, 123, 155, 173, 14, 190, 214, 115, 27))
(2,Set(115, 173))
(3,Set(2, 72, 173))
(4,Set(62, 88, 141, 107, 24, 155, 72, 6, 126, 173, 190, 115, 2, 52))<jupyter_text>What we want is to extract the tag names from `tag_hash.txt` given the tag ID. We can now call `joinVertices` on `graphFromSongs`, and pass the RDD of tags `tagRDD` with a function that extracts the tags. Note that in the dataset called `tags.txt`, a `` assigned next to the song ID means that no tag is associated with that song. In such a case, we simply return the initial song with an empty tag, otherwise, we add the set of tags into the song. Below, you can see an example output if you run `songsNtags.vertices.take(5).foreach(println)`:```(1084,Song(Tequila Sunrise,Fiji,Set()))(1410,Song(The Sweetest Taboo,Sade,Set(beautiful, 1980s, 80's, rock, sexy, favorite, female vocalist, smooth, female, pop, best, 80s, soul, easy listening, classic, chillout, urban, lounge, romantic, soft, singer-songwriter, lovely, chill, dance, love songs, mellow, love, jazz, female vocals, british, female vocalists, rnb, smooth jazz, favorites, ballad, relax, adult contemporary, great, melancholy, relaxing)))(3066,Song(Bow Chicka Wow Wow,<NAME>,Set(sexy, smooth, male vocalists, pop, <3, hip-hop, love song, wdzh-fm, awesome, easy listening, love at first listen, r&b, electronic, love songs, love, american, 2010s, nostalgic, rnb, favorites, wkqi-fm)))(1894,Song(Love Your Love The Most,Eric Church,Set(beautiful, makes me smile, fav, pop, modern country, country, great song, love songs, favourite songs, usa, love, favorites, male country, my favorite)))(466,Song(Stupify,Disturbed,Set(punk rock, rock, 90s, alternative metal, favorite, favourite, favorite songs, good, nu metal, 00s, angry, progressive rock, awesome, male vocalist, hardcore, sex, alternative, favs, heavy, 2000s, heavy metal, classic rock, hard rock, alternative rock, american, female vocalists, catchy, favorites, metal, my music, nu-metal, aitch)))```<jupyter_code>val songsNtags = graphFromSongs.joinVertices(tagRDD) {
  (id, s, ks) => ks.toList match {
    case List("#") => s
    case _ => {
      val taghashFile = "data/tag_hash.txt"
      // val tags: Map[Int, String] = Source.fromFile(taghashFile).getLines().<FILL IN>
      val tags: Map[Int, String] = Source
                                    .fromFile(taghashFile)
                                    .getLines
                                    .map(line => line.split(", "))
                                    .map(line => (line(0).toInt, line(1))).toMap
      val songTags = ks.map(_.toInt).flatMap(tags.get)
      Song(s.title, s.artist, songTags.toSet)
    }
  }
}

songsNtags.vertices.take(5).foreach(println)<jupyter_output>(1084,Song(Tequila Sunrise,Fiji,Set()))
(1410,Song(The Sweetest Taboo,Sade,Set(beautiful, 1980s, 80's, rock, sexy, favorite, female vocalist, smooth, female, pop, best, 80s, soul, easy listening, classic, chillout, urban, lounge, romantic, soft, singer-songwriter, lovely, chill, dance, love songs, mellow, love, jazz, female vocals, british, female vocalists, rnb, smooth jazz, favorites, ballad, relax, adult contemporary, great, melancholy, relaxing)))
(3066,Song(Bow Chicka Wow Wow,<NAME>,Set(sexy, smooth, male vocalists, pop, <3, hip-hop, love song, wdzh-fm, awesome, easy listening, love at first listen, r&b, electronic, love songs, love, american, 2010s, nostalgic, rnb, favorites, wkqi-fm)))
(1894,Song(Love Your Love The Most,Eric Church,Set(beautiful, makes me smile, fav, pop, modern country, country, great song, love songs, favourite songs, usa, love, favorites, male country, my favorite)))
(466,Song(Stupify,Disturbed,Set(punk rock, rock, 90s, alternative metal, favorite, favou[...]<jupyter_text>Define a similarity measure between two nodesNow, let's measure the similarity between two songs using the *Jaccard metric*, which is the ratio of the number of common tags between two songs, and their total number of tags. If none of the songs is tagged, we assume that their similarity score is zero. For this, we define the `similarity` method as below.<jupyter_code>def similarity(one: Song, other: Song): Double = {
  val numCommonTags = (one.tags intersect other.tags).size
  val numTotalTags = (one.tags union other.tags).size
  if (numTotalTags > 0)
    numCommonTags.toDouble / numTotalTags.toDouble
  else 
    0.0
}<jupyter_output><empty_output><jupyter_text>Creating an affinity matrixNow, we need to calculate the similarity between each pair of songs in our database. If there are 1,000 songs, we will have to compute, and store, one million similarity scores. What if we had 1,000,000 songs? Obviously, computing similarities between every pair will be inefficient. Instead, we can restrict this to the songs that have a relatively high similarity score. At the end of the day, we want to cluster songs that are similar. Therefore, we will filter the nodes with the following function.<jupyter_code>def quiteSimilar(one: Song, other: Song, threshold: Double): Boolean = {
  val commonTags = one.tags.intersect(other.tags)
  val combinedTags = one.tags.union(other.tags)
  commonTags.size > combinedTags.size * threshold
}<jupyter_output><empty_output><jupyter_text>We define a function to remove the duplicate songs in our graph data.<jupyter_code>def differentSong(one: Song, other: Song): Boolean = one.title != other.title || one.artist != other.artist<jupyter_output><empty_output><jupyter_text>With these two functions, we can now create `RDD[Edge[Double]]` that will contain a similarity measure between the nodes that are quite similar. A simple check `similarConnections.count` shows that we only need to store 1,506 similarity scores instead of 10 million.<jupyter_code>/* * * * * * * * * * *
 * HELPER FUNCTIONS  *
 * * * * * * * * * * */

def differentSong(one: Song, other: Song): Boolean = one.title != other.title || one.artist != other.artist 
    
def quiteSimilar(one: Song, other: Song, threshold: Double): Boolean = {
  val commonTags = one.tags.intersect(other.tags)
  val combinedTags = one.tags.union(other.tags)
  commonTags.size > combinedTags.size * threshold
}

def similarity(one: Song, other: Song): Double = {
  val numCommonTags = (one.tags intersect other.tags).size
  val numTotalTags = (one.tags union other.tags).size
  if (numTotalTags > 0)
    numCommonTags.toDouble / numTotalTags.toDouble
  else 
    0.0
}

/* * * * * * * * * * *
 * ORIGINAL NOTEBOOK *
 * * * * * * * * * * */

// First, get the songs with tags
songs = songsNtags.vertices

// Then, compute the similarity between each pair of songs with a similarity score larger than 0.7
val similarConnections: RDD[Edge[Double]] = {
  val ss = songs.cartesian(songs)
    
  // similarSongs are the songs with a similarity score larger than 0.7
  // Note that not compare a song with itself (use the differentSong method)
  // val similarSongs = ss.filter { <FILL IN> }
  val similarSongs = ss.filter {s => differentSong(s._1._2, s._2._2) && quiteSimilar(s._1._2, s._2._2, 0.7)}
    
  // Now compute the Jaccard metric for the similarSongs. The result should ba an Edge for each pair of songs
  // with the vertexIds at the two ends, and the Javvard values as the edge value.
  similarSongs.map { p => {
    // val jacIdx = <FILL IN>
    val jacIdx = similarity(p._1._2, p._2._2)
    Edge(p._1._1, p._2._1, jacIdx)}
  }
}<jupyter_output><empty_output><jupyter_text>Let's create our similarity graph `similarByTagsGraph`.<jupyter_code>val similarByTagsGraph = Graph(songs, similarConnections)<jupyter_output><empty_output><jupyter_text>Some of our songs have very few tags, so let's filter those out and keep the songs with more than five tags.<jupyter_code>// val similarHighTagsGraph = similarByTagsGraph.subgraph(<FILL IN>)
val similarHighTagsGraph = similarByTagsGraph.subgraph(vpred = (id, node) => node.tags.size > 5)<jupyter_output><empty_output><jupyter_text>Let's look closer into the graph.<jupyter_code>similarHighTagsGraph.triplets.take(6).foreach(t => println(t.srcAttr + " ~~~ " + t.dstAttr + " => " + t.attr))<jupyter_output>Song(Fancy (w\/ T.I. & Swizz Beatz),Drake,Set(rap, wdzh-fm, 2010, hip hop, wjlb-fm, whtd-fm, wkqi-fm)) ~~~ Song(Any Girl (w\/ Lloyd),Lloyd Banks,Set(rap, wdzh-fm, hip hop, wjlb-fm, whtd-fm, wkqi-fm)) => 0.8571428571428571
Song(Roll With It,Easton Corbin,Set(new country, modern country, country, great song, 2010, my favorite)) ~~~ Song(You Lie,The Band Perry,Set(new country, modern country, country, great song, female vocalists, my favorite)) => 0.7142857142857143
Song(Any Girl (w\/ Lloyd),Lloyd Banks,Set(rap, wdzh-fm, hip hop, wjlb-fm, whtd-fm, wkqi-fm)) ~~~ Song(Fancy (w\/ T.I. & <NAME>),Drake,Set(rap, wdzh-fm, 2010, hip hop, wjlb-fm, whtd-fm, wkqi-fm)) => 0.8571428571428571
Song(Any Girl (w\/ Lloyd),Lloyd Banks,Set(rap, wdzh-fm, hip hop, wjlb-fm, whtd-fm, wkqi-fm)) ~~~ Song(I'm Going In (w\/ Young Jeezy & Lil Wayne),Drake,Set(rap, hip-hop, wdzh-fm, wjlb-fm, whtd-fm, wkqi-fm)) => 0.7142857142857143
Song(Everything Falls,Fee,Set(rock, worship, christian, contemporary christian, go[...]
{'fraction_non_alphanumeric': 0.12860594003126333, 'fraction_numerical': 0.03112121642745488, 'mean_word_length': 4.224573125463994, 'pattern_counts': {'":': 0, '<': 62, '<?xml version=': 0, '>': 82, 'https://': 0, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '12', 'dir': 'jupyter-structured-clean-dedup', 'id': '403639', 'license': 'Apache-2.0', 'n_tokens_mistral': 4956, 'n_tokens_neox': 4456, 'n_words': 1536, 'path': 'lab_2/graphx_songs.ipynb', 'repo_name': 'adrianomundo/data-intensive-computing', 'size': '14114'}