refactor BigDataSpark/w1/wikipedia

Author: spamegg1

courses/BigDataSpark/w1/wikipedia/build.sbt

@@ -1,29 +1,34 @@
 course := "bigdata"
 assignment := "wikipedia"
 
-scalaVersion := "3.1.0"
+// HAS TO BE: sbt 1.7.0, spark 3.2.0, Java 11. Does not work otherwise.
+scalaVersion := "3.4.2"
 scalacOptions ++= Seq("-language:implicitConversions", "-deprecation")
 libraryDependencies ++= Seq(
-  "org.scalameta" %% "munit" % "0.7.26" % Test,
-  excludes(("org.apache.spark" %% "spark-core" % "3.2.0").cross(CrossVersion.for3Use2_13)),
-  excludes(("org.apache.spark" %% "spark-sql" % "3.2.0").cross(CrossVersion.for3Use2_13))
+  "org.scalameta" %% "munit" % "1.0.0" % Test,
+  excludes(
+    ("org.apache.spark" %% "spark-core" % "3.2.0").cross(CrossVersion.for3Use2_13)
+  ),
+  excludes(
+    ("org.apache.spark" %% "spark-sql" % "3.2.0").cross(CrossVersion.for3Use2_13)
+  )
 )
 
-//netty-all replaces all these excludes
+// netty-all replaces all these excludes
 def excludes(m: ModuleID): ModuleID =
-  m.exclude("io.netty", "netty-common").
-    exclude("io.netty", "netty-handler").
-    exclude("io.netty", "netty-transport").
-    exclude("io.netty", "netty-buffer").
-    exclude("io.netty", "netty-codec").
-    exclude("io.netty", "netty-resolver").
-    exclude("io.netty", "netty-transport-native-epoll").
-    exclude("io.netty", "netty-transport-native-unix-common").
-    exclude("javax.xml.bind", "jaxb-api").
-    exclude("jakarta.xml.bind", "jaxb-api").
-    exclude("javax.activation", "activation").
-    exclude("jakarta.annotation", "jakarta.annotation-api").
-    exclude("javax.annotation", "javax.annotation-api")
+  m.exclude("io.netty", "netty-common")
+    .exclude("io.netty", "netty-handler")
+    .exclude("io.netty", "netty-transport")
+    .exclude("io.netty", "netty-buffer")
+    .exclude("io.netty", "netty-codec")
+    .exclude("io.netty", "netty-resolver")
+    .exclude("io.netty", "netty-transport-native-epoll")
+    .exclude("io.netty", "netty-transport-native-unix-common")
+    .exclude("javax.xml.bind", "jaxb-api")
+    .exclude("jakarta.xml.bind", "jaxb-api")
+    .exclude("javax.activation", "activation")
+    .exclude("jakarta.annotation", "jakarta.annotation-api")
+    .exclude("javax.annotation", "javax.annotation-api")
 
 // Without forking, ctrl-c doesn't actually fully stop Spark
 run / fork := true

courses/BigDataSpark/w1/wikipedia/project/build.properties

@@ -1 +1 @@
-sbt.version=1.6.1
+sbt.version=1.7.0

courses/BigDataSpark/w1/wikipedia/src/main/scala/wikipedia/WikipediaData.scala

@@ -4,11 +4,12 @@ import scala.io.Source
 import scala.io.Codec
 
 object WikipediaData:
-
   private[wikipedia] def lines: List[String] =
     Option(getClass.getResourceAsStream("/wikipedia/wikipedia-grading.dat")) match
-      case None => sys.error(
-        "Please download the dataset as explained in the assignment instructions")
+      case None =>
+        sys.error(
+          "Please download the dataset as explained in the assignment instructions"
+        )
       case Some(resource) =>
         Source
           .fromInputStream(resource)(Codec.UTF8)
@@ -19,5 +20,5 @@ object WikipediaData:
     val subs = "</title><text>"
     val i = line.indexOf(subs)
     val title = line.substring(14, i)
-    val text  = line.substring(i + subs.length, line.length-16)
+    val text = line.substring(i + subs.length, line.length - 16)
     WikipediaArticle(title, text)

courses/BigDataSpark/w1/wikipedia/src/main/scala/wikipedia/WikipediaRanking.scala

@@ -4,7 +4,6 @@ import org.apache.spark.SparkConf
 import org.apache.spark.SparkContext
 import org.apache.spark.SparkContext.*
 import org.apache.log4j.{Logger, Level}
-
 import org.apache.spark.rdd.RDD
 import scala.util.Properties.isWin
 
@@ -17,10 +16,7 @@ case class WikipediaArticle(title: String, text: String):
   def mentionsLanguage(lang: String): Boolean = text.split(' ').contains(lang)
 
 object WikipediaRanking extends WikipediaRankingInterface:
-  // Reduce Spark logging verbosity
-  Logger
-    .getLogger("org.apache.spark")
-    .setLevel(Level.ERROR)
+  Logger.getLogger("org.apache.spark").setLevel(Level.ERROR) // Reduce Spark log verbosity
 
   if isWin then
     System.setProperty(
@@ -46,21 +42,17 @@ object WikipediaRanking extends WikipediaRankingInterface:
     "Groovy"
   )
 
-  val conf: SparkConf = // TODO
-    new SparkConf()
-      .setMaster("local[4]")
-      .setAppName("MyApp")
-
+  val conf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("MyApp") // TODO
   val sc: SparkContext = new SparkContext(conf) // TODO
 
   // Hint: use a combination of `sc.parallelize`,
   // `WikipediaData.lines` and `WikipediaData.parse`
   val wikiRdd: RDD[WikipediaArticle] = // TODO
     sc.parallelize(WikipediaData.lines map WikipediaData.parse, 4)
 
-  /** Returns the number of articles on which the language `lang` occurs. Hint1:
-    * consider using method `aggregate` on RDD[T]. Hint2: consider using method
-    * `mentionsLanguage` on `WikipediaArticle`
+  /** Returns the number of articles on which the language `lang` occurs. Hint1: consider
+    * using method `aggregate` on RDD[T]. Hint2: consider using method `mentionsLanguage`
+    * on `WikipediaArticle`
     */
   def occurrencesOfLang(lang: String, rdd: RDD[WikipediaArticle]): Int = // TODO
     val fun = (count: Int, article: WikipediaArticle) =>
@@ -79,22 +71,18 @@ object WikipediaRanking extends WikipediaRankingInterface:
       langs: List[String],
       rdd: RDD[WikipediaArticle]
   ): List[(String, Int)] = // TODO
-    val langRanks =
-      for lang <- langs
-      yield (lang, occurrencesOfLang(lang, rdd))
-
+    val langRanks = for lang <- langs yield (lang, occurrencesOfLang(lang, rdd))
     langRanks.sortBy(-_._2)
 
-  /* Compute an inverted index of the set of articles, mapping each language
-   * to the Wikipedia pages in which it occurs.
+  /* Compute an inverted index of the set of articles, mapping each language to the
+   * Wikipedia pages in which it occurs.
    */
   def makeIndex(
       langs: List[String],
       rdd: RDD[WikipediaArticle]
   ): RDD[(String, Iterable[WikipediaArticle])] = // TODO
     val fun = (article: WikipediaArticle) =>
-      for lang <- langs filter article.mentionsLanguage
-      yield (lang, article)
+      for lang <- langs filter article.mentionsLanguage yield (lang, article)
 
     rdd // flatMap, for/yield, filter, groupByKey are LAZY
       .flatMap(fun) // RDD[(String, WikipediaArticle)]
@@ -129,8 +117,7 @@ object WikipediaRanking extends WikipediaRankingInterface:
       rdd: RDD[WikipediaArticle]
   ): List[(String, Int)] = // TODO
     val fun = (article: WikipediaArticle) =>
-      for lang <- langs filter article.mentionsLanguage
-      yield (lang, 1)
+      for lang <- langs filter article.mentionsLanguage yield (lang, 1)
 
     rdd
       .flatMap(fun)
@@ -139,30 +126,24 @@ object WikipediaRanking extends WikipediaRankingInterface:
       .collect // LAZY up to here!
       .toList
 
-  def main(args: Array[String]): Unit =
-
-    /* Languages ranked according to (1) */
-    val langsRanked: List[(String, Int)] =
+  def main: Unit =
+    val langsRanked: List[(String, Int)] = /* Languages ranked according to (1) */
       timed("Part 1: naive ranking", rankLangs(langs, wikiRdd))
 
-    /* An inverted index mapping languages to wikipedia pages
-     * on which they appear */
-    def index: RDD[(String, Iterable[WikipediaArticle])] =
-      makeIndex(langs, wikiRdd)
+    /* An inverted index mapping languages to wikipedia pages on which they appear */
+    def index: RDD[(String, Iterable[WikipediaArticle])] = makeIndex(langs, wikiRdd)
 
     /* Languages ranked according to (2), using the inverted index */
     val langsRanked2: List[(String, Int)] =
       timed("Part 2: ranking using inverted index", rankLangsUsingIndex(index))
 
-    /* Languages ranked according to (3) */
-    val langsRanked3: List[(String, Int)] =
+    val langsRanked3: List[(String, Int)] = /* Languages ranked according to (3) */
       timed(
         "Part 3: ranking using reduceByKey",
         rankLangsReduceByKey(langs, wikiRdd)
       )
 
-    /* Output the speed of each ranking */
-    println(timing)
+    println(timing) /* Output the speed of each ranking */
     sc.stop()
 
   val timing = new StringBuffer

courses/BigDataSpark/w1/wikipedia/src/main/scala/wikipedia/WikipediaRankingInterface.scala

@@ -4,25 +4,23 @@ import org.apache.spark.SparkContext
 import org.apache.spark.SparkContext.*
 import org.apache.spark.rdd.RDD
 
-/**
- * The interface used by the grading infrastructure. Do not change signatures
- * or your submission will fail with a NoSuchMethodError.
- */
+/** The interface used by the grading infrastructure. Do not change signatures or your
+  * submission will fail with a NoSuchMethodError.
+  */
 trait WikipediaRankingInterface:
-  def makeIndex(langs: List[String],
-                  rdd: RDD[WikipediaArticle])
-                     : RDD[(String, Iterable[WikipediaArticle])]
-  def occurrencesOfLang(lang: String,
-                         rdd: RDD[WikipediaArticle])
-                            : Int
-  def rankLangs(langs: List[String],
-                  rdd: RDD[WikipediaArticle])
-                     : List[(String, Int)]
-  def rankLangsReduceByKey(langs: List[String],
-                             rdd: RDD[WikipediaArticle])
-                                : List[(String, Int)]
-  def rankLangsUsingIndex(index: RDD[(String, Iterable[WikipediaArticle])])
-                               : List[(String, Int)]
+  def makeIndex(
+      langs: List[String],
+      rdd: RDD[WikipediaArticle]
+  ): RDD[(String, Iterable[WikipediaArticle])]
+  def occurrencesOfLang(lang: String, rdd: RDD[WikipediaArticle]): Int
+  def rankLangs(langs: List[String], rdd: RDD[WikipediaArticle]): List[(String, Int)]
+  def rankLangsReduceByKey(
+      langs: List[String],
+      rdd: RDD[WikipediaArticle]
+  ): List[(String, Int)]
+  def rankLangsUsingIndex(
+      index: RDD[(String, Iterable[WikipediaArticle])]
+  ): List[(String, Int)]
   def langs: List[String]
   def sc: SparkContext
   def wikiRdd: RDD[WikipediaArticle]